System.Reactive

The Reactive system is a Lua implementation for the the observer pattern to support sequences of data or events and adds operators that allow you to compose sequences together declaratively.

For the current version, it can only be used to in single os-thread platform, or only be used in one os-thread, we can declare them in a lock thread at the beginning and keep processing the operations in one os-thread.

For more advance and simple usages, you should check the watch & reactive part.

A simple example for start:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	-- The dump use default logger for output
	System.Logger.Default:AddHandler(print)

	-- From 1 to 10, skip the first 3 elements, get the sum
	-- [Info]Dump-->49
	-- [Info]Dump completed
	Observable.Range(1, 10):Skip(3):Sum():Dump()
end)

Table of Contents

System.IObservable

The System.IObservable defines the provider for push-based notification, it declare one abstract method with two overloads:

Subscribe

Notifies the provider that an observer is ready to receive notifications.

• Usage : subscription, observer = observable:Subscribe(observer[, subscription])

  • Arguments:

    • observer - the observer to be subscribed.
    • subscription - the object used to track the subscription, when the object is disposed, the subscription must be cancelled.

  • Returns:

    • subscription - the object used to track the subscription, it'll be created if not provided, otherwise the provided subscription object will be returned.
    • observer - the observer

• Usage : subscription, observer = observable:Subscribe([onNext], [onError], [onCompleted][, subscription])

  • Arguments:

    • onNext - the function used to handle the pushed data.
    • onError - the function used to handle the error data.
    • onCompleted - the function used to handle the complete data.
    • subscription - the object used to track the subscription, when the object is disposed, the subscription must be cancelled.

  • Returns:

    • subscription - the object used to track the subscription, it'll be created if not provided, otherwise the provided subscription object will be returned.
    • observer - the observer that created based on those arguments.

System.IObserver

The System.IObserver provides the mechanism for receiving push-based notifications, it declare three abstract methods :

Abstract Method	Usage
OnNext(self, ...)	Provides the observer with new data
OnError(self, exception)	Notifies the observer that the provider has experienced an error condition
OnCompleted(self)	Notifies the observer that the provider has finished sending push-based notifications

It also have well-defined property Subscription defined like

-- The subscription will be used by the observer
property "Subscription"         {
    type                        = Subscription,
    field                       = "__subscription",
    default                     = function(self) return Subscription() end,
    handler                     = function(self, new, old)
        if new and not new.IsUnsubscribed then
            new.OnUnsubscribe   = new.OnUnsubscribe + function() return rawset(self, "__subscription", nil) end
        end
        return old and not old.IsUnsubscribed and old:Dispose()
    end
}

So we can access the property to get a new subscription, and change its value will dispose the old one, also if we dipose the subcription with other codes, the property will clean the value, that'll make sure we get a new subscription if we want re-use the observer.

• Usage:

  local observable = IObserver{
  	OnNext = function(self, ...) print("Got", ...) end,
  	OnError = function(self, ex) print("Some thing error", tostring(ex)) end,
  	OnCompleted = function(self) print("The subscription is finished") end,
  }
  

The System.IObserver can be used as anonymous class to create objects, but normally we don't need to do that.

System.Subscription

The System.Subscription is a final class used to track the subscriptions, we can dispose it to cancel the subscription.

It provides one property IsUnsubscribed to check if the subscription is disposed, and an event OnUnsubscribe fired when it's disposing.

If we create an observable to check an observer's subscription, we must use the IsUnsubscribed property to check if the observer has cancelled the subscription, please don't use the event, it may cause the memeory leak that the observable may have a reference in the event handler which make it un-collectable.

class "System.Subscription"    (function(_ENV)
	-- fired when the subscription is disposed
	event "OnUnsubscribe"

	--- Whether is unsubscribed
	property "IsUnsubscribed"       {}
end)

• Usage

  require "PLoop" (function(_ENV)
  	-- Add default handler for Dump method
  	Logger.Default:AddHandler(print)
  
  	-- root observable used to generate data sequence
  	local root = Subject()
  
  	-- the proxy used to check the subscription
  	-- for simple we don't cancel the subscription for the root
  	-- don't worry about the Observable, it'll be explained later
  	local proxy = Observable(function(observer, subscription)
  		-- get data from root then send to the observer
  		root:Subscribe(function(...)
  			-- check the subscription
  			if not subscription.IsUnsubscribed then
  				observer:OnNext(...)
  			end
  		end)
  	end)
  
  	-- record the subscription, Dump is a simple to show the result
  	local subscription = proxy:Dump()
  	root:OnNext("will be print")      -- [Info]Dump-->will be print
  	root:OnNext("also will be print") -- [Info]Dump-->also will be print
  
  	-- dispose the subscription
  	subscription:Dispose()
  	root:OnNext("won't be print")
  end)
  

  For simple, we can create a child subscription, when the parent subscription is cancelled, the child will all be cancelled.

  	require "PLoop" (function(_ENV)
  		-- Add default handler for Dump method
  		Logger.Default:AddHandler(print)
  
  		-- root observable used to generate data sequence
  		local root = Subject()
  
  		-- the proxy used to check the subscription
  		-- for simple we don't cancel the subscription for the root
  		-- don't worry about the Observable, it'll be explained later
  		local proxy = Observable(function(observer, subscription)
  			-- get data from root then send to the observer
  			root:Subscribe(
  				function(...)
  					observer:OnNext(...)
  				end, nil, nil,
  
  				-- create the child subscription, will be auto disposed
  				Subscription(subscription)
  			)
  		end)
  
  		-- record the subscription, Dump is a simple to show the result
  		local subscription = proxy:Dump()
  		root:OnNext("will be print")      -- [Info]Dump-->will be print
  		root:OnNext("also will be print") -- [Info]Dump-->also will be print
  
  		-- dispose the subscription
  		subscription:Dispose()
  		root:OnNext("won't be print")
  	end)
  

  If we don't need handle the subscription in the proxy, we also can use the parent subscription directly

  	require "PLoop" (function(_ENV)
  		-- Add default handler for Dump method
  		Logger.Default:AddHandler(print)
  
  		-- root observable used to generate data sequence
  		local root = Subject()
  
  		-- the proxy used to check the subscription
  		-- for simple we don't cancel the subscription for the root
  		-- don't worry about the Observable, it'll be explained later
  		local proxy = Observable(function(observer, subscription)
  			-- get data from root then send to the observer
  			root:Subscribe(
  				function(...)
  					observer:OnNext(...)
  				end, nil, nil,
  				subscription
  			)
  		end)
  
  		-- record the subscription, Dump is a simple to show the result
  		local subscription = proxy:Dump()
  		root:OnNext("will be print")      -- [Info]Dump-->will be print
  		root:OnNext("also will be print") -- [Info]Dump-->also will be print
  
  		-- dispose the subscription
  		subscription:Dispose()
  		root:OnNext("won't be print")
  	end)
  

