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