Observable.contains;

doc/README.md

@@ -28,6 +28,7 @@ RxLua
   - [combineLatest](#combinelatestobservables-combinator)
   - [compact](#compact)
   - [concat](#concatsources)
+  - [contains](#containsvalue)
   - [count](#countpredicate)
   - [distinct](#distinct)
   - [distinctUntilChanged](#distinctuntilchangedcomparator)
@@ -312,6 +313,16 @@ Returns a new Observable that produces the values produced by all the specified
 
 ---
 
+#### `:contains(value)`
+
+Returns a new Observable that produces a single boolean value representing whether or not the specified value was produced by the original.
+
+| Name | Type | Default | Description |
+|------|------|---------|-------------|
+| `value` | * |  | The value to search for.  == is used for equality testing. |
+
+---
+
 #### `:count(predicate)`
 
 Returns an Observable that produces a single value representing the number of values produced by the source value that satisfy an optional predicate.

rx.lua

@@ -487,6 +487,46 @@ function Observable:concat(other, ...)
   end)
 end
 
+--- Returns a new Observable that produces a single boolean value representing whether or not the
+-- specified value was produced by the original.
+-- @arg {*} value - The value to search for.  == is used for equality testing.
+-- @returns {Observable}
+function Observable:contains(value)
+  return Observable.create(function(observer)
+    local subscription
+
+    local function onNext(...)
+      local args = util.pack(...)
+
+      if #args == 0 and value == nil then
+        observer:onNext(true)
+        if subscription then subscription:unsubscribe() end
+        return observer:onCompleted()
+      end
+
+      for i = 1, #args do
+        if args[i] == value then
+          observer:onNext(true)
+        if subscription then subscription:unsubscribe() end
+          return observer:onCompleted()
+        end
+      end
+    end
+
+    local function onError(e)
+      return observer:onError(e)
+    end
+
+    local function onCompleted()
+      observer:onNext(false)
+      return observer:onCompleted()
+    end
+
+    subscription = self:subscribe(onNext, onError, onCompleted)
+    return subscription
+  end)
+end
+
 --- Returns an Observable that produces a single value representing the number of values produced
 -- by the source value that satisfy an optional predicate.
 -- @arg {function=} predicate - The predicate used to match values.

src/observable.lua

@@ -402,6 +402,46 @@ function Observable:concat(other, ...)
   end)
 end
 
+--- Returns a new Observable that produces a single boolean value representing whether or not the
+-- specified value was produced by the original.
+-- @arg {*} value - The value to search for.  == is used for equality testing.
+-- @returns {Observable}
+function Observable:contains(value)
+  return Observable.create(function(observer)
+    local subscription
+
+    local function onNext(...)
+      local args = util.pack(...)
+
+      if #args == 0 and value == nil then
+        observer:onNext(true)
+        if subscription then subscription:unsubscribe() end
+        return observer:onCompleted()
+      end
+
+      for i = 1, #args do
+        if args[i] == value then
+          observer:onNext(true)
+        if subscription then subscription:unsubscribe() end
+          return observer:onCompleted()
+        end
+      end
+    end
+
+    local function onError(e)
+      return observer:onError(e)
+    end
+
+    local function onCompleted()
+      observer:onNext(false)
+      return observer:onCompleted()
+    end
+
+    subscription = self:subscribe(onNext, onError, onCompleted)
+    return subscription
+  end)
+end
+
 --- Returns an Observable that produces a single value representing the number of values produced
 -- by the source value that satisfy an optional predicate.
 -- @arg {function=} predicate - The predicate used to match values.

tests/contains.lua

@@ -0,0 +1,29 @@
+describe('contains', function()
+  it('errors when its parent errors', function()
+    expect(Rx.Observable.throw():contains(1).subscribe).to.fail()
+  end)
+
+  it('returns false if the source Observable produces no values', function()
+    expect(Rx.Observable.empty():contains(3)).to.produce(false)
+  end)
+
+  it('returns true if the value is nil and the Observable produces an empty value', function()
+    local observable = Rx.Observable.create(function(observer)
+      observer:onNext(nil)
+      observer:onCompleted()
+    end)
+
+    expect(observable:contains(nil)).to.produce(true)
+  end)
+
+  it('returns true if the source Observable produces the specified value', function()
+    local observable = Rx.Observable.fromRange(5)
+    expect(observable:contains(3)).to.produce(true)
+  end)
+
+  it('supports multiple values', function()
+    local observable = Rx.Observable.fromRange(6):wrap(3)
+    expect(observable).to.produce({{1, 2, 3}, {4, 5, 6}})
+    expect(observable:contains(5)).to.produce(true)
+  end)
+end)

tests/observable.lua

@@ -185,6 +185,7 @@ describe('Observable', function()
   dofile('tests/combineLatest.lua')
   dofile('tests/compact.lua')
   dofile('tests/concat.lua')
+  dofile('tests/contains.lua')
   dofile('tests/distinct.lua')
   dofile('tests/distinctUntilChanged.lua')
   dofile('tests/filter.lua')