System.IConnectableObservable

The interface System.IConnectableObservable used to provide the connect mechanism for observable queues, so the observable won't send data Immediatly until it's connected. it declare one abstract method :

Abstract Method	Usage
Connect(self)	Connect the underlying observable queue

System.Reactive.Observable

There is no need to use those interface directly, they are well implemented in the System.Reactive with plenty features.

The Observable class is used to generate the observable objects, they are used to generate data sequences and send them to the observers.

So here is how we create a new observable objects:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- Create an observable
	local observable = Observable(function(observer, subscription)
		observer:OnNext(1, 2)
		observer:OnNext(3, 4)
		observer:OnCompleted()
	end)

	-- Got	1	2
	-- Got	3	4
	-- Completed
	observable:Subscribe(function(...)
		print("Got", ...)
	end, function(ex)
		print("There is an exception", ex)
	end, function()
		print("Completed")
	end)

	-- [Info]Dump-->1, 2
	-- [Info]Dump-->3, 4
	-- [Info]Dump completed
	observable:Dump()
end)

So we can create an observable by pass a function to the Observable class, the function will receive the observer and the subscription so we can send the data sequence to it.

In the function, we could use the OnNext method to send the data(multi data supported), the OnCompleted to notify the observer that the data sequence is finished, so the observer will be unsubscribed. If there is some exception, we should use the OnError(ex) and pass an exception object in it, and the observer also will be unsubscribed.

The Observable also provided several static method to simple the creation of the observables.

Observable.Create

Observable.Create(subscribe) works the same like Observable(subscribe).

Observable.Generate

Creates a new observable with initstate, condition checker, iterate and a result selector.

• Declaration:

__Arguments__{ Any, Callable, Callable, Callable/"...=>..." }
function Generate(init, condition, iterate, resultselector)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->1
	-- [Info]Dump-->2
	-- [Info]Dump-->3
	-- [Info]Dump-->4
	-- [Info]Dump completed
	Observable.Generate(1, "x=>x<5", "x=>x+1"):Dump()
end)

Observable.Just (Observable.Return)

Returns an Observable that just provide one value

• Declaration:

__Arguments__{ Any }
function Just(value)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->1
	-- [Info]Dump completed
	Observable.Just(1):Dump()
end)

Observable.Empty

Returns and Observable that immediately completes without producing a value

• Declaration:

function Empty()
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump completed
	Observable.Empty():Dump()
end)

Observable.Never

Returns an Observable that never produces values and never completes.

Observable.Throw

Returns an Observable that immediately produces an error.

• Declaration:

__Arguments__{ Exception + String }
function Throw(ex)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump failed-->Something error
	Observable.Throw("Something error"):Dump()
end)

Observable.Defer

Creates the Observable only when the observer subscribes.

• Declaration:

__Arguments__{ Callable }
function Defer(ex)
end

• Usage:

require "PLoop" (function(_ENV)
    import "System.Reactive"

    System.Logger.Default:AddHandler(print)

    local range = Observable.Range(1, 3)
    function calcFib()
        local f = 0
        local n = 1
        return Observable(function(observer, subscription)
            range:Subscribe(function(v)
                f, n = n, f + n
                observer:OnNext(f)
            end)
        end)
    end
    local map = calcFib()

    -- [Info]Dump-->1
    -- [Info]Dump-->1
    -- [Info]Dump-->2
    map:Dump()

    -- it won't re-init
    -- [Info]Dump-->3
    -- [Info]Dump-->5
    -- [Info]Dump-->8
    map:Dump()

    -- Use Defer
    local map = Observable.Defer(calcFib)

    -- [Info]Dump-->1
    -- [Info]Dump-->1
    -- [Info]Dump-->2
    map:Dump()

    -- [Info]Dump-->1
    -- [Info]Dump-->1
    -- [Info]Dump-->2
    map:Dump()
end)

Observable.From

Converts collection objects into Observables, or Converts event delegate objects into Observables, or Converts tables into Observables with iterator.

• Declaration:

__Arguments__{ Iterable }
function From(iter)
end

__Arguments__{ Delegate }
function From(deletegate)
end

