Observable.mon

/* 
* Copyright (c) 2018 Software AG, Darmstadt, Germany and/or its licensors
*
* SPDX-License-Identifier: Apache-2.0
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
*    http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
                                                
package com.industry.rx_epl;

using com.apama.exceptions.Exception;

using com.industry.rx_epl.operators.Amb;
using com.industry.rx_epl.operators.Async;
using com.industry.rx_epl.operators.Average;
using com.industry.rx_epl.operators.AverageDecimal;
using com.industry.rx_epl.operators.Buffer;
using com.industry.rx_epl.operators.BufferCount;
using com.industry.rx_epl.operators.BufferCountSkip;
using com.industry.rx_epl.operators.BufferTime;
using com.industry.rx_epl.operators.BufferTimeOrCount;
using com.industry.rx_epl.operators.CatchError;
using com.industry.rx_epl.operators.CombineLatest;
using com.industry.rx_epl.operators.CombineLatestToSequence;
using com.industry.rx_epl.operators.ComplexPipe;
using com.industry.rx_epl.operators.ComplexPipeOn;
using com.industry.rx_epl.operators.ComplexPipeOnNew;
using com.industry.rx_epl.operators.Concat;
using com.industry.rx_epl.operators.ConcatString;
using com.industry.rx_epl.operators.Contains;
using com.industry.rx_epl.operators.Count;
using com.industry.rx_epl.operators.Debounce;
using com.industry.rx_epl.operators.DefaultIfEmpty;
using com.industry.rx_epl.operators.Delay;
using com.industry.rx_epl.operators.Distinct;
using com.industry.rx_epl.operators.DistinctBy;
using com.industry.rx_epl.operators.DistinctByField;
using com.industry.rx_epl.operators.DistinctByUntilChanged;
using com.industry.rx_epl.operators.DistinctByFieldUntilChanged;
using com.industry.rx_epl.operators.DistinctUntilChanged;
using com.industry.rx_epl.operators.Do;
using com.industry.rx_epl.operators.ElementAt;
using com.industry.rx_epl.operators.Every;
using com.industry.rx_epl.operators.Filter;
using com.industry.rx_epl.operators.First;
using com.industry.rx_epl.operators.FlatMap;
using com.industry.rx_epl.operators.GroupBy;
using com.industry.rx_epl.operators.GroupByField;
using com.industry.rx_epl.operators.GroupByWindow;
using com.industry.rx_epl.operators.IgnoreElements;
using com.industry.rx_epl.operators.Last;
using com.industry.rx_epl.operators.Let;
using com.industry.rx_epl.operators.Map;
using com.industry.rx_epl.operators.Max;
using com.industry.rx_epl.operators.MaxDecimal;
using com.industry.rx_epl.operators.MaxFloat;
using com.industry.rx_epl.operators.MaxInteger;
using com.industry.rx_epl.operators.Min;
using com.industry.rx_epl.operators.MinDecimal;
using com.industry.rx_epl.operators.MinFloat;
using com.industry.rx_epl.operators.MinInteger;
using com.industry.rx_epl.operators.Merge;
using com.industry.rx_epl.operators.MergeAll;
using com.industry.rx_epl.operators.Pausable;
using com.industry.rx_epl.operators.PausableBuffered;
using com.industry.rx_epl.operators.Pairwise;
using com.industry.rx_epl.operators.Pipe;
using com.industry.rx_epl.operators.PipeOn;
using com.industry.rx_epl.operators.PipeOnNew;
using com.industry.rx_epl.operators.Pluck;
using com.industry.rx_epl.operators.Reduce;
using com.industry.rx_epl.operators.ReduceWithInitial;
using com.industry.rx_epl.operators.Repeat;
using com.industry.rx_epl.operators.Retry;
using com.industry.rx_epl.operators.Sample;
using com.industry.rx_epl.operators.SampleCount;
using com.industry.rx_epl.operators.SampleTime;
using com.industry.rx_epl.operators.SampleTimeOrCount;
using com.industry.rx_epl.operators.Scan;
using com.industry.rx_epl.operators.ScanWithInitial;
using com.industry.rx_epl.operators.SequenceEqual;
using com.industry.rx_epl.operators.Skip;
using com.industry.rx_epl.operators.SkipLast;
using com.industry.rx_epl.operators.SkipUntil;
using com.industry.rx_epl.operators.SkipWhile;
using com.industry.rx_epl.operators.Sort;
using com.industry.rx_epl.operators.SortAsc;
using com.industry.rx_epl.operators.SortDesc;
using com.industry.rx_epl.operators.StartWith;
using com.industry.rx_epl.operators.Sum;
using com.industry.rx_epl.operators.SumDecimal;
using com.industry.rx_epl.operators.SumFloat;
using com.industry.rx_epl.operators.SumInteger;
using com.industry.rx_epl.operators.SwitchMap;
using com.industry.rx_epl.operators.SwitchOnNext;
using com.industry.rx_epl.operators.Take;
using com.industry.rx_epl.operators.TakeLast;
using com.industry.rx_epl.operators.TakeUntil;
using com.industry.rx_epl.operators.TakeWhile;
using com.industry.rx_epl.operators.ThrottleFirst;
using com.industry.rx_epl.operators.ThrottleLast;
using com.industry.rx_epl.operators.TimeInterval;
using com.industry.rx_epl.operators.Timeout;
using com.industry.rx_epl.operators.Timestamp;
using com.industry.rx_epl.operators.ToSortedList;
using com.industry.rx_epl.operators.ToSortedListAsc;
using com.industry.rx_epl.operators.ToSortedListDesc;
using com.industry.rx_epl.operators.UpdateTimestamp;
using com.industry.rx_epl.operators.WithLatestFrom;
using com.industry.rx_epl.operators.WithLatestFromToSequence;
using com.industry.rx_epl.operators.WindowCount;
using com.industry.rx_epl.operators.WindowTime;
using com.industry.rx_epl.operators.WindowTimeOrCount;
using com.industry.rx_epl.operators.Zip;
using com.industry.rx_epl.operators.ZipToSequence;

using com.industry.rx_epl.operators.internals.CreateOnConnection;
using com.industry.rx_epl.operators.internals.EmptyNeverError;
using com.industry.rx_epl.operators.internals.FromChannelOnConnection;
using com.industry.rx_epl.operators.internals.FromIteratorOnConnection;
using com.industry.rx_epl.operators.internals.FromStreamOnConnection;
using com.industry.rx_epl.operators.internals.FromOnConnection;
using com.industry.rx_epl.operators.internals.GetSyncHandler;
using com.industry.rx_epl.operators.internals.IntervalOnConnection;
using com.industry.rx_epl.operators.internals.JustOnConnection;
using com.industry.rx_epl.operators.internals.ObserveOnHandler;
using com.industry.rx_epl.operators.internals.PublishHandler;
using com.industry.rx_epl.operators.internals.PublishReplayHandler;
using com.industry.rx_epl.operators.internals.ObserveToHandler;
using com.industry.rx_epl.operators.internals.RangeOnConnection;
using com.industry.rx_epl.operators.internals.ObserveFromDispose;
using com.industry.rx_epl.operators.internals.ObserveFromOnConnection;
using com.industry.rx_epl.operators.internals.RefCountOnConnection;
using com.industry.rx_epl.operators.internals.StartOnConnection;
using com.industry.rx_epl.operators.internals.SubscribeOnConnection;
using com.industry.rx_epl.operators.internals.SubscribeOn;
using com.industry.rx_epl.operators.internals.ToStream;
using com.industry.rx_epl.operators.internals.ToChannel;

event Observable {	
	action<IObserver> returns ISubscription onConnection;
	
	boolean connectable;
	IDisposable connection;
	
	action<> returns IDisposable connect;
	
	/** @private */
	static action _create() returns Observable {
		Observable o := new Observable;
		o.connect := invalidConnect;
		return o;
	}
	
	/** @private */
	static action _createAsIObservable() returns IObservable {
		Observable o := new Observable;
		o.connect := invalidConnect;
		return o.asIObservable();
	}
	
	/** @private */
	static action _createConnectable(IDisposable connection, action<> returns IDisposable connect) returns Observable {
		Observable o := new Observable;
		o.connect := connect;
		o.connectable := true;
		o.connection := connection;
		return o;
	}

	static action create(action<IResolver> resolver) returns IObservable {
		// A user defined resolver cannot handle pull requests so it's hot
		Observable o := _create();
		o.onConnection := CreateOnConnection.create(resolver);
		return o.asIObservable();
	}

	static action fromValues(any values) returns IObservable {
		Observable o := _create();
		o.onConnection := FromOnConnection.create(values);
		return o.asIObservable();
	}
	
	static action fromIterator(action<> returns any iterator) returns IObservable {
		Observable o := _create();
		o.onConnection := FromIteratorOnConnection.create(iterator);
		return o.asIObservable();
	}
	
	static action just(any value) returns IObservable {
		Observable o := _create();
		o.onConnection := JustOnConnection.create(value);
		return o.asIObservable();		
	}
	
	/**
	 * Note: the channel listener start asynchronously - so won't receive events until a short time later
	 */
	static action fromChannel(string channelName) returns IObservable {
		Observable o := _create();
		o.onConnection := FromChannelOnConnection.create(channelName);
		return o.asIObservable();		
	}

	/**
	 * Generates a series of event with an incrementing float value
	 * @param interval the time between subsequent events
	 */
	static action interval(float seconds) returns IObservable {
		Observable o := _create();
		o.onConnection := IntervalOnConnection.create(seconds);
		return o.asIObservable();	
	}
	
	static action range(integer start, integer end) returns IObservable {
		Observable o := _create();
		o.onConnection := RangeOnConnection.create(start, end);
		return o.asIObservable();	
	}
	
	static action repeat(any value, integer count) returns IObservable {
		Observable j := <Observable> just(value).observable;
		return j.repeatObservable(count);
	}
	
	static action combineLatest(sequence<IObservable> observables, action<sequence<any> > returns any combiner) returns IObservable {
		IObservable first := observables[0];
		
		sequence<IObservable> others := (new sequence<IObservable>);
		others.appendSequence(observables);
		others.remove(0);
		
		return first.combineLatest(others, combiner);
	}
	
	static action start(action<> returns any valueGenerator) returns IObservable {
		Observable o := _create();
		o.onConnection := StartOnConnection.create(valueGenerator);
		return o.asIObservable();
	}
	
	static action merge(sequence<IObservable> observables) returns IObservable {
		IObservable first := observables[0];
		
		sequence<IObservable> others := (new sequence<IObservable>);
		others.appendSequence(observables);
		others.remove(0);
		
		return first.merge(others);
	}
	
	static action zip(sequence<IObservable> observables, action<sequence<any> > returns any combiner) returns IObservable {
		IObservable first := observables[0];
		
		sequence<IObservable> others := (new sequence<IObservable>);
		others.appendSequence(observables);
		others.remove(0);
		
		return first.zip(others, combiner);
	}
	
	static action fromStream(stream<any> valueStream) returns IObservable {
		Observable o := _create();
		o.onConnection := FromStreamOnConnection.create(valueStream);
		return o.asIObservable();
	}
	
	static action timer(any value, float delay) returns IObservable {
		return just(value).delay(delay);
	}

	static action observeFromChannel(string channelName) returns IObservable {
		Observable o := _create();
		o.onConnection := ObserveFromOnConnection.create(channelName);
		return o.asIObservable();
	}
	
	static action concat(sequence<IObservable> observables) returns IObservable {
		IObservable first := observables[0];
		
		sequence<IObservable> others := (new sequence<IObservable>);
		others.appendSequence(observables);
		others.remove(0);
		
		return first.concat(others);
	}

	static action empty() returns IObservable {
		return create(EmptyNeverError.emptyGenerator);
	}
	
	static action never() returns IObservable {
		return create(EmptyNeverError.neverGenerator);
	}
	
	static action error() returns IObservable {
		return create(EmptyNeverError.errorGenerator);
	}
	
	static action sequenceEqual(sequence<IObservable> observables) returns IObservable {
		IObservable first := observables[0];
		
		sequence<IObservable> others := (new sequence<IObservable>);
		others.appendSequence(observables);
		others.remove(0);
		
		return first.sequenceEqual(others);
	}
	
	static action amb(sequence<IObservable> observables) returns IObservable {
		IObservable first := observables[0];
		
		sequence<IObservable> others := (new sequence<IObservable>);
		others.appendSequence(observables);
		others.remove(0);
		
		return first.amb(others);
	}
	
	static action combineLatestToSequence(sequence<IObservable> observables) returns IObservable {
		IObservable first := observables[0];
		
		sequence<IObservable> others := (new sequence<IObservable>);
		others.appendSequence(observables);
		others.remove(0);
		
		return first.combineLatestToSequence(others);
	}
	
	static action zipToSequence(sequence<IObservable> observables) returns IObservable {
		IObservable first := observables[0];
		
		sequence<IObservable> others := (new sequence<IObservable>);
		others.appendSequence(observables);
		others.remove(0);
		
		return first.zipToSequence(others);
	}
	
	action connectObserver(IObserver observer) returns ISubscription {
		return onConnection(observer);
	}
	
	action subscribe(Subscriber subscriber) returns ISubscription {
		Observable o := _create();
		
		IObserver observer := subscriber.asIObserver();

		o.onConnection := SubscribeOnConnection.create(onConnection); 
		
		return o.onConnection(observer);
	}
	
	action map(any mapper) returns IObservable {
		Observable o := _create();		
		o.onConnection := Map.create(mapper)(onConnection);
		return o.asIObservable();
	}
	
	action publish() returns IObservable {
		IDisposable connection := IDisposable(new action<>);
		PublishHandler p := PublishHandler.create(onConnection, connection);
		Observable o := _createConnectable(connection, p.connect);
		
		o.onConnection := p.onConnection;
		
		return o.asIObservable();
	}
	
	action refCount() returns IObservable {
		if not connectable {
			throw Exception("Can only refCount a published Observable", "UnsupportedOperation");
		}
		Observable o := _create();
		o.onConnection := RefCountOnConnection.create(connection, connect, onConnection);
		return o.asIObservable();
	}
	
	action share() returns IObservable {
		return publish().refCount();
	}

	action take(integer count) returns IObservable {
		Observable o := _create();
		o.onConnection := Take.create(count)(onConnection);
		return o.asIObservable();	
	}

	action takeLast(integer count) returns IObservable {
		Observable o := _create();
		o.onConnection := TakeLast.create(count)(onConnection);
		return o.asIObservable();	
	}

	action first() returns IObservable {
		Observable o := _create();
		o.onConnection := First.create()(onConnection);
		return o.asIObservable();	
	}
	
	action last() returns IObservable {
		Observable o := _create();
		o.onConnection := Last.create()(onConnection);
		return o.asIObservable();	
	}
	
	action skip(integer count) returns IObservable {
		Observable o := _create();
		o.onConnection := Skip.create(count)(onConnection);
		return o.asIObservable();			
	}

	action skipLast(integer count) returns IObservable {
		Observable o := _create();
		o.onConnection := SkipLast.create(count)(onConnection);
		return o.asIObservable();			
	}
	
	action filter(any predicate) returns IObservable {
		Observable o := _create();
		o.onConnection := Filter.create(predicate)(onConnection);
		return o.asIObservable();			
	}

	action scan(any scanner) returns IObservable {
		Observable o := _create();
		o.onConnection := Scan.create(scanner)(onConnection);
		return o.asIObservable();		
	}
	
	action scanWithInitial(any scanner, any initialValue) returns IObservable {
		Observable o := _create();
		o.onConnection := ScanWithInitial.create(scanner, initialValue)(onConnection);
		return o.asIObservable();		
	}
	
	action distinct() returns IObservable {
		Observable o := _create();
		o.onConnection := Distinct.create()(onConnection);
		return o.asIObservable();		
	}
	
	action distinctBy(any by) returns IObservable {
		Observable o := _create();
		o.onConnection := DistinctBy.create(by)(onConnection);
		return o.asIObservable();			
	}
	
	action combineLatestFromInstance(sequence<IObservable> otherObservables, action<sequence<any> > returns any combiner) returns IObservable {
		Observable o := _create();
		o.onConnection := CombineLatest.create(otherObservables, combiner)(onConnection);
		return o.asIObservable();
	}
	
	action withLatestFrom(sequence<IObservable> otherObservables, action<sequence<any> > returns any combiner) returns IObservable {
		Observable o := _create();
		o.onConnection := WithLatestFrom.create(otherObservables, combiner)(onConnection);
		return o.asIObservable();
	}
	
	action delay(float seconds) returns IObservable {
		Observable o := _create();
		o.onConnection := Delay.create(seconds)(onConnection);
		return o.asIObservable();
	}
	
	action async() returns IObservable {
		Observable o := _create();
		o.onConnection := Async.create()(onConnection);
		return o.asIObservable();
	}
	
	action do(Subscriber doer) returns IObservable {
		Observable o := _create();
		o.onConnection := Do.create(doer)(onConnection);
		return o.asIObservable();
	}
	
	/**
	 * Note: on termination of the connection the replay buffer will be emptied
	 */
	action publishReplay(integer count) returns IObservable {
		IDisposable connection := IDisposable(new action<>);
		
		Observable publishObservable := <Observable> publish().observable;
		PublishReplayHandler replayHandler := PublishReplayHandler.create(publishObservable.connection, connection, publishObservable.connect, publishObservable.onConnection, count);
		Observable o := _createConnectable(connection, replayHandler.downstreamConnect);
		
		o.onConnection := replayHandler.onConnection;
		
		return o.asIObservable();
	}
	
	/**
	 * Note: Unlike in rxjs v5.5 (but like rxjs v5.4) - if the refcount drops to 0 then the source will be disconnected from and the replay buffer will be emptied
	 */
	action shareReplay(integer count) returns IObservable {
		return publishReplay(count).refCount();
	}
	
	action takeUntil(IObservable until) returns IObservable {
		Observable o := _create();
		o.onConnection := TakeUntil.create(until)(onConnection);
		return o.asIObservable();
	}
	
	// Called repeat when accessed via the interface
	action repeatObservable(integer count) returns IObservable {
		Observable o := _create();
		o.onConnection := Repeat.create(count)(onConnection);
		return o.asIObservable();
	}
	
	action flatMap(any mapper) returns IObservable {
		Observable o := _create();
		o.onConnection := FlatMap.create(mapper)(onConnection);
		return o.asIObservable();
	}
	
	action mergeFromInstance(sequence<IObservable> otherObservables) returns IObservable {
		Observable o := _create();
		o.onConnection := Merge.create(otherObservables)(onConnection);
		return o.asIObservable();
	}
	
	action zipFromInstance(sequence<IObservable> otherObservables, action<sequence<any> > returns any combiner) returns IObservable {
		Observable o := _create();
		o.onConnection := Zip.create(otherObservables, combiner)(onConnection);
		return o.asIObservable();
	}
	
	action switchOnNext() returns IObservable {
		Observable o := _create();
		o.onConnection := SwitchOnNext.create()(onConnection);
		return o.asIObservable();
	}
	
	action skipUntil(IObservable until) returns IObservable {
		Observable o := _create();
		o.onConnection := SkipUntil.create(until)(onConnection);
		return o.asIObservable();
	}
	
	action elementAt(integer n) returns IObservable {
		Observable o := _create();
		o.onConnection := ElementAt.create(n)(onConnection);
		return o.asIObservable();
	}
	
	action reduce(any reducer) returns IObservable {
		Observable o := _create();
		o.onConnection := Reduce.create(reducer)(onConnection);
		return o.asIObservable();
	}
	
	action reduceWithInitial(any reducer, any initialValue) returns IObservable {
		Observable o := _create();
		o.onConnection := ReduceWithInitial.create(reducer, initialValue)(onConnection);
		return o.asIObservable();
	}
	
	/**
	 * The stream will automatically quit and dispose if/when the observable does.
	 * It is necessary to call dispose on the disposableStream if quit is called manually on the stream.
	 */
	action toStream() returns DisposableStream {
		return ToStream.create(self.asIObservable());
	}
	
	action catchError(IObservable onError) returns IObservable {
		Observable o := _create();
		o.onConnection := CatchError.create(onError)(onConnection);
		return o.asIObservable();
	}
	
	action retry(integer retryCount) returns IObservable {
		Observable o := _create();
		o.onConnection := Retry.create(retryCount)(onConnection);
		return o.asIObservable();
	}
	
	/**
	 * Dispose should be called once all downstreams have finished and no new connections are expected otherwise the application leaks a connection listener (Not a big deal)
	 */
	action observeOn(action<IObservable, action<> > observableConstructorAndDisposer, context targetContext) returns IDisposable {
		return ObserveOnHandler.create(observeToChannel, observeFromChannel, targetContext, observableConstructorAndDisposer);
	}
	
	/**
	 * Note: The resulting observables are hot so if you don't want to miss any then they should each be publishReplay(n).connect()'ed
	 */
	action groupBy(any groupBy) returns IObservable {
		Observable o := _create();
		o.onConnection := GroupBy.create(groupBy)(onConnection);
		return o.asIObservable();
	}
	
	/**
	 * Dispose should be called once all downstreams have finished and no new connections are expected otherwise the application leaks a connection listener (Not a big deal)
	 */
	action observeToChannel(string channelName) returns IDisposable {
		return ObserveToHandler.create(onConnection, channelName);
	}

	action observeOnNew(action<IObservable, action<> > observableConstructorAndDisposer) returns IDisposable {
		return ObserveOnHandler.create(observeToChannel, observeFromChannel, context("ObserveOnNew" + integer.getUnique().toString(), false), observableConstructorAndDisposer);
	}
	
	action every(any predicate) returns IObservable {
		Observable o := _create();
		o.onConnection := Every.create(predicate)(onConnection);
		return o.asIObservable();	
	}

	action contains(any predicate) returns IObservable {
		Observable o := _create();
		o.onConnection := Contains.create(predicate)(onConnection);
		return o.asIObservable();	
	}
	
	action takeWhile(any predicate) returns IObservable {
		Observable o := _create();
		o.onConnection := TakeWhile.create(predicate)(onConnection);
		return o.asIObservable();	
	}

	action skipWhile(any predicate) returns IObservable {
		Observable o := _create();
		o.onConnection := SkipWhile.create(predicate)(onConnection);
		return o.asIObservable();	
	}
	
	action concatFromInstance(sequence<IObservable> otherObservables) returns IObservable {
		Observable o := _create();
		o.onConnection := Concat.create(otherObservables)(onConnection);
		return o.asIObservable();	
	}
	
	action startWith(sequence<any> values) returns IObservable {
		Observable o := _create();
		o.onConnection := StartWith.create(values)(onConnection);
		return o.asIObservable();	
	}
	
	action defaultIfEmpty(any defaultValue) returns IObservable {
		Observable o := _create();
		o.onConnection := DefaultIfEmpty.create(defaultValue)(onConnection);
		return o.asIObservable();	
	}
	
	action ignoreElements() returns IObservable {
		Observable o := _create();
		o.onConnection := IgnoreElements.create()(onConnection);
		return o.asIObservable();	
	}
	
	action count() returns IObservable {
		Observable o := _create();
		o.onConnection := Count.create()(onConnection);
		return o.asIObservable();
	}
	
	/**
	 * Takes any numeric type (integer, float, decimal)
	 * The result is always a float
	 */
	action sum() returns IObservable {
		Observable o := _create();
		o.onConnection := Sum.create()(onConnection);
		return o.asIObservable();
	}
	
	action sumInteger() returns IObservable {
		Observable o := _create();
		o.onConnection := SumInteger.create()(onConnection);
		return o.asIObservable();
	}
	
	action sumFloat() returns IObservable {
		Observable o := _create();
		o.onConnection := SumFloat.create()(onConnection);
		return o.asIObservable();
	}
	
	action sumDecimal() returns IObservable {
		Observable o := _create();
		o.onConnection := SumDecimal.create()(onConnection);
		return o.asIObservable();
	}
	
	/**
	 * Will use .toString() for any non-string types
	 */
	action concatString() returns IObservable {
		Observable o := _create();
		o.onConnection := ConcatString.create()(onConnection);
		return o.asIObservable();
	}
	
	/**
	 * Takes any numeric type (integer, float, decimal)
	 * The result is always a float
	 */
	action max() returns IObservable {
		Observable o := _create();
		o.onConnection := Max.create()(onConnection);
		return o.asIObservable();
	}
	
	action maxInteger() returns IObservable {
		Observable o := _create();
		o.onConnection := MaxInteger.create()(onConnection);
		return o.asIObservable();
	}
	
	action maxFloat() returns IObservable {
		Observable o := _create();
		o.onConnection := MaxFloat.create()(onConnection);
		return o.asIObservable();
	}
	
	action maxDecimal() returns IObservable {
		Observable o := _create();
		o.onConnection := MaxDecimal.create()(onConnection);
		return o.asIObservable();
	}
	
	/**
	 * Takes any numeric type (integer, float, decimal)
	 * The result is always a float
	 */
	action min() returns IObservable {
		Observable o := _create();
		o.onConnection := Min.create()(onConnection);
		return o.asIObservable();
	}
	
	action minInteger() returns IObservable {
		Observable o := _create();
		o.onConnection := MinInteger.create()(onConnection);
		return o.asIObservable();
	}
	
	action minFloat() returns IObservable {
		Observable o := _create();
		o.onConnection := MinFloat.create()(onConnection);
		return o.asIObservable();
	}
	
	action minDecimal() returns IObservable {
		Observable o := _create();
		o.onConnection := MinDecimal.create()(onConnection);
		return o.asIObservable();
	}
	
	/**
	 * Takes any numeric type (integer, float, decimal)
	 * The result is always a float
	 */
	action average() returns IObservable {
		Observable o := _create();
		o.onConnection := Average.create()(onConnection);
		return o.asIObservable();
	}
	
	/**
	 * Takes any numeric type (integer, float, decimal)
	 * The result is always a decimal
	 */
	action averageDecimal() returns IObservable {
		Observable o := _create();
		o.onConnection := AverageDecimal.create()(onConnection);
		return o.asIObservable();
	}
	
	action sequenceEqualFromInstance(sequence<IObservable> otherObservables) returns IObservable {
		Observable o := _create();		
		o.onConnection := SequenceEqual.create(otherObservables)(onConnection);
		return o.asIObservable();	
	}
	
	action ambFromInstance(sequence<IObservable> otherObservables) returns IObservable {
		Observable o := _create();		
		o.onConnection := Amb.create(otherObservables)(onConnection);
		return o.asIObservable();
	}
	
	action timestamp() returns IObservable {
		Observable o := _create();		
		o.onConnection := Timestamp.create()(onConnection);
		return o.asIObservable();
	}
	
	action updateTimestamp() returns IObservable {
		Observable o := _create();		
		o.onConnection := UpdateTimestamp.create()(onConnection);
		return o.asIObservable();
	}
	
	action timeout(float timeout) returns IObservable {
		Observable o := _create();
		o.onConnection := Timeout.create(timeout)(onConnection);
		return o.asIObservable();	
	}
	
	action getSync() returns any {
		return GetSyncHandler.create(self.asIObservable(), new any);
	}
	
	action getSyncOr(any defaultValue) returns any {		
		return GetSyncHandler.create(self.asIObservable(), defaultValue);
	}
	
	action timeInterval() returns IObservable {
		Observable o := _create();
		o.onConnection := TimeInterval.create()(onConnection);
		return o.asIObservable();	
	}
	
	action distinctUntilChanged() returns IObservable {
		Observable o := _create();
		o.onConnection := DistinctUntilChanged.create()(onConnection);
		return o.asIObservable();	
	}
	
	action distinctByUntilChanged(any by) returns IObservable {
		Observable o := _create();
		o.onConnection := DistinctByUntilChanged.create(by)(onConnection);
		return o.asIObservable();	
	}
	
	action switchMap(any mapper) returns IObservable {
		Observable o := _create();
		o.onConnection := SwitchMap.create(mapper)(onConnection);
		return o.asIObservable();
	}
	
	action debounce(float duration) returns IObservable {
		Observable o := _create();
		o.onConnection := Debounce.create(duration)(onConnection);
		return o.asIObservable();
	}
	
	action groupByWindow(IObservable trigger) returns IObservable {
		Observable o := _create();
		o.onConnection := GroupByWindow.create(trigger)(onConnection);
		return o.asIObservable();
	}
	
	action throttleFirst(float throttleDuration) returns IObservable {
		Observable o := _create();
		o.onConnection := ThrottleFirst.create(throttleDuration)(onConnection);
		return o.asIObservable();	
	}
	
	action throttleLast(float throttleDuration) returns IObservable {
		Observable o := _create();
		o.onConnection := ThrottleLast.create(throttleDuration)(onConnection);
		return o.asIObservable();	
	}
	
	action sample(IObservable trigger) returns IObservable {
		Observable o := _create();
		o.onConnection := Sample.create(trigger)(onConnection);
		return o.asIObservable();
	}
	
	action buffer(IObservable trigger) returns IObservable {
		Observable o := _create();
		o.onConnection := Buffer.create(trigger)(onConnection);
		return o.asIObservable();	
	}
	
	action bufferTime(float duration) returns IObservable {
		Observable o := _create();
		o.onConnection := BufferTime.create(duration)(onConnection);
		return o.asIObservable();
	}
	
	action bufferCount(integer count) returns IObservable {
		Observable o := _create();
		o.onConnection := BufferCount.create(count)(onConnection);
		return o.asIObservable();	
	}
	
	action bufferTimeOrCount(float duration, integer count) returns IObservable {
		Observable o := _create();
		o.onConnection := BufferTimeOrCount.create(duration, count)(onConnection);
		return o.asIObservable();		
	}
	
	action combineLatestToSequenceFromInstance(sequence<IObservable> otherObservables) returns IObservable {
		Observable o := _create();		
		o.onConnection := CombineLatestToSequence.create(otherObservables)(onConnection);
		return o.asIObservable();	
	}
	
	action withLatestFromToSequence(sequence<IObservable> otherObservables) returns IObservable {
		Observable o := _create();
		o.onConnection := WithLatestFromToSequence.create(otherObservables)(onConnection);
		return o.asIObservable();	
	}
	
	action zipToSequenceFromInstance(sequence<IObservable> otherObservables) returns IObservable {
		Observable o := _create();
		o.onConnection := ZipToSequence.create(otherObservables)(onConnection);
		return o.asIObservable();
	}
	
	action windowTime(float time) returns IObservable {
		Observable o := _create();
		o.onConnection := WindowTime.create(time)(onConnection);
		return o.asIObservable();
	}
	
	action windowCount(integer count) returns IObservable {
		Observable o := _create();
		o.onConnection := WindowCount.create(count)(onConnection);
		return o.asIObservable();
	}
	
	action windowTimeOrCount(float time, integer count) returns IObservable {
		Observable o := _create();
		o.onConnection := WindowTimeOrCount.create(time, count)(onConnection);
		return o.asIObservable();
	}
	
	action sampleTime(float time) returns IObservable {
		Observable o := _create();
		o.onConnection := SampleTime.create(time)(onConnection);
		return o.asIObservable();
	}
	
	action sampleCount(integer count) returns IObservable {
		Observable o := _create();
		o.onConnection := SampleCount.create(count)(onConnection);
		return o.asIObservable();
	}
	
	action sampleTimeOrCount(float time, integer count) returns IObservable {
		Observable o := _create();
		o.onConnection := SampleTimeOrCount.create(time, count)(onConnection);
		return o.asIObservable();
	}

	action bufferCountSkip(integer count, integer skip) returns IObservable {
		Observable o := _create();
		o.onConnection := BufferCountSkip.create(count, skip)(onConnection);
		return o.asIObservable();	
	}
	
	action pluck(any fieldName) returns IObservable {
		Observable o := _create();
		o.onConnection := Pluck.create(fieldName)(onConnection);
		return o.asIObservable();	
	}
	
	action pipe(sequence<action<action<IObserver> returns ISubscription> returns action<IObserver> returns ISubscription> pipeModifiers) returns IObservable {
		Observable o := _create();
		o.onConnection := Pipe.create(pipeModifiers)(onConnection);
		return o.asIObservable();
	}
	
	action mergeAll() returns IObservable {
		Observable o := _create();
		o.onConnection := MergeAll.create()(onConnection);
		return o.asIObservable();
	}
	
	action pipeOn(sequence<action<action<IObserver> returns ISubscription> returns action<IObserver> returns ISubscription> pipeModifiers, context targetContext) returns IObservable {
		Observable o := _create();
		o.onConnection := PipeOn.create(pipeModifiers, targetContext)(onConnection);
		return o.asIObservable();
	}
	
	action pipeOnNew(sequence<action<action<IObserver> returns ISubscription> returns action<IObserver> returns ISubscription> pipeModifiers) returns IObservable {
		Observable o := _create();
		o.onConnection := PipeOnNew.create(pipeModifiers)(onConnection);
		return o.asIObservable();
	}
	
	action let(action<action<IObserver> returns ISubscription> returns action<IObserver> returns ISubscription pipeModifier) returns IObservable {
		Observable o := _create();
		o.onConnection := Let.create(pipeModifier)(onConnection);
		return o.asIObservable();
	}
	
	action subscribeOn(Subscriber subscriber, context targetContext) returns ISubscription {
		return SubscribeOn.create(subscribe, subscriber, targetContext);
	}
	
	action subscribeOnNew(Subscriber subscriber) returns ISubscription {
		return SubscribeOn.create(subscribe, subscriber, context("SubscribeOnContext" + integer.getUnique().toString(), false));
	}
	
	action pairwise() returns IObservable {
		Observable o := _create();
		o.onConnection := Pairwise.create()(onConnection);
		return o.asIObservable();
	}
	
	action groupByField(any fieldName) returns IObservable {
		Observable o := _create();
		o.onConnection := GroupByField.create(fieldName)(onConnection);
		return o.asIObservable();
	}
	
	action complexPipe(action<IObservable> returns IObservable complexPipe) returns IObservable {
		Observable o := _create();
		o.onConnection := ComplexPipe.create(complexPipe)(onConnection);
		return o.asIObservable();
	}
	
	action complexPipeOn(action<IObservable> returns IObservable complexPipe, context targetContext) returns IObservable {
		Observable o := _create();
		o.onConnection := ComplexPipeOn.create(complexPipe, targetContext)(onConnection);
		return o.asIObservable();
	}
	
	action complexPipeOnNew(action<IObservable> returns IObservable complexPipe) returns IObservable {
		Observable o := _create();
		o.onConnection := ComplexPipeOnNew.create(complexPipe)(onConnection);
		return o.asIObservable();
	}
	
	action decouple() returns IObservable {
		string channelName := "DecoupleChannel" + integer.getUnique().toString();
		IDisposable d := observeToChannel(channelName);
		return observeFromChannel(channelName);
	}
	
	action distinctByField(string fieldName) returns IObservable {
		Observable o := _create();
		o.onConnection := DistinctByField.create(fieldName)(onConnection);
		return o.asIObservable();
	}
	
	action distinctByFieldUntilChanged(string fieldName) returns IObservable {
		Observable o := _create();
		o.onConnection := DistinctByFieldUntilChanged.create(fieldName)(onConnection);
		return o.asIObservable();
	}
	
	action pausable(IObservable pauser) returns IObservable {
		Observable o := _create();
		o.onConnection := Pausable.create(pauser)(onConnection);
		return o.asIObservable();
	}
	
	action pausableBuffered(IObservable trigger) returns IObservable {
		Observable o := _create();
		o.onConnection := PausableBuffered.create(trigger)(onConnection);
		return o.asIObservable();
	}
	
	action toChannel(string channelName) returns IDisposable {
		return ToChannel.create(self.asIObservable(), channelName);
	}
	
	action toSortedList(any comparator) returns IObservable {
		Observable o := _create();
		o.onConnection := ToSortedList.create(comparator)(onConnection);
		return o.asIObservable();
	}
	
	action toSortedListAsc() returns IObservable {
		Observable o := _create();
		o.onConnection := ToSortedListAsc.create()(onConnection);
		return o.asIObservable();
	}
	
	action toSortedListDesc() returns IObservable {
		Observable o := _create();
		o.onConnection := ToSortedListDesc.create()(onConnection);
		return o.asIObservable();
	}
	
	action sort(any comparator) returns IObservable {
		Observable o := _create();
		o.onConnection := Sort.create(comparator)(onConnection);
		return o.asIObservable();
	}
	
	action sortAsc() returns IObservable {
		Observable o := _create();
		o.onConnection := SortAsc.create()(onConnection);
		return o.asIObservable();
	}
	
	action sortDesc() returns IObservable {
		Observable o := _create();
		o.onConnection := SortDesc.create()(onConnection);
		return o.asIObservable();
	}
	
	static action invalidConnect() returns IDisposable {
		throw Exception("Can only connect a published Observable", "UnsupportedOperation");
	}
	
	action setOnConnection(action<IObserver> returns ISubscription onConnection) {
		self.onConnection := onConnection;
	}
	
	static action getSetOnConnection(IObservable o) returns action<action<IObserver> returns ISubscription> {
		return (<Observable>o.observable).setOnConnection;
	}
		
	action asIObservable() returns IObservable {
		return IObservable(self, 
		                   connectObserver, subscribe, map, publish, connect, refCount, share, take, takeLast, first, last, skip, skipLast, filter, scan, scanWithInitial, distinct, distinctBy, combineLatestFromInstance, withLatestFrom, delay, async, do, publishReplay, shareReplay, takeUntil, repeatObservable, flatMap, mergeFromInstance, zipFromInstance, switchOnNext, skipUntil, elementAt, reduce, reduceWithInitial, toStream, catchError, retry, observeOn, groupBy, observeToChannel, observeOnNew, every, contains, takeWhile, skipWhile, concatFromInstance, startWith, defaultIfEmpty, ignoreElements, count, sum, sumInteger, sumFloat, sumDecimal, concatString, max, maxInteger, maxFloat, maxDecimal, min, minInteger, minFloat, minDecimal, average, averageDecimal, sequenceEqualFromInstance, ambFromInstance, timestamp, updateTimestamp, timeout, getSync, getSyncOr, timeInterval, distinctUntilChanged, distinctByUntilChanged, switchMap, debounce, windowTime, throttleFirst, throttleLast, sampleTime, buffer, bufferTime, bufferCount, bufferTimeOrCount, combineLatestToSequenceFromInstance, withLatestFromToSequence, zipToSequenceFromInstance, groupByWindow, windowCount, windowTimeOrCount, sample, sampleCount, sampleTimeOrCount, bufferCountSkip, pipe, pluck, mergeAll, pipeOn, pipeOnNew, subscribeOn, subscribeOnNew, pairwise, let, groupByField, complexPipe, complexPipeOn, complexPipeOnNew, decouple, distinctByField, distinctByFieldUntilChanged, pausable, pausableBuffered, toChannel, toSortedList, toSortedListAsc, toSortedListDesc, sort, sortAsc, sortDesc);
	}
}