Compare commits

..

1 Commits

Author SHA1 Message Date
Fabian Stamm
564166a781 Adding support for reading the last value. 2019-08-22 13:54:56 +02:00
15 changed files with 120 additions and 404 deletions

View File

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

1
.gitignore vendored
View File

@ -1,3 +1,2 @@
node_modules/
lib/
esm/

View File

@ -1,17 +0,0 @@
{
"name": "utils",
"version": "2.2.17",
"description": "Some helpful utility classes and functions",
"author": "Fabian Stamm <dev@fabianstamm.de>",
"contributors": [],
"root": "./esm/",
"files": [
"**/*.ts",
"**/*.js",
"esm/readme.md"
],
"hooks": {
"prepublish": "pre.js"
},
"readme": "readme.md"
}

8
package-lock.json generated
View File

@ -1,13 +1,13 @@
{
"name": "@hibas123/utils",
"version": "2.2.17",
"version": "2.1.1",
"lockfileVersion": 1,
"requires": true,
"dependencies": {
"typescript": {
"version": "4.0.5",
"resolved": "https://npm.hibas123.de/typescript/-/typescript-4.0.5.tgz",
"integrity": "sha512-ywmr/VrTVCmNTJ6iV2LwIrfG1P+lv6luD8sUJs+2eI9NLGigaN+nUQc13iHqisq7bra9lnmUSYqbJvegraBOPQ==",
"version": "3.5.3",
"resolved": "https://registry.npmjs.org/typescript/-/typescript-3.5.3.tgz",
"integrity": "sha512-ACzBtm/PhXBDId6a6sDJfroT2pOWt/oOnk4/dElG5G33ZL776N3Y6/6bKZJBFpd+b05F3Ct9qDjMeJmRWtE2/g==",
"dev": true
}
}

View File

@ -1,16 +1,13 @@
{
"name": "@hibas123/utils",
"version": "3.0.0",
"version": "2.1.1",
"description": "Different Utilities, that are not worth own packages",
"type": "module",
"main": "esm/index.js",
"types": "esm/index.d.ts",
"module": "esm/index.js",
"main": "lib/index.js",
"types": "lib/index.d.ts",
"scripts": {
"prepublishOnly": "npm run build",
"prepublishOnly": "tsc",
"build": "tsc",
"watch-ts": "tsc -w",
"postpublish": "denreg publish"
"watch-ts": "tsc -w"
},
"author": "Fabian Stamm <dev@fabianstamm.de>",
"license": "MIT",
@ -19,11 +16,11 @@
"type": "git"
},
"devDependencies": {
"typescript": "^4.0.5"
"typescript": "^3.5.3"
},
"files": [
"src/",
"esm/",
"lib/",
"tsconfig.json",
"readme.md"
],

15
pre.js
View File

@ -1,15 +0,0 @@
const rjson = async (file) => JSON.parse(await Deno.readTextFile(file));
const wjson = (file, data) =>
Deno.writeTextFile(file, JSON.stringify(data, undefined, 3));
const pkg = await rjson("package.json");
const meta = await rjson("meta.json");
console.log("Changing meta.version from", meta.version, "to", pkg.version);
meta.version = pkg.version || meta.version;
await wjson("meta.json", meta);
await Deno.copyFile("readme.md", "esm/readme.md");
await Deno.copyFile("esm/index.js", "esm/mod.js");
await Deno.copyFile("esm/index.d.ts", "esm/mod.d.ts");

View File

@ -35,8 +35,8 @@ Usage:
const server = new Observable(); // Get new Observable Server
// Server can send and subscribe to messages
// The publicAPI will only make the receiving parts available
// Server can only send, not subscribe to messages
// Receiving is only possible via the public API
const public = server.getPublicApi();
const func = (data)=>{
@ -44,15 +44,10 @@ Usage:
}
// func will be callen when a message is available
let unsubscribe = public.subscribe(func);
public.subscribe(func);
server.send("Hello World");
// Unsubscribe using the returned function from subscribe
unsubscribe();
// OR
// This will unsubscribe the function. Please note, that it can
// only unsubscribe the exact function, that is used in subscribe
public.unsubscribe(func);
@ -61,44 +56,6 @@ Usage:
server.send("Hello World2");
```
## AwaitStore
This component can be used to await a specific value or to act as an variable with events.
Usage:
``` typescript
import { AwaitStore } from "@hibas123/util";
const server = new AwaitStore<number>(0); // Set initial value
// Server can send and subscribe to messages
// Only receiving is possible using the public API
const public = server.getPublicApi();
const func = (data)=>{
console.log(data);
}
// awaitValue will also return a .ignore() function, that can be used to abort the promise completely
public.awaitValue(5).then(()=>console.log("Got 5"));
// func will be callen when a message is available and once with the current value
// This will call func with the current value (0) and on every change after.
public.subscribe(func);
// This send will trigger the subscribtion func and also release the awaitValue causing "Got 5"
// to appear in the console
server.send(5);
// This will unsubscribe the function. Please note, that it can
// only unsubscribe the exact function, that is used in subscribe
public.unsubscribe(func);
// This now won't call func anymore
server.send(8);
```
## License
MIT

View File

@ -1,77 +0,0 @@
import Observable from "./observable.js";
import Signal from "./signal.js";
interface IAsyncIteratorMessage<T> {
error: Error | undefined;
data: T | undefined;
close: true | undefined;
}
export default class AsyncIteratorFromCB<T> implements AsyncIterable<T> {
#onClose: (() => void)[] = [];
#onData = new Observable<IAsyncIteratorMessage<T>>();
constructor() {}
public onClose(callback: () => void) {
this.#onClose.push(callback);
}
public getCallback() {
return (error: Error | undefined, data: T | undefined) => {
this.#onData.send({ error, data, close: undefined });
};
}
public send(data: T) {
this.#onData.send({
error: undefined,
data,
close: undefined,
});
}
public close() {
this.#onData.send({
close: true,
data: undefined,
error: undefined,
});
this.#onData.close();
this.#onClose.forEach((cb) => cb());
}
[Symbol.asyncIterator](): AsyncIterator<T, undefined, any> {
const queue: IAsyncIteratorMessage<T>[] = [];
const signal = new Signal();
this.#onData.subscribe((data) => {
queue.push(data);
signal.sendSignal();
});
return {
async next() {
const send = () => {
const value = queue.shift();
if (!value) throw new Error("Error in AsyncIter");
if (value.close) {
return {
done: true,
value: undefined as any,
};
} else if (value.error) {
throw value.error;
} else {
return {
done: false,
value: value.data as T,
};
}
};
if (queue.length > 0) {
return send();
} else {
await signal.awaitSignal();
return send();
}
},
};
}
}

View File

@ -1,105 +0,0 @@
import Observable, { ObserverCallback } from "./observable.js";
export type CheckFunction<T> = (val: T) => boolean;
export default class AwaitStore<T = any> {
private observable = new Observable<T>();
constructor(private _value: T) {
this.subscribe = this.subscribe.bind(this);
this.unsubscribe = this.unsubscribe.bind(this);
}
/**
* Get the current value
*/
get value() {
return this._value;
}
/**
* Set a new value and notify subscribers
* @param value Value to be set
*/
send(value: T) {
this._value = value;
this.observable.send(value);
}
/**
* Get the current value as well as all changes
* @param handler Handler called on change
*/
subscribe(handler: ObserverCallback<T>) {
handler(this._value);
return this.observable.subscribe(handler);
}
/**
* Unsubscribe from changes
* @param handler The handler to unsubscribe
*/
unsubscribe(handler: ObserverCallback<T>) {
this.observable.unsubscribe(handler);
}
/**
* Await a specific value and return.
*
* 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
*
* @param val Value to await
*/
awaitValue(
val: T | CheckFunction<T>
): PromiseLike<void> & {
catch: (cb: (err: any) => PromiseLike<void>) => PromiseLike<void>;
ignore: () => void;
} {
let ignore: () => void = () => undefined;
let prms = new Promise<void>((yes) => {
const cb = () => {
if (typeof val === "function") {
if ((val as CheckFunction<T>)(this._value)) {
yes();
this.unsubscribe(cb);
}
} else if (this._value === val) {
yes();
this.unsubscribe(cb);
}
};
this.subscribe(cb);
});
return {
then: prms.then.bind(prms),
catch: prms.catch.bind(prms),
ignore: () => ignore(),
};
}
/**
* Creates Public API with subscribe and unsubscribe
*
* @returns {object}
*/
getPublicApi() {
if (this.observable.closed) throw new Error("Observable is closed!");
return {
subscribe: (callback: ObserverCallback<T>) => this.subscribe(callback),
unsubscribe: (callback: ObserverCallback<T>) =>
this.unsubscribe(callback),
awaitValue: (value: T) => this.awaitValue(value),
};
}
/**
* Close this store. All subscribers will be unsubscribed and any further operations will fail
*/
close() {
this.observable.close();
}
}

View File

@ -1,22 +1,11 @@
import Lock, { Release } from "./lock.js";
import Lock, { Release } from "./lock";
import Observable, { ObserverCallback, ObserverCallbackCollect, ObservableInterface } from "./observable";
import Observable, {
ObserverCallback,
ObserverCallbackCollect,
ObservableInterface,
} from "./observable.js";
import AwaitStore from "./awaiter.js";
import AsyncIteratorFromCB from "./asynciter.js";
import Signal from "./signal.js";
export type {
export {
Lock,
Release,
Observable,
ObserverCallback,
ObserverCallbackCollect,
ObservableInterface,
};
export { Lock, Observable, AwaitStore, AsyncIteratorFromCB, Signal };
ObservableInterface
}

View File

@ -32,8 +32,8 @@ export default class Lock {
return new Promise<Release>((resolve) => {
this.toCome.push(() => {
resolve({ release: this.lock() });
});
});
})
})
}
}

View File

@ -20,6 +20,7 @@ export type ObservableInterface<T> = {
};
const ClosedSymbol = Symbol("Observable Closed");
const LastValueSymbol = Symbol("Observable LastValue");
export default class Observable<T = any> {
private subscriber: ObserverCallback<T>[] = [];
@ -30,40 +31,60 @@ export default class Observable<T = any> {
// Use symbol to make sure this property cannot be changed from the outside
private [ClosedSymbol] = false;
private [LastValueSymbol]: T | undefined = undefined;
get lastValue() {
return this[LastValueSymbol];
}
get closed() {
return this[ClosedSymbol];
}
constructor(private collect_intervall: number = 100) { }
/**
* Subscribe to changes
* @param callback Callback called once a value is available
*/
subscribe(callback: ObserverCallback<T>) {
if (this[ClosedSymbol]) throw new Error("Observable is closed!");
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);
let oldcb = this.subscriber.find(e => e === callback);
if (!oldcb)
this.subscriber.push(callback)
}
/**
* Unsubscribe specified callback. After that, it will be discarded.
* @param callback The callback originally subscribed
*/
unsubscribe(callback: ObserverCallback<T> | ObserverCallbackCollect<T>) {
if (this[ClosedSymbol]) return;
if (this[ClosedSymbol])
return;
let idx = this.subscriber.findIndex((e) => e === callback);
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);
idx = this.subscriberCollect.findIndex(e => e === callback);
if (idx >= 0)
this.subscriberCollect.splice(idx, 1);
}
}
/**
* Subscribe for a collection of changes
* @param callback Callback called once a or some values are available.
*/
subscribeCollect(callback: ObserverCallbackCollect<T>) {
if (this[ClosedSymbol]) throw new Error("Observable is closed!");
if (this[ClosedSymbol])
throw new Error("Observable is closed!");
let oldcb = this.subscriberCollect.find((e) => e === callback);
if (!oldcb) this.subscriberCollect.push(callback);
return () => this.unsubscribe(callback);
let oldcb = this.subscriberCollect.find(e => e === callback);
if (!oldcb)
this.subscriberCollect.push(callback)
}
/**
@ -72,15 +93,13 @@ export default class Observable<T = any> {
* @returns {object}
*/
getPublicApi(): ObservableInterface<T> {
if (this[ClosedSymbol]) throw new Error("Observable is closed!");
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),
};
unsubscribe: (callback: ObserverCallback<T> | ObserverCallbackCollect<T>) => this.unsubscribe(callback),
subscribeCollect: (callback: ObserverCallbackCollect<T>) => this.subscribeCollect(callback)
}
}
/**
@ -88,29 +107,22 @@ export default class Observable<T = any> {
* @param data data to be sent
*/
send(data: T) {
if (this[ClosedSymbol]) throw new Error("Observable is closed!");
if (this[ClosedSymbol])
throw new Error("Observable is closed!");
Array.from(this.subscriber.values()).forEach((e) => {
try {
e(data);
} catch (err) {
// Catch error, so it doesn't affect other subscribers
console.error(err);
}
});
if (this.subscribeCollect.length > 0) {
this[LastValueSymbol] = data;
this.subscriber.forEach(e => e(data));
this.events.push(data);
if (!this.timeout) {
this.timeout = setTimeout(() => {
this.subscriberCollect.forEach((cb) => {
cb(this.events);
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.
@ -121,6 +133,7 @@ export default class Observable<T = any> {
this.subscriber = [];
this.subscriberCollect = [];
this.events = [];
if (this.timeout) clearTimeout(this.timeout);
if (this.timeout)
clearTimeout(this.timeout)
}
}

View File

@ -1,12 +0,0 @@
export default class Signal {
awaiter: (() => void)[] = [];
sendSignal(): void {
this.awaiter.forEach((a) => a());
this.awaiter = [];
}
awaitSignal(): Promise<void> {
return new Promise((resolve) => {
this.awaiter.push(resolve);
});
}
}

View File

@ -1,14 +1,15 @@
{
"compilerOptions": {
"target": "esnext",
"module": "esnext",
"moduleResolution": "Node",
"outDir": "esm",
"target": "es2015",
"module": "commonjs",
"moduleResolution": "node",
"outDir": "lib",
"preserveWatchOutput": true,
"declaration": true,
"sourceMap": true,
"strict": true,
"isolatedModules": true
"strict": true
},
"include": ["./src"]
"include": [
"./src"
]
}

View File

@ -1,8 +0,0 @@
# THIS IS AN AUTOGENERATED FILE. DO NOT EDIT THIS FILE DIRECTLY.
# yarn lockfile v1
typescript@^4.0.5:
version "4.9.5"
resolved "https://npm.hibas123.de/typescript/-/typescript-4.9.5.tgz#095979f9bcc0d09da324d58d03ce8f8374cbe65a"
integrity sha512-1FXk9E2Hm+QzZQ7z+McJiHL4NW1F2EzMu9Nq9i3zAaGqibafqYwCVU6WyWAuyQRRzOlxou8xZSyXLEN8oKj24g==