__Arguments__{ Table, Callable/pairs }
function From(table, iter)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->1, 1
	-- [Info]Dump-->2, 2
	-- [Info]Dump-->3, 3
	-- [Info]Dump-->4, 4
	-- [Info]Dump-->5, 5
	-- [Info]Dump completed
	Observable.From(List(5)):Dump()

	class "Person" (function(_ENV)
		event "OnNameChanged"

		property "Name" { type = String, event = OnNameChanged }
	end)

	p = Person()
	Observable.From(p.OnNameChanged):Dump()

	-- [Info]Dump-->table: 011B8928, Ann, nil, Name
	p.Name = "Ann"

	-- [Info]Dump-->1, a
	-- [Info]Dump-->2, b
	-- [Info]Dump completed
	Observable.From{ a = 1, b = 2 }:Dump()
end)

Observable.Range

Creates an Observable that emits a particular range of sequential integers

• Declaration:

__Arguments__{ Number, Number, Number/nil }
function Range(start, stop, step)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->7
	-- [Info]Dump-->8
	-- [Info]Dump-->9
	-- [Info]Dump-->10
	-- [Info]Dump completed
	Observable.Range(1, 10):Skip(6):Dump()
end)

Observable.Repeat

Creates an Observable that emits a particular item multiple times

• Declaration:

__Arguments__{ Number, Any * 1 }
function Repeat(count, ...)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->5
	-- [Info]Dump-->5
	-- [Info]Dump-->5
	-- [Info]Dump-->5
	-- [Info]Dump completed
	Observable.Repeat(4, 5):Dump()
end)

Observable.Start

Creates an Observable that emits the return value of a function-like directive, the operation will be processed in a coroutine.

• Declaration:

__Arguments__{ Callable, Any * 0 }
function Start(func, ...)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->16
	-- [Info]Dump completed
	Observable.Start("x=>x^2", 4):Dump()
end)

System.Reactive.Subject

A Subject is a sort of bridge or proxy that acts both as an observer and as an Observable. Because it is observer, it can subscribe to one or more Observables, and because it is Observable, it can pass through the items it observes by reemitting them, and it can also emit new items.

Because a Subject subscribes to an Observable, it will trigger that Observable to begin emitting items (if that Observable is “cold” — that is, if it waits for a subscription before it begins to emit items). This can have the effect of making the resulting Subject a “hot” Observable variant of the original “cold” Observable.

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	local subject = Subject()

	subject:Subscribe(function(...)
		print("Subject Send", ...)
	end)
	subject:Dump()

	-- Subject Send	1
	-- [Info]Dump-->1
	subject:OnNext(1)

	-- Subject Send	2
	-- [Info]Dump-->2
	subject:OnNext(2)

	-- [Info]Dump completed
	subject:OnCompleted()
end)

So we can use several observers to subscribe one subject, they all will receive the data sequences.

We also can use a subject as the observer:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	local subject  = Subject()

	Observable.Range(1, 5):Subscribe(subject)

	-- nothing, since the original observable already completed
	-- when the subject subscribe it
	subject:Dump()

	-- If we give the obervable when create the
	-- subject, the obervable should still be cold
	-- until any observer subscribe the subject
	local subject = Subject(Observable.Range(1, 2))

	-- [Info]Dump-->1
	-- [Info]Dump-->2
	-- [Info]Dump completed
	subject:Dump()
end)

Besides the default Subject class, the system already provide several other subject classes:

AsyncSubject

Only emits the last value (and only the last value) emitted by the source Observable, and only after that source Observable completes

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->10
	-- [Info]Dump completed
	AsyncSubject(Observable.Range(1, 10)):Dump()
end)

BehaviorSubject

Emitting the item most recently emitted by the source Observable (or a seed/default value if none has yet been emitted) and then continues to emit any other items emitted later by the source Observable

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	local subject = BehaviorSubject()
	subject:OnNext(1)
	subject:OnNext(2)

	-- Get the previous data
	-- [Info]Dump-->2
	subject:Dump()

	-- [Info]Dump-->3
	subject:OnNext(3)

	-- [Info]Dump completed
	subject:OnCompleted()
end)

The BehaviorSubject is a special subject, it's used as single value proxy in watch & reactive system:

require "PLoop" (function(_ENV)
	-- Add default handler for Dump method
	Logger.Default:AddHandler(print)

	a = reactive[Number](5)

	-- [Info]Dump-->5
	a:Dump()

	-- 5   5
	print(a.Value, a)

	-- true
	print(a.Value + 3 == a + 3)

	-- [Info]Dump-->4
	a.Value = 4

	-- false	xxxxx.lua:14: the Value must be number, got string
	print(pcall(function() a.Value = "test" end))
end)

We can access its Value property to get/assign value, it has full meta-methods defined to be worked like a real value, but beware that the compare oper like __eq length oper __len not works in the Lua 5.1, the bitwise oper only works after Lua 5.3.

PublishSubject

Only emit the items from the observable source when manully use the Connect method.

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	local subject = PublishSubject(Observable.Range(1, 3))
	subject:Dump()

	-- [Info]Dump-->1
	-- [Info]Dump-->2
	-- [Info]Dump-->3
	-- [Info]Dump completed
	subject:Connect()
end)

ReplaySubject

Emits to any observer all of the items that were emitted by the source Observable(s), regardless of when the observer subscribes

• Declaration:

__Arguments__{ IObservable, Number/nil }
function ReplaySubject(self, observable, max)
end

__Arguments__{ Number/nil }
function ReplaySubject(self, max)
end

• Usage

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- The max size to hold the reply values
	local subject = ReplaySubject(5)

	Observable.Range(1, 10):Do(print,nil,"x=>print('complete')"):Subscribe(subject)

	-- [Info]Dump-->6
	-- [Info]Dump-->7
	-- [Info]Dump-->8
	-- [Info]Dump-->9
	-- [Info]Dump-->10
	subject:Dump()
end)

System.Reactive.Operator

The Operator class provided many useful methods for all IObservable objects(include the objects created from the Observable). We already used the Skip method in the preivous examples.

