OpenServer/Utils
1
0
Fork 0
Code Issues Pull Requests 1 Releases Wiki Activity

Remove unnecessary timer from obeservable

Browse Source
This commit is contained in:
Fabian Stamm 2020-04-13 03:11:12 +02:00
parent 2545f3a050
commit 858a508190
7 changed files with 106 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
.editorconfig Normal file
View File

@ -0,0 +1,6 @@
root=true
[*]
charset = utf-8
indent_size = 3
indent_style = space
insert_final_newline = true

2
package-lock.json generated
View File

@ -1,6 +1,6 @@
{ {
"name": "@hibas123/utils", "name": "@hibas123/utils",
"version": "2.2.2", "version": "2.2.3",
"lockfileVersion": 1, "lockfileVersion": 1,
"requires": true, "requires": true,
"dependencies": { "dependencies": {

54
package.json
View File

@ -1,28 +1,28 @@
{ {
"name": "@hibas123/utils", "name": "@hibas123/utils",
"version": "2.2.3", "version": "2.2.4",
"description": "Different Utilities, that are not worth own packages", "description": "Different Utilities, that are not worth own packages",
"main": "lib/index.js", "main": "lib/index.js",
"types": "lib/index.d.ts", "types": "lib/index.d.ts",
"scripts": { "scripts": {
"prepublishOnly": "tsc", "prepublishOnly": "tsc",
"build": "tsc", "build": "tsc",
"watch-ts": "tsc -w" "watch-ts": "tsc -w"
}, },
"author": "Fabian Stamm <dev@fabianstamm.de>", "author": "Fabian Stamm <dev@fabianstamm.de>",
"license": "MIT", "license": "MIT",
"repository": { "repository": {
"url": "https://git.stamm.me/OpenServer/Utils.git", "url": "https://git.stamm.me/OpenServer/Utils.git",
"type": "git" "type": "git"
}, },
"devDependencies": { "devDependencies": {
"typescript": "^3.4.5" "typescript": "^3.4.5"
}, },
"files": [ "files": [
"src/", "src/",
"lib/", "lib/",
"tsconfig.json", "tsconfig.json",
"readme.md" "readme.md"
], ],
"private": false "private": false
} }

42
src/awaiter.ts
View File

@ -41,24 +41,28 @@ export default class AwaitStore<T = any> {
} }
/** /**
* Await a specific value and return. * Await a specific value and return.
* *
* For example if val = true then this function would block until the value * For example if val = true then this function would block until the value
* is actually true. If it is true, then the promise will resolve immediatly * is actually true. If it is true, then the promise will resolve immediatly
* *
* @param val Value to await * @param val Value to await
*/ */
awaitValue(val: T): PromiseLike<void> & { catch: (cb: (err: any) => PromiseLike<void>) => PromiseLike<void>, ignore: () => void } { awaitValue(
val: T
): PromiseLike<void> & {
catch: (cb: (err: any) => PromiseLike<void>) => PromiseLike<void>;
ignore: () => void;
} {
let ignore: () => void = () => undefined; let ignore: () => void = () => undefined;
let prms = new Promise<void>(yes => { let prms = new Promise<void>((yes) => {
const cb = () => { const cb = () => {
if (this._value === val) { if (this._value === val) {
yes(); yes();
this.unsubscribe(cb); this.unsubscribe(cb);
} }
} };
this.subscribe(cb); this.subscribe(cb);
}); });
@ -66,23 +70,23 @@ export default class AwaitStore<T = any> {
return { return {
then: prms.then.bind(prms), then: prms.then.bind(prms),
catch: prms.catch.bind(prms), catch: prms.catch.bind(prms),
ignore: () => ignore() ignore: () => ignore(),
} };
} }
/** /**
* Creates Public API with subscribe and unsubscribe * Creates Public API with subscribe and unsubscribe
* *
* @returns {object} * @returns {object}
*/ */
getPublicApi() { getPublicApi() {
if (this.observable.closed) if (this.observable.closed) throw new Error("Observable is closed!");
throw new Error("Observable is closed!");
return { return {
subscribe: (callback: ObserverCallback<T>) => this.subscribe(callback), subscribe: (callback: ObserverCallback<T>) => this.subscribe(callback),
unsubscribe: (callback: ObserverCallback<T>) => this.unsubscribe(callback), unsubscribe: (callback: ObserverCallback<T>) =>
awaitValue: (value: T) => this.awaitValue(value) this.unsubscribe(callback),
} awaitValue: (value: T) => this.awaitValue(value),
};
} }
/** /**
@ -91,4 +95,4 @@ export default class AwaitStore<T = any> {
close() { close() {
this.observable.close(); this.observable.close();
} }
} }

22
src/index.ts
View File

@ -1,13 +1,17 @@
import Lock, { Release } from "./lock"; import Lock, { Release } from "./lock";
import Observable, { ObserverCallback, ObserverCallbackCollect, ObservableInterface } from "./observable"; import Observable, {
ObserverCallback,
ObserverCallbackCollect,
ObservableInterface,
} from "./observable";
import AwaitStore from "./awaiter"; import AwaitStore from "./awaiter";
export { export {
Lock, Lock,
Release, Release,
Observable, Observable,
ObserverCallback, ObserverCallback,
ObserverCallbackCollect, ObserverCallbackCollect,
ObservableInterface, ObservableInterface,
AwaitStore AwaitStore,
} };

10
src/lock.ts
View File

@ -2,14 +2,14 @@ export type Release = { release: () => void };
/** /**
* Basic Locking mechanism for JavaScript * Basic Locking mechanism for JavaScript
* *
*/ */
export default class Lock { export default class Lock {
private _locked: boolean = false; private _locked: boolean = false;
/** /**
* Returns the state of the Locken * Returns the state of the Locken
* *
* @returns {boolean} * @returns {boolean}
*/ */
get locked(): boolean { get locked(): boolean {
@ -23,7 +23,7 @@ export default class Lock {
/** /**
* Waits till lock is free and returns a release function * Waits till lock is free and returns a release function
* *
* @return {function} * @return {function}
*/ */
async getLock(): Promise<Release> { async getLock(): Promise<Release> {
@ -32,8 +32,8 @@ export default class Lock {
return new Promise<Release>((resolve) => { return new Promise<Release>((resolve) => {
this.toCome.push(() => { this.toCome.push(() => {
resolve({ release: this.lock() }); resolve({ release: this.lock() });
}) });
}) });
} }
} }

68
src/observable.ts
View File

@ -8,9 +8,9 @@ export type ObservableInterface<T> = {
*/ */
subscribe: (callback: ObserverCallback<T>) => void; subscribe: (callback: ObserverCallback<T>) => void;
/** /**
* Unsubscribe fron Observable * Unsubscribe fron Observable
* @param {function} callback * @param {function} callback
*/ */
unsubscribe: (callback: ObserverCallback<T>) => void; unsubscribe: (callback: ObserverCallback<T>) => void;
/** /**
* Subscribe to Observable in collect mode * Subscribe to Observable in collect mode
@ -34,55 +34,51 @@ export default class Observable<T = any> {
return this[ClosedSymbol]; return this[ClosedSymbol];
} }
constructor(private collect_intervall: number = 100) { } constructor(private collect_intervall: number = 100) {}
subscribe(callback: ObserverCallback<T>) { subscribe(callback: ObserverCallback<T>) {
if (this[ClosedSymbol]) if (this[ClosedSymbol]) throw new Error("Observable is closed!");
throw new Error("Observable is closed!");
let oldcb = this.subscriber.find(e => e === callback); let oldcb = this.subscriber.find((e) => e === callback);
if (!oldcb) if (!oldcb) this.subscriber.push(callback);
this.subscriber.push(callback)
return () => this.unsubscribe(callback); return () => this.unsubscribe(callback);
} }
unsubscribe(callback: ObserverCallback<T> | ObserverCallbackCollect<T>) { unsubscribe(callback: ObserverCallback<T> | ObserverCallbackCollect<T>) {
if (this[ClosedSymbol]) if (this[ClosedSymbol]) return;
return;
let idx = this.subscriber.findIndex(e => e === callback); let idx = this.subscriber.findIndex((e) => e === callback);
if (idx >= 0) { if (idx >= 0) {
this.subscriber.splice(idx, 1); this.subscriber.splice(idx, 1);
} else { } else {
idx = this.subscriberCollect.findIndex(e => e === callback); idx = this.subscriberCollect.findIndex((e) => e === callback);
if (idx >= 0) if (idx >= 0) this.subscriberCollect.splice(idx, 1);
this.subscriberCollect.splice(idx, 1);
} }
} }
subscribeCollect(callback: ObserverCallbackCollect<T>) { subscribeCollect(callback: ObserverCallbackCollect<T>) {
if (this[ClosedSymbol]) if (this[ClosedSymbol]) throw new Error("Observable is closed!");
throw new Error("Observable is closed!");
let oldcb = this.subscriberCollect.find(e => e === callback); let oldcb = this.subscriberCollect.find((e) => e === callback);
if (!oldcb) if (!oldcb) this.subscriberCollect.push(callback);
this.subscriberCollect.push(callback)
} }
/** /**
* Creates Public API with subscribe and unsubscribe * Creates Public API with subscribe and unsubscribe
* *
* @returns {object} * @returns {object}
*/ */
getPublicApi(): ObservableInterface<T> { getPublicApi(): ObservableInterface<T> {
if (this[ClosedSymbol]) if (this[ClosedSymbol]) throw new Error("Observable is closed!");
throw new Error("Observable is closed!");
return { return {
subscribe: (callback: ObserverCallback<T>) => this.subscribe(callback), subscribe: (callback: ObserverCallback<T>) => this.subscribe(callback),
unsubscribe: (callback: ObserverCallback<T> | ObserverCallbackCollect<T>) => this.unsubscribe(callback), unsubscribe: (
subscribeCollect: (callback: ObserverCallbackCollect<T>) => this.subscribeCollect(callback) callback: ObserverCallback<T> | ObserverCallbackCollect<T>
} ) => this.unsubscribe(callback),
subscribeCollect: (callback: ObserverCallbackCollect<T>) =>
this.subscribeCollect(callback),
};
} }
/** /**
@ -90,22 +86,21 @@ export default class Observable<T = any> {
* @param data data to be sent * @param data data to be sent
*/ */
send(data: T) { send(data: T) {
if (this[ClosedSymbol]) if (this[ClosedSymbol]) throw new Error("Observable is closed!");
throw new Error("Observable is closed!")
Array.from(this.subscriber.values()).forEach(e => { Array.from(this.subscriber.values()).forEach((e) => {
try { try {
e(data) e(data);
} catch (err) { } catch (err) {
// Catch error, so it doesn't affect other subscribers // Catch error, so it doesn't affect other subscribers
console.error(err) console.error(err);
} }
}); });
this.events.push(data); this.events.push(data);
if (!this.timeout) { if (!this.timeout && this.subscriberCollect.length > 0) {
this.timeout = setTimeout(() => { this.timeout = setTimeout(() => {
this.subscriberCollect.forEach(cb => { this.subscriberCollect.forEach((cb) => {
cb(this.events) cb(this.events);
}); });
this.events = []; this.events = [];
this.timeout = undefined; this.timeout = undefined;
@ -122,7 +117,6 @@ export default class Observable<T = any> {
this.subscriber = []; this.subscriber = [];
this.subscriberCollect = []; this.subscriberCollect = [];
this.events = []; this.events = [];
if (this.timeout) if (this.timeout) clearTimeout(this.timeout);
clearTimeout(this.timeout)
} }
} }