120 lines
3.6 KiB
TypeScript
Executable File
120 lines
3.6 KiB
TypeScript
Executable File
export type ObserverCallbackCollect<T> = (data: T[]) => void;
|
|
export type ObserverCallback<T> = (data: T) => void;
|
|
|
|
export type ObservableInterface<T> = {
|
|
/**
|
|
* Subscribe to Observable
|
|
* @param {function} callback
|
|
*/
|
|
subscribe: (callback: ObserverCallback<T>) => void;
|
|
/**
|
|
* Unsubscribe fron Observable
|
|
* @param {function} callback
|
|
*/
|
|
unsubscribe: (callback: ObserverCallback<T>) => void;
|
|
/**
|
|
* Subscribe to Observable in collect mode
|
|
* @param {function} callback
|
|
*/
|
|
subscribeCollect: (callback: ObserverCallbackCollect<T>) => void;
|
|
};
|
|
|
|
const ClosedSymbol = Symbol("Observable Closed");
|
|
|
|
export default class Observable<T = any> {
|
|
private subscriber: ObserverCallback<T>[] = [];
|
|
private subscriberCollect: ObserverCallbackCollect<T>[] = [];
|
|
private events: T[] = [];
|
|
private timeout: number | undefined = undefined;
|
|
|
|
// Use symbol to make sure this property cannot be changed from the outside
|
|
private [ClosedSymbol] = false;
|
|
|
|
get closed() {
|
|
return this[ClosedSymbol];
|
|
}
|
|
|
|
constructor(private collect_intervall: number = 100) { }
|
|
|
|
subscribe(callback: ObserverCallback<T>) {
|
|
if (this[ClosedSymbol])
|
|
throw new Error("Observable is closed!");
|
|
|
|
let oldcb = this.subscriber.find(e => e === callback);
|
|
if (!oldcb)
|
|
this.subscriber.push(callback)
|
|
|
|
return () => this.unsubscribe(callback);
|
|
}
|
|
|
|
unsubscribe(callback: ObserverCallback<T> | ObserverCallbackCollect<T>) {
|
|
if (this[ClosedSymbol])
|
|
return;
|
|
|
|
let idx = this.subscriber.findIndex(e => e === callback);
|
|
if (idx >= 0) {
|
|
this.subscriber.splice(idx, 1);
|
|
} else {
|
|
idx = this.subscriberCollect.findIndex(e => e === callback);
|
|
if (idx >= 0)
|
|
this.subscriberCollect.splice(idx, 1);
|
|
}
|
|
}
|
|
|
|
subscribeCollect(callback: ObserverCallbackCollect<T>) {
|
|
if (this[ClosedSymbol])
|
|
throw new Error("Observable is closed!");
|
|
|
|
let oldcb = this.subscriberCollect.find(e => e === callback);
|
|
if (!oldcb)
|
|
this.subscriberCollect.push(callback)
|
|
}
|
|
|
|
/**
|
|
* Creates Public API with subscribe and unsubscribe
|
|
*
|
|
* @returns {object}
|
|
*/
|
|
getPublicApi(): ObservableInterface<T> {
|
|
if (this[ClosedSymbol])
|
|
throw new Error("Observable is closed!");
|
|
return {
|
|
subscribe: (callback: ObserverCallback<T>) => this.subscribe(callback),
|
|
unsubscribe: (callback: ObserverCallback<T> | ObserverCallbackCollect<T>) => this.unsubscribe(callback),
|
|
subscribeCollect: (callback: ObserverCallbackCollect<T>) => this.subscribeCollect(callback)
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Sends data to all subscribers
|
|
* @param data data to be sent
|
|
*/
|
|
send(data: T) {
|
|
if (this[ClosedSymbol])
|
|
throw new Error("Observable is closed!")
|
|
this.subscriber.forEach(e => e(data));
|
|
this.events.push(data);
|
|
if (!this.timeout) {
|
|
this.timeout = setTimeout(() => {
|
|
this.subscriberCollect.forEach(cb => {
|
|
cb(this.events)
|
|
});
|
|
this.events = [];
|
|
this.timeout = undefined;
|
|
}, this.collect_intervall);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Closes Observable. This will remove all subscribers and mark this observable as closed.
|
|
* You won't be able to reopen this observable. All maybe collected data will be discardet.
|
|
*/
|
|
close() {
|
|
this[ClosedSymbol] = true;
|
|
this.subscriber = [];
|
|
this.subscriberCollect = [];
|
|
this.events = [];
|
|
if (this.timeout)
|
|
clearTimeout(this.timeout)
|
|
}
|
|
} |