The operator can be chained to provide useful filter, transform and other usages. Most operator will return a new defer observable.

Utility Operator

Dump

Dump the sequence for test.

• Declaration:

__Arguments__{ NEString/"Dump" }
function Dump(self, name)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Test-->16
	-- [Info]Test completed
	Observable.Start("x=>x^2", 4):Dump("Test")
end)

Do

Invokes actions with side effecting behavior for each element in the observable sequence

• Declaration:

__Arguments__{ Callable, Callable/nil, Callable/nil }
function Do(self, onNext, onError, onCompleted)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	-- System.Logger.Default:AddHandler(print)

	-- Do	1
	-- Do	2
	-- Do	3
	-- Do	4
	-- Do	5
	Observable.Range(1, 5):Do("...=>print('Do', ...)"):Dump()
end)

Catch

Catch the exception and replace with another sequence if provided

• Declaration:

__Arguments__{ Callable }
function Catch(self, onError)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->1
	-- [Info]Dump-->2
	-- [Info]Dump-->3
	-- [Info]Dump completed
	Observable.Throw("Some wrong"):Catch(function(ex)
		return Observable.Range(1, 3)
	end):Dump()

	-- [Info]Dump failed-->Some wrong
	Observable.Throw("Some wrong"):Catch(function() end):Dump()
end)

Finally

Process operations when the sequence is completed, error or unsubscribed

• Declaration:

 __Arguments__{ Callable }
function Finally(self, finally)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->1
	-- [Info]Dump-->2
	-- [Info]Dump-->3
	-- finished
	-- [Info]Dump completed
	Observable.Range(1, 3):Finally("=>print('finished')"):Dump()
end)

OnErrorResumeNext

Start the next observable sequence when the sequence is failed or completed

• Declaration:

__Arguments__{ IObservable }
function OnErrorResumeNext(self, observable)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->1
	-- [Info]Dump-->2
	-- [Info]Dump-->3
	-- [Info]Dump-->10
	-- [Info]Dump completed
	Observable.Range(1, 3):OnErrorResumeNext(Observable.Just(10)):Dump()
end)

Retry

Retry the sequence if failed

• Declaration:

__Arguments__{ NaturalNumber/nil }
function Retry(self, count)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	local count = 0

	-- [Info]Dump-->1
	-- [Info]Dump-->2
	-- [Info]Dump-->4
	-- [Info]Dump-->5
	-- [Info]Dump completed
	Observable(function(observer)
		for i = 1, 5 do
			count = count + 1

			if count == 3 then
				observer:OnError(Exception("Failed"))
			else
				observer:OnNext(count)
			end
		end

		observer:OnCompleted()
	end):Retry(3):Dump()
end)

AsObservable

Encapsulate the sequence as a new observable sequence, so the outside can't access the real sequece directly

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->1
	-- [Info]Dump-->4
	-- [Info]Dump-->9
	-- [Info]Dump-->16
	-- [Info]Dump-->25
	-- [Info]Dump completed
	Observable.Range(1, 5):Map("x=>x^2"):AsObservable():Dump()
end)

ForEach

Process all elements as they arrived, works like the Subscribe, but will block the current coroutine

• Declaration:

__Arguments__{ Callable, Callable/nil, Callable/nil }
function ForEach(self, onNext, onError, onCompleted)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	__Async__() function test(observable)
		observable:ForEach("x=>print(x)")

		print("Finished Test")
	end

	-- 1
	-- 2
	-- 3
	-- 4
	-- 5
	-- Finished Test
	test(Observable.Range(1, 5))
end)

ToIterator

Convert the observable sequence to an iterator, must be used in a coroutine

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	__Async__() function test(observable)
		for k, v in observable:ToIterator() do
			print(k, v)
		end
	end

	-- 1	1
	-- 2	4
	-- 3	9
	-- 4	16
	-- 5	25
	test(Observable.Range(1, 5):Map("x=>x, x^2"))
end)

Filter Operator

Where (Filter)

Applying a filter to a sequence

• Declaration:

__Arguments__{ Callable }
function Where(self, condition)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->6
	-- [Info]Dump-->7
	-- [Info]Dump-->8
	-- [Info]Dump-->9
	-- [Info]Dump-->10
	-- [Info]Dump completed
	Observable.Range(1, 10):Where("x=>x>5"):Dump()
end)

Distinct

Applying a filter that only allow distinct items

• Declaration:

__Arguments__{ Callable/"...=>..." }
function Distinct(self, selector)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->{["id"]=1,["name"]="Ann"}, 1
	-- [Info]Dump-->{["id"]=2,["name"]="Ben"}, 2
	-- [Info]Dump-->{["id"]=3,["name"]="King"}, 4
	-- [Info]Dump completed
	Observable.From(List{
		{ id = 1, name = "Ann" },
		{ id = 2, name = "Ben" },
		{ id = 1, name = "AnotherAnn" },
		{ id = 3, name = "King" },
	}):Distinct("x=>x.id"):Dump()
end)

DistinctUntilChanged

Applying a filter that only value diff from the previous can pass

• Declaration:

__Arguments__{ Callable/"...=>..." }
function DistinctUntilChanged(self, selector)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->1, 1
	-- [Info]Dump-->2, 3
	-- [Info]Dump-->1, 4
	-- [Info]Dump-->3, 5
	-- [Info]Dump-->4, 7
	-- [Info]Dump-->3, 8
	-- [Info]Dump-->5, 9
	Observable.From{ 1, 1, 2, 1, 3, 3, 4, 3, 5}:DistinctUntilChanged():Dump()
end)

IgnoreElements

Ignored all elements, only receive complete or error notifications

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump completed
	Observable.Range(1, 10):IgnoreElements():Dump()
