com.industry.rx_epl.IObservable <>

IObservable is the interface returned by almost all RxEPL operators, this allows the operators to be chained in with a fluent syntax.

Observable.fromValues([1,2,3,4)
	.map(Lambda.function1("x => x * 10"))
	.sum()
	.subscribe(Subscriber.create().onNext(logValue));

Operators and Methods

Transforms
Filters
Combiners
Error Handling
- CatchError
- Retry
Utils
Conditional
Math and Aggregation

Transforms

# .map(action<value: T1> returns T2) returns IObservable<T2> <>

Apply a function to each value and pass on the result.

action multiplyBy10(integer value) returns integer {
	return value * 10;
}

Observable.fromValues([1,2,3])
	.map(multiplyBy10)
	...
	
// Output: 10, 20, 30

Filters

# .filter(predicate: action<value: T> returns boolean) returns IObservable<T> <>

Filter the values by the provided predicate.

action greaterThan3(integer value) returns boolean {
	return value > 3;
}

Observable.fromValues([1,2,3,4,5,6])
	.filter(greaterThan3)
	...

// Output: 4, 5, 6

# .distinct() returns IObservable<T> <>

Remove all duplicate values.

Note: This requires storing every unique value and therefore care should be taken when used on long running observables. A safer alternative is DistinctUntilChanged

Observable.fromValues([1,2,1,2,3,2,3])
	.distinct()
	...

// Output: 1, 2, 3

Combiners

# .merge(other: sequence<IObservable<T>>) returns IObservable<T> <>

Merge the outputs of all of the provided other observables.

Observable.interval(0.1)
	.merge([Observable.interval(0.1)])
	...

// Output: 0,0,1,1,2,2,3,3...

# .mergeAll() returns IObservable<T> <>

Removes a layer of nesting from observables. Received values which are IObservables are merged into the output. Received values which are sequences/dictionaries have their values merged into the output.

Observable.fromValues([<any> Observable.interval(0.1), [1,2,3]])
	.mergeAll()
	...

// Output: 1,2,3,0,1,2,3,4...

# .withLatestFrom(other: sequence<IObservable<any>>, combiner: action<values: sequence<any>> returns any) returns IObservable<T> <>

Every time a value is received, take the latest values from the other observables and produce an output by running the combiner.

Note: The combiner takes the values in the same order as the observables are defined (starting with the main source observable).

action createSequenceString(sequence<any> values) returns any {
	sequence<string> strings := new sequence<string>;
	any value;
	for value in values {
		strings.append(value.valueToString());
	}
	return "[" + ",".join(strings) + "]";
}

Observable.interval(1.0)
	.withLatestFrom([Observable.interval(0.1)], createSequenceString)
	...

// Output: "[0,9]","[1,19]","[2,29]"...

# .withLatestFromToSequence(other: sequence<IObservable<any>>) returns IObservable<T> <>

Every time a value is received, take the latest values from the other observables and produce a sequence<any> containing the values from all.

Note: The resulting sequence contains the values in the same order as the observables are defined (starting with the main source observable).

Observable.interval(1.0)
	.withLatestFrom([Observable.interval(0.1)])
	...

// Output: [0,9], [1,19], [2,29]...

# .combineLatest(other: sequence<IObservable<any>>, combiner: action<values: sequence<any>> returns any) returns IObservable<T> <>

Every time a value is received, from either the source or the other observables, produce an output by running the combiner.

Note: The combiner takes the values in the same order as the observables are defined (starting with the main source observable).

action createSequenceString(sequence<any> values) returns any {
	sequence<string> strings := new sequence<string>;
	any value;
	for value in values {
		strings.append(value.valueToString());
	}
	return "[" + ",".join(strings) + "]";
}

Observable.interval(1.0)
	.combineLatest([Observable.interval(0.5)], createSequenceString)
	...

// Output: "[0,0]","[0,1]","[0,2]","[1,2]","[1,3]","[1,4]","[2,4]"...

# .combineLatestToSequence(other: sequence<IObservable<any>>) returns IObservable<T> <>

Every time a value is received, from either the source or the other observables, produce a sequence<any> containing the values from all.

Note: The resulting sequence contains the values in the same order as the observables are defined (starting with the main source observable).

Observable.interval(1.0)
	.combineLatestToSequence([Observable.interval(0.1)])
	...

// Output: [0,0],[0,1],[0,2],[1,2],[1,3],[1,4],[2,4]...

# .zip(other: sequence<IObservable<any>>, combiner: action<values: sequence<any>> returns any) returns IObservable<T> <>

Combine multiple observables by taking the n'th value from every observable, producing an output by running the combiner.

I.e. The first output is the result from the combiner running on the first value from every source.

Note: The combiner takes the values in the same order as the observables are defined (starting with the main source observable).

Note2: The result is terminated when any of the sources run out of values, but only after the output for those values has been generated.

Note3: This requires storing values until their counterpart in another observable is found, this could be expensive with long running observables.

action createSequenceString(sequence<any> values) returns any {
	sequence<string> strings := new sequence<string>;
	any value;
	for value in values {
		strings.append(value.valueToString());
	}
	return "[" + ",".join(strings) + "]";
}

Observable.interval(1.0)
	.zip([Observable.interval(0.5), Observable.fromValues(["a","b","c"])], createSequenceString)
	...

// Output: "[0,0,a]","[1,1,b]","[2,2,c]"

# .zipToSequence(other: sequence<IObservable<any>>) returns IObservable<T> <>

Combine multiple observables by taking the n'th value from every observable, producing a sequence<any> containing all of the n'th values.

I.e. The first output is a sequence<any> containing the first value from every source.

Note: The output sequence<any> contains the values in the same order as the observables are defined (starting with the main source observable).

Note2: The result is terminated when any of the sources run out of values, but only after the output for those values has been generated.

Note3: This requires storing values until their counterpart in another observable is found, this could be expensive with long running observables.

Observable.interval(1.0)
	.zipToSequence([Observable.interval(0.5), Observable.fromValues(["a","b","c"])])
	...

// Output: [0,0,a],[1,1,b],[2,2,c]

# .concat(other: sequence<IObservable<any>>) returns IObservable<T> <>

After the current observable completes, instead of completing, connect to the next source observable. This repeats until all sources have completed.

Note: This will potentially miss values if the other observables are "hot" (Miss values when not connected. Eg. Values from a channel or stream).

Observable.fromValues([1,2,3])
	.concat([Observable.fromValues([4,5,6])])
	...

// Output: 1,2,3,4,5,6

Error Handling

# .catchError(substitute: IObservable<T>) returns IObservable<T> <>

Switch to an alternative data source in the event of an error.

action someActionThatThrows(integer value) returns integer {
	throw com.apama.exceptions.Exception("Ahhhh", "RuntimeException");
}

Observable.fromValues([1,2,3,4])
	.map(someActionThatThrows)
	.catchError(Observable.fromValues(5,6,7,8))
	...

// Output: 5,6,7,8

# .retry(attempts: integer) returns IObservable<T> <>

Reconnect to the datasource in the event of an error. At most attempts times.

action sometimesThrows(integer value) returns integer {
	if (5).rand() = 0 {
		throw com.apama.exceptions.Exception("Ahhhh", "RuntimeException");
	}
	return value;
}

Observable.fromValues([1,2,3,4])
	.map(sometimesThrows)
	.retry(3)
	...

// Example Output: 1,1,2,3,1,2,3,4
// Example Output: 1,1,2,1,2,1, Exception("Ahhhh", "RuntimeException")

Utils

# .subscribe(Subscriber) returns ISubscription <>

Connect to the source observable, and register listeners for values, errors and completion.

The returned ISubscription can be used to manually terminate the subscription.

action logValue(any value) {
	log value.valueToString();
}

action logError(com.apama.exceptions.Exception e) {
	log e.toString();
}

action logComplete() {
	log "Done!";
}

Observable.fromValues([1,2,3])
	.subscribe(Subscriber.create().onNext(logValue).onError(logError).onComplete(logComplete));

// Output: 1,2,3,Done!

# .subscribeOn(Subscriber, context) returns ISubscription <>
# .subscribeOnNew(Subscriber) returns ISubscription <>

Connect to the source observable on a different context, and register listeners for values, errors and completion.

The returned ISubscription can be used to manually terminate the subscription.

Important Note: This will not recommended for complicated observables! Instead create the observable on a spawned context, or use ObserveOn.

action logValue(any value) {
	log value.valueToString();
}

action logComplete() {
	log "Done!";
}

Observable.fromValues([1,2,3])
	.subscribeOn(Subscriber.create().onNext(logValue).onComplete(logComplete), context("Context2"));

// Output (on Context2): 1,2,3,Done!

A better alternative:

action createAndRunObservable() {
	ISubscription s := Observable.fromValues([1,2,3])
		.subscribe(Subscriber.create().onNext(logValue).onComplete(logComplete))
}

spawn createAndRunObservable() to context("Context2");

# .do(Subscriber) returns IObservable<T> <>

Snoops the output of an observable (at the point where the do is added), registering listeners for values, errors and completion, without subscribing to the observable.

This is very useful for debugging.

action logValue(integer value) {
	log value.toString();
}

action logDone() {
	log "Done!";
}

Observable.fromValues([1,2,3,4])
	.do(Subscriber.create().onNext(logValue).onComplete(logDone))
	...

// Do Output: 1,2,3,4,Done!
// Output: 1,2,3,4

Conditional

# .contains(predicate: action<value: T> returns boolean) returns IObservable<boolean> <>

Check if at least one value matches the predicate.

Note: An empty observable returns false.

action greaterThan3(integer value) returns boolean {
	return value > 3;
}

Observable.fromValues([1,2,3,4])
	.contains(greaterThan3)
	...

// Output: true

# .every(predicate: action<value: T> returns boolean) returns IObservable<boolean> <>

Check if every value matches the predicate.

Note: An empty observable returns true.

action lessThan3(integer value) returns boolean {
	return value < 3;
}

Observable.fromValues([1,2,3,4])
	.every(lessThan3)
	...

// Output: false

# .sequenceEqual(others: sequence<IObservable<T>>) returns IObservable<boolean> <>

Check if all of the provided observables contain the same values, in the same order.

Observable.interval(1.0).take(4)
	.sequenceEqual([Observable.fromValues([0,1,2,3]), Observable.range(0,3)])
	...

// Output: true

# .isEmpty() returns IObservable<boolean> <>

Check if the observable completes without emitting any values.

Note: It will return false immediately if any values are received.

Observable.fromValues([1,2,3])
	.isEmpty()
	...

// Output: false

Observable.empty()
	.isEmpty()
	...

// Output: true

# .amb(others: sequence<IObservable<T>>) returns IObservable<T> <>
# .race(others: sequence<IObservable<T>>) returns IObservable<T> <>

Race the observables, the one which provides values first provides all of the values.

Note: Amb and Race are aliases.

Observable.timer("I lost", 5.0)
	.amb([Observable.timer("I won!", 1.0)])
	...

// Output: "I won!"

# .defaultIfEmpty(value: T) returns IObservable<T> <>

Emit a default value if the source completes without emitting any values.

Observable.empty()
	.defaultIfEmpty("A default value")
	...

// Output: "A default value"

# .switchIfEmpty(alternative: IObservable<T>) returns IObservable<T> <>

Switch to a different observable source if the main source completes without emitting any values.

Observable.empty()
	.switchIfEmpty(Observable.fromValues([1,2,3]))
	...

// Output: 1,2,3

Math and Aggregation

# .reduce(action<aggregate: T2, value: T1> returns T2) returns IObservable<T2> <>

Aggregate the data, emitting the aggregated value on completion.

Note: The first value is used as the starting value for the aggregation.

action sum(integer currentSum, integer value) returns integer {
	return currentSum + value;
}

Observable.fromValues([1,2,3])
	.reduce(sum)
	...

// Output: 6

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

IObservable.md

IObservable.md

com.industry.rx_epl.IObservable <>

Operators and Methods

Transforms

Filters

Combiners

Error Handling

Utils

Conditional

Math and Aggregation

Files

IObservable.md

Latest commit

History

IObservable.md

File metadata and controls

com.industry.rx_epl.IObservable <>

Operators and Methods

Transforms

Filters

Combiners

Error Handling

Utils

Conditional

Math and Aggregation