smartrx/ts/smartrx.classes.observableintake.ts

106 lines
2.4 KiB
TypeScript
Raw Normal View History

import * as plugins from './smartrx.plugins';
import { Observable, Subscription } from 'rxjs';
import { Deferred } from 'smartq';
2017-11-01 13:59:28 +00:00
/**
* ObservableIntake
*/
export class ObservableIntake<T> {
observable: Observable<T>;
completed: Promise<void>;
private completedDeffered: Deferred<void>;
2017-11-01 13:59:28 +00:00
private observableFunctions: any = {
next: payloadArg => {
2017-11-01 13:59:28 +00:00
// nothing
},
complete: payloadArg => {
2017-11-01 13:59:28 +00:00
// nothing
}
};
private generator = null;
private buffered = false;
private payloadBuffer = [];
2017-11-01 13:59:28 +00:00
constructor() {
this.observable = Observable.create(observerArg => {
2017-11-01 13:59:28 +00:00
this.observableFunctions.next = (...args) => {
return observerArg.next(...args);
};
2017-11-01 13:59:28 +00:00
this.observableFunctions.complete = (...args) => {
this.completedDeffered.resolve();
return observerArg.complete(...args);
};
});
this.completedDeffered = plugins.smartpromise.defer();
this.completed = this.completedDeffered.promise;
2017-11-01 13:59:28 +00:00
}
setObservable(observableFunc) {
this.observable = observableFunc();
2017-11-01 13:59:28 +00:00
}
push(payloadArg: T) {
2017-11-01 16:01:30 +00:00
if (this.buffered) {
this.payloadBuffer.push(payloadArg);
2017-11-01 16:01:30 +00:00
} else {
this.internalPush(payloadArg);
2017-11-01 16:01:30 +00:00
}
}
/**
* pushes many payloads as array
* @param payloadArgArray
*/
pushMany(payloadArgArray: T[]) {
2017-11-01 16:01:30 +00:00
for (let item of payloadArgArray) {
this.push(item);
2017-11-01 16:01:30 +00:00
}
}
/**
* sets a generator to query the next pushed value
* @param generatorArg
*/
setGenerator(generatorArg) {
this.generator = generatorArg;
2017-11-01 16:01:30 +00:00
}
makeBuffered() {
this.buffered = true;
2017-11-01 13:59:28 +00:00
}
subscribe(...args) {
return this.observable.subscribe(...args);
2017-11-01 13:59:28 +00:00
}
2017-11-01 16:01:30 +00:00
/**
* request the next values in the quantity specified
* @param howManyArg if a generator is set, of a buffer exists, this allows retrieving values
*/
request(howManyArg: number) {
2017-11-01 16:01:30 +00:00
if (howManyArg === 0) {
return;
2017-11-01 16:01:30 +00:00
} else {
for (let i = 0; i !== howManyArg; i++) {
if (this.payloadBuffer.length > 0) {
this.internalPush(this.payloadBuffer.shift());
2017-11-01 16:01:30 +00:00
} else {
const nextPayload = this.generator();
this.internalPush(nextPayload);
2017-11-01 16:01:30 +00:00
}
}
}
}
/**
* signals the completion of this observable
*/
signalComplete() {
this.observableFunctions.complete();
2017-11-01 13:59:28 +00:00
}
2017-11-01 16:01:30 +00:00
private internalPush(payloadArg) {
this.observableFunctions.next(payloadArg);
2017-11-01 16:01:30 +00:00
}
2017-11-01 13:59:28 +00:00
}