end)

Skip

Skip the given count elements

• Declaration:

__Arguments__{ Number }
function Skip(self, count)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->9
	-- [Info]Dump-->10
	-- [Info]Dump completed
	Observable.Range(1, 10):Skip(8):Dump()
end)

Take

Only take the elements of the given count

• Declaration:

__Arguments__{ Number }
function Take(self, count)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->1
	-- [Info]Dump-->2
	-- [Info]Dump completed
	Observable.Range(1, 10):Take(2):Dump()
end)

SkipWhile

filter out all values until a value fails the predicate, then the remaining sequence can be returned

• Declaration:

__Arguments__{ Callable }
function SkipWhile(self, condition)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->4, 4
	-- [Info]Dump-->3, 5
	-- [Info]Dump-->2, 6
	-- [Info]Dump-->1, 7
	-- [Info]Dump completed
	Observable.From{ 1, 2, 3, 4, 3, 2, 1}:SkipWhile("x=>x<4"):Dump()
end)

TakeWhile

return all values while the predicate passes, and when the first value fails the sequence will complete

• Declaration:

__Arguments__{ Callable }
function TakeWhile(self, condition)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->1, 1
	-- [Info]Dump-->2, 2
	-- [Info]Dump-->3, 3
	-- [Info]Dump completed
	Observable.From{ 1, 2, 3, 4, 3, 2, 1}:TakeWhile("x=>x<4"):Dump()
end)

SkipLast

Skip the last elements of the given count

• Declaration:

__Arguments__{ Number }
function SkipLast(self, last)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->1
	-- [Info]Dump-->2
	-- [Info]Dump-->3
	-- [Info]Dump-->4
	-- [Info]Dump-->5
	-- [Info]Dump-->6
	-- [Info]Dump completed
	Observable.Range(1, 10):SkipLast(4):Dump()
end)

TakeLast

Take the last elements of the given count

• Declaration:

__Arguments__{ Number }
function SkipLast(self, last)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->7
	-- [Info]Dump-->8
	-- [Info]Dump-->9
	-- [Info]Dump-->10
	-- [Info]Dump completed
	Observable.Range(1, 10):TakeLast(4):Dump()
end)

SkipUntil

Skip all values until any value is produced by a secondary observable sequence

• Declaration:

__Arguments__{ IObservable }
function SkipUntil(self, other)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	local subject = Subject()
	local subject2= Subject()

	subject:SkipUntil(subject2):Dump()

	subject:OnNext(1)
	subject:OnNext(2)

	subject2:OnNext("go")

	-- [Info]Dump-->3
	subject:OnNext(3)

	-- [Info]Dump-->4
	subject:OnNext(4)
end)

TakeUntil

Take all values until any value is produced by a secondary observable sequence

• Declaration:

__Arguments__{ IObservable }
function TakeUntil(self, other)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	local subject = Subject()
	local subject2= Subject()

	subject:TakeUntil(subject2):Dump()

	-- [Info]Dump-->1
	subject:OnNext(1)

	-- [Info]Dump-->2
	subject:OnNext(2)

	-- [Info]Dump completed
	subject2:OnNext("go")

	subject:OnNext(3)
	subject:OnNext(4)
end)

MatchPrefix

Take all values that match the prefix elements

• Declaration:

__Arguments__{ System.Any * 1 }
function MatchPrefix(self, ...)
end

• Usage:

require "PLoop" (function(_ENV)
    import "System.Reactive"

    Logger.Default:AddHandler(print)

    local sub = Subject()
    local obs = sub:MatchPrefix("player")

    obs:Dump()

    -- [11/24/20 00:47:14][Info]Dump-->player, 1
    sub:OnNext("player", 1)

    sub:OnNext("target", 2)

    -- [11/24/20 00:47:14][Info]Dump-->player, 3
    sub:OnNext("player", 3)
end)

Inspection Operator

Any

Returns a single value sequence indicate whether the target observable sequence contains any value or value meet the predicate

• Declaration:

__Arguments__{ Callable/"=>true" }
function Any(self, predicate)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->true
	-- [Info]Dump completed
	Observable.Range(1, 10):Any("x=>x>5"):Dump()
end)

All

Returns a single value sequence indicate whether the target observable sequence's all values meet the predicate

• Declaration:

__Arguments__{ Callable }
function All(self, predicate)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->false
	-- [Info]Dump completed
	Observable.Range(1, 10):All("x=>x>2"):Dump()
end)

Contains

Returns a single value sequence indicate whether the target observable sequence contains a specific value

• Declaration:

__Arguments__{ Any }
function Contains(self, expected)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->true
	-- [Info]Dump completed
	Observable.Range(1, 10):Contains(3):Dump()
end)

Default

Returns a sequence with single default value if the target observable sequence doesn't contains any item

• Declaration:

__Arguments__{ System.Any * 1 }
function Default(self, ...)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->100
	-- [Info]Dump completed
	Observable.Range(1, 10):Where("x=>x>20"):Default(100):Dump()
end)

NotEmpty

Raise exception if the sequence doen't provide any elements

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [10/28/20 17:10:51][Info]Dump failed-->The sequence doesn't provide any elements
	Observable.Range(1, 5):ElementAt(10):NotEmpty():Dump()
end)

ElementAt

Returns a sequence with the value at the given index(0-base) of the target observable sequence

• Declaration:

__Arguments__{ NaturalNumber }
function ElementAt(self, index)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->8
	-- [Info]Dump completed
	Observable.Range(1, 10):Skip(2):ElementAt(5):Dump()
end)

SequenceEqual

Compares two observable sequences whether those sequences has the same values in the same order and that the sequences are the same length

• Declaration:

