Add Observable.with;

doc/README.md

@@ -41,6 +41,7 @@ RxLua
   - [unpack](#unpack)
   - [unwrap](#unwrap)
   - [window](#windowsize)
+  - [with](#withsources)
   - [wrap](#wrapsize)
 - [Scheduler](#scheduler)
 - [ImmediateScheduler](#immediatescheduler)
@@ -546,6 +547,20 @@ Returns:
 
 ---
 
+#### `:with(sources)`
+
+Returns an Observable that produces values from the original along with the most recently produced value from all other specified Observables. Note that only the first argument from each source Observable is used.
+
+Arguments:
+
+- `sources` (`Observable...`) - The Observables to include the most recent values from.
+
+Returns:
+
+- `Observable`
+
+---
+
 #### `:wrap(size)`
 
 Returns an Observable that buffers values from the original and produces them as multiple values.

rx.lua

@@ -778,6 +778,43 @@ function Observable:window(size)
   end)
 end
 
+--- Returns an Observable that produces values from the original along with the most recently
+-- produced value from all other specified Observables. Note that only the first argument from each
+-- source Observable is used.
+-- @arg {Observable...} sources - The Observables to include the most recent values from.
+-- @returns {Observable}
+function Observable:with(...)
+  local sources = {...}
+
+  return Observable.create(function(observer)
+    local latest = {}
+
+    local function setLatest(i)
+      return function(value)
+        latest[i] = value
+      end
+    end
+
+    local function onNext(value)
+      return observer:onNext(value, unpack(latest))
+    end
+
+    local function onError(e)
+      return observer:onError(e)
+    end
+
+    local function onComplete()
+      return observer:onComplete()
+    end
+
+    for i = 1, #sources do
+      sources[i]:subscribe(setLatest(i), noop, noop)
+    end
+
+    return self:subscribe(onNext, onError, onComplete)
+  end)
+end
+
 --- Returns an Observable that buffers values from the original and produces them as multiple
 -- values.
 -- @arg {number} size - The size of the buffer.