smartrx/ts/smartrx.classes.observableintake.ts

104 lines
2.4 KiB
TypeScript
Raw Normal View History

import * as plugins from './smartrx.plugins';
2017-11-01 13:59:28 +00:00
/**
* ObservableIntake
*/
export class ObservableIntake<T> {
2020-05-26 00:18:02 +00:00
public observable: plugins.rxjs.Observable<T>;
2019-09-10 05:25:59 +00:00
public completed: Promise<void>;
2020-05-26 00:18:02 +00:00
private completedDeffered: plugins.smartpromise.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() {
2020-05-26 00:18:02 +00:00
this.observable = plugins.rxjs.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
}
2019-09-10 05:25:59 +00:00
public setObservable(observableFunc) {
this.observable = observableFunc();
2017-11-01 13:59:28 +00:00
}
2019-09-10 05:25:59 +00:00
public 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
*/
2019-09-10 05:25:59 +00:00
public pushMany(payloadArgArray: T[]) {
for (const 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
*/
2019-09-10 05:25:59 +00:00
public setGenerator(generatorArg) {
this.generator = generatorArg;
2017-11-01 16:01:30 +00:00
}
2019-09-10 05:25:59 +00:00
public makeBuffered() {
this.buffered = true;
2017-11-01 13:59:28 +00:00
}
2019-09-10 05:25:59 +00:00
public 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
*/
2019-09-10 05:25:59 +00:00
public 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
*/
2019-09-10 05:25:59 +00:00
public 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
}