__Arguments__{ IObservable, Callable/"x,y=>x==y" }
function SequenceEqual(self, other, compare)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->100 true
	-- [Info]Dump completed
	Observable.Range(1, 10):SequenceEqual(Observable.From(List(10)):Map("val,key=>val")):Dump()
end)

Aggregation Operator

Aggregate

Returns a sequence with a single value generated from the source sequence

• Declaration:

__Arguments__{ Callable, System.Any/nil }
function Aggregate(self, accumulator, seed)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->55
	-- [Info]Dump completed
	Observable.Range(1, 10):Aggregate(
		function(a, b) return a + b end
	):Dump()
end)

Count

Returns a sequence with a single value being the count of the values in the source sequence

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->7
	-- [Info]Dump completed
	Observable.Range(1, 10):Skip(3):Count():Dump()
end)

Min

Returns a sequence with a single value being the min value of the source sequence

• Declaration:

__Arguments__{ Callable/"x,y=>x<y" }
function Min(self, compare)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->4
	-- [Info]Dump completed
	Observable.Range(1, 10):Skip(3):Min():Dump()
end)

Max

Returns a sequence with a single value being the max value of the source sequence

• Declaration:

__Arguments__{ Callable/"x,y=>x<y" }
function Max(self, compare)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->10
	-- [Info]Dump completed
	Observable.Range(1, 10):Skip(3):Max():Dump()
end)

Sum

Returns a sequence with a single value being the sum value of the source sequence

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->49
	-- [Info]Dump completed
	Observable.Range(1, 10):Skip(3):Sum():Dump()
end)

Average

Returns a sequence with a single value being the average value of the source sequence

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->7
	-- [Info]Dump completed
	Observable.Range(1, 10):Skip(3):Average():Dump()
end)

First

Returns a sequence with a single value being the first value of the source sequence

• Declaration:

__Arguments__{ Callable/"=>true" }
function First(self, predicate)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->4
	-- [Info]Dump completed
	Observable.Range(1, 10):Skip(3):First():Dump()
end)

Last

Returns a sequence with a single value being the last value of the source sequence

• Declaration:

__Arguments__{ Callable/"=>true" }
function Last(self, predicate)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->5
	-- [Info]Dump completed
	Observable.Range(1, 10):Take(5):Last():Dump()
end)

Scan

Returns a sequence with calculated values from the source sequence, if emits the seed, the first value will be used as the seed

• Declaration:

__Arguments__{ Callable, System.Any/nil }
function Scan(self, accumulator, seed)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->3
	-- [Info]Dump-->6
	-- [Info]Dump-->10
	-- [Info]Dump-->15
	-- [Info]Dump-->21
	-- [Info]Dump-->28
	-- [Info]Dump-->36
	-- [Info]Dump-->45
	-- [Info]Dump-->55
	-- [Info]Dump completed
	Observable.Range(1, 10):Scan(function(a,b) return a+b end):Dump()
end)

Partitioning Operator

GroupBy

Returns a sequence with groups generated by the source sequence

• Declaration:

__Arguments__{ Callable }
function GroupBy(self, selector)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Key[1]-->1
	-- [Info]Key[2]-->2
	-- [Info]Key[0]-->3
	-- [Info]Key[1]-->4
	-- [Info]Key[2]-->5
	-- [Info]Key[0]-->6
	-- [Info]Key[1]-->7
	-- [Info]Key[2]-->8
	-- [Info]Key[0]-->9
	-- [Info]Key[1]-->10
	-- [Info]Key[1] completed
	-- [Info]Key[2] completed
	-- [Info]Key[0] completed
	Observable.Range(1, 10)
		:GroupBy(function(i) return i % 3 end)
		:ForEach(
			function(group)
				group:Dump("Key[" .. group.Key .. "]")
			end
		)
end)

This is a little complex compares the previous examples, so the values generate from the source will be checked by the selector, the values will be grouped by the return value(the same return value, the same group).

The GroupBy will generate a new sequence, its values are the groups, the groups are all subjects, with a Key contains the return value from the selector.

MinBy

Returns an observable sequence containing a list of zero or more elements that have a minimum key value

• Declaration:

__Arguments__{ Callable, Callable/"x,y=>x<y" }
function MinBy(self, selector, compare)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->{[1]=3,[2]=6,[3]=9}
	-- [Info]Dump completed
	Observable.Range(1, 10)
		:MinBy(function(i) return i % 3 end)
		:Dump()
end)

MaxBy

Returns an observable sequence containing a list of zero or more elements that have a maximum key value

• Declaration:

__Arguments__{ Callable, Callable/"x,y=>x<y" }
function MaxBy(self, selector, compare)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->{[1]=2,[2]=5,[3]=8}
	-- [Info]Dump completed
	Observable.Range(1, 10)
		:MaxBy(function(i) return i % 3 end)
		:Dump()
end)

Transformation Operator

Select (Map)

Returns an observable sequence with elements converted from the the source sequence

• Declaration:

__Arguments__{ Callable }
function Select(self, selector)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->1
	-- [Info]Dump-->2
	-- [Info]Dump-->0
	-- [Info]Dump-->1
	-- [Info]Dump-->2
	-- [Info]Dump completed
	Observable.Range(1, 5)
		:Select(function(i) return i % 3 end)
		:Dump()
end)

SelectMany (FlatMap)

Convert the source sequence's elements into several observable sequence, then combined those child sequence to produce a final sequence

• Declaration:

__Arguments__{ Callable }
function SelectMany(self, selector)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->10
	-- [Info]Dump-->11
	-- [Info]Dump-->12
	-- [Info]Dump-->20
	-- [Info]Dump-->21
	-- [Info]Dump-->22
	-- [Info]Dump completed
	Observable.Range(1, 2)
		:SelectMany(function(i)
			return Observable.Range(i * 10, i * 10 + 2)
		end)
		:Dump()
end)

Combining Operator

Concat

Concatenates two observable sequences. Returns an observable sequence that contains the elements of the first sequence, followed by those of the second the sequence

• Declaration:

__Arguments__{ IObservable * 1 }
function Concat(self, ...)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->1
	-- [Info]Dump-->2
	-- [Info]Dump-->5
	-- [Info]Dump-->6
	-- [Info]Dump-->10
	-- [Info]Dump-->11
	-- [Info]Dump completed
	Observable.Range(1, 2):Concat(Observable.Range(5, 6), Observable.Range(10, 11)):Dump()
end)

Repeat

Repeats the observable sequence indefinitely and sequentially

• Declaration:

__Arguments__{ NaturalNumber/nil }
function Repeat(self, count)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->1
	-- [Info]Dump-->2
	-- [Info]Dump-->1
	-- [Info]Dump-->2
	-- [Info]Dump completed
	Observable.Range(1, 2):Repeat(2):Dump()
end)

StartWith

Prefix values to a sequence

• Declaration:

__Arguments__{ System.Any * 0 }
function StartWith(self, ...)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->0
	-- [Info]Dump-->1
	-- [Info]Dump-->10
	-- [Info]Dump-->11
	-- [Info]Dump completed
	Observable.Range(10, 11):StartWith(Observable.Range(0, 1)):Dump()
end)

Amb

Return values from the sequence that is first to produce values, and ignore the other sequences

• Declaration:

__Arguments__{ IObservable * 1 }
function Amb(self, ...)
end

-- Also a static method to the Observable
__Static__() __Arguments__{ IObservable * 2 }
function Observable.Amb(...)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	local subject1 = Subject()
	local subject2 = Subject()

	Observable.Amb(subject1, subject2):Dump()

	-- [Info]Dump-->2, 1
	subject2:OnNext(2, 1)

	subject1:OnNext(1, 1)
	subject1:OnNext(1, 2)

	-- [Info]Dump-->2, 2
	subject2:OnNext(2, 2)
end)

Merge

Merge multi sequence, their results will be merged as the result sequence

• Declaration:

__Arguments__{ IObservable * 1 }
function Merge(self, ...)
end

-- Also a static method to the Observable
__Static__() __Arguments__{ IObservable * 2 }
function Observable.Merge(...)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	local subject1 = Subject()
	local subject2 = Subject()

	Observable.Merge(subject1, subject2):Dump()

	-- [Info]Dump-->2, 1
	subject2:OnNext(2, 1)

	-- [Info]Dump-->1, 1
	subject1:OnNext(1, 1)

	-- [Info]Dump-->1, 2
	subject1:OnNext(1, 2)

	-- [Info]Dump-->2, 2
	subject2:OnNext(2, 2)
end)

Switch

Switch will subscribe to the outer sequence and as each inner sequence is yielded it will subscribe to the new inner sequence and dispose of the subscription to the previous inner sequence

• Declaration:

__Arguments__{ IObservable * 1 }
function Switch(self, ...)
end

-- Also a static method to the Observable
__Static__() __Arguments__{ IObservable * 2 }
function Observable.Switch(...)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	local subject1 = Subject()
	local subject2 = Subject()
	local subject3 = Subject()

	Observable.Switch(subject1, subject2, subject3):Dump()

	-- [Info]Dump-->1, 1
	subject1:OnNext(1, 1)

	-- [Info]Dump-->2, 1
	subject2:OnNext(2, 1)

	-- [Info]Dump-->3, 1
	subject3:OnNext(3, 1)

	subject1:OnNext(1, 2)

	subject2:OnNext(2, 2)

	-- [Info]Dump-->3, 2
	subject3:OnNext(3, 2)
end)

CombineLatest

The CombineLatest extension method allows you to take the most recent value from two sequences, and with a given function transform those into a value for the result sequence

• Declaration:

__Arguments__{ IObservable, Callable/"...=>..." }
function CombineLatest(self, secseq, resultSelector)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	local subject1 = Subject()
	local subject2 = Subject()

	subject1:CombineLatest(subject2):Dump()

	subject1:OnNext(1, 1)

	-- [Info]Dump-->1, 1, 2, 1
	subject2:OnNext(2, 1)

	-- [Info]Dump-->1, 2, 2, 1
	subject1:OnNext(1, 2)

	-- [Info]Dump-->1, 2, 2, 2
	subject2:OnNext(2, 2)
end)

Zip

the Zip method brings together two sequences of values as pairs

• Declaration:

__Arguments__{ IObservable, Callable/"...=>..." }
function Zip(self, secseq, resultSelector)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	local subject1 = Subject()
	local subject2 = Subject()

	subject1:Zip(subject2):Dump()

	subject1:OnNext(1, 1)

	-- [Info]Dump-->1, 1, 2, 1
	subject2:OnNext(2, 1)

	subject1:OnNext(1, 2)

	-- [Info]Dump-->1, 2, 2, 2
	subject2:OnNext(2, 2)
end)

Join

combine items emitted by two Observables whenever an item from one Observable is emitted during a time window defined according to an item emitted by the other Observable

• Declaration:

__Arguments__{ IObservable, Callable, Callable, Callable/"...=>..." }
function Join(self, right, leftDurationSelector, rightDurationSelector, resultSelector)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	local subject1 = Subject()
	local subject2 = Subject()

	local leftselector = Subject()
	local rightselector = Subject()

	subject1:Join(subject2, function() return leftselector end, function() return rightselector end):Dump()

	subject1:OnNext(1, 1)

	-- [Info]Dump-->1, 1, 2, 1
	subject2:OnNext(2, 1)

	leftselector:OnNext(true)

	-- [Info]Dump-->1, 2, 2, 1
	subject1:OnNext(1, 2)

	-- [Info]Dump-->1, 2, 2, 2
	subject2:OnNext(2, 2)
end)

Plan Operator

Combine sets of items emitted by two or more Observables by means of Pattern and Plan intermediaries

The And method can be used to generate a Pattern object.

• Declaration:

__Arguments__{ IObservable * 1 }
function And(self, ...)
end

The method will return a Pattern object, it contains two methods used to generate the plan

• Pattern Methods:

-- Add more observables and return the pattern
__Arguments__{ IObservable * 1 }
function Pattern:And(...)
end

-- Generate the plan based on the pattern and result selector
__Arguments__{ Callable/"...=>..." }
function Pattern:Then(resultSelector)
end

The Then method will return a Plan object, it can be used in Observable.When to generate the final observable object.

• Observable.When

-- Return an observable based on the plan
__Static__() __Arguments__{ Plan }
function Observable.When(plan)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->1, 10
	-- [Info]Dump-->2, 11
	-- [Info]Dump completed
	Observable.When(
		Observable.Range(1, 2)
		:And(Observable.Range(10, 11))
		:Then()
	):Dump()
end)

Time-shifted Operator

Buffer

The Buffer operator allows you to store away a range of values and then re-publish them as a list once the buffer is full

• Declaration:

__Arguments__{ NaturalNumber, NaturalNumber/nil }
function Buffer(self, total, skip)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [Info]Dump-->{1,2,3,4,5}
	-- [Info]Dump-->{6,7,8,9,10}
	-- [Info]Dump completed
	Observable.Range(1, 10):Buffer(5):Dump()

	-- [Info]Dump-->{1,2,3,4,5}
	-- [Info]Dump-->{4,5,6,7,8}
	-- [Info]Dump-->{7,8,9,10}
	-- [Info]Dump completed
	Observable.Range(1, 10):Buffer(5, 3):Dump()
end)

To generate the second list, we'll skip the given skip number values, and the default value of the skip is the value of total.

Window

periodically subdivide items from an Observable into Observable windows and emit these windows rather than emitting the items one at a time

• Declaration:

__Arguments__{ NaturalNumber, NaturalNumber/nil }
function Window(self, total, skip)
end

__Arguments__{ IObservable }
function Window(self, sampler)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	local count = 1

	-- [Info][Window:1]-->1
	-- [Info][Window:1]-->2
	-- [Info][Window:1]-->3
	-- [Info][Window:1]-->4
	-- [Info][Window:1]-->5
	-- [Info][Window:1] completed
	-- [Info][Window:2]-->6
	-- [Info][Window:2]-->7
	-- [Info][Window:2]-->8
	-- [Info][Window:2]-->9
	-- [Info][Window:2]-->10
	-- [Info][Window:2] completed
	Observable.Range(1, 10):Window(5):Subscribe(function(observable)
		observable:Dump("[Window:" .. count .. "]")
		count = count + 1
	end)
end)

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	local count = 1
	local main = Subject()
	local sampler = Subject()

	main:Window(sampler):Subscribe(function(observable)
		observable:Dump("[Window:" .. count .. "]")
		count = count + 1
	end)

	-- [Info][Window:1]-->1
	main:OnNext(1)

	-- [Info][Window:1]-->2
	main:OnNext(2)

	-- [Info][Window:1] completed
	sampler:OnNext("step")

	-- [Info][Window:2]-->3
	main:OnNext(3)

	-- [Info][Window:2] completed
	sampler:OnNext("step")

	-- [Info][Window:3]-->4
	main:OnNext(4)
end)

Sample

Returns a new Observable that produces its most recent value every time the specified observable produces a value

• Declaration:

_Arguments__{ IObservable }
function Sample(self, sampler)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	local count = 1
	local main = Subject()
	local sampler = Subject()

	main:Sample(sampler):Dump()

	main:OnNext(1)
	main:OnNext(2)

	-- [Info]Dump-->2
	sampler:OnNext("step")

	main:OnNext(3)

	-- [Info]Dump-->3
	sampler:OnNext("step")

	main:OnNext(4)
end)

Throttle (Debounce)

Ignores values from an observable sequence which are followed by another value before dueTime(may not usable in some platform)

• Declaration:

__Arguments__{ Number }
function Throttle(self, dueTime)
end

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	-- [11:45:26][Info]Dump-->1289501
	-- [11:45:27][Info]Dump-->2714446
	-- [11:45:28][Info]Dump-->4123779
	-- [11:45:29][Info]Dump-->5543850
	-- [11:45:30][Info]Dump-->6868097
	-- [11:45:31][Info]Dump-->8106815
	-- [11:45:32][Info]Dump-->9530516
	-- [11:45:33][Info]Dump completed
	Observable.Range(1, $10^{7}$):Throttle(1):Dump()
end)

Subject Operator

Publish

Generate the PublishSubject from the observable.

• Usage:

require "PLoop" (function(_ENV)
	import "System.Reactive"

	System.Logger.Default:AddHandler(print)

	local subject = Observable.Range(1, 4):Publish()
	subject:Dump()

	-- [Info]Dump-->1
	-- [Info]Dump-->2
	-- [Info]Dump-->3
	-- [Info]Dump-->4
	-- [Info]Dump completed
	subject:Connect()
end)

Replay

Generate a ReplaySubject from the observable

• Declaration:

__Arguments__{ NaturalNumber/nil }
function Replay(self, size)
end

ToSubject

Generate a subject from the observable

• Declaration:

__Arguments__{ -Subject/Subject }
function ToSubject(self, subject)
end