@effection/events - npm Package Compare versions

Comparing version 2.0.0-side-effects.1628189696867 to 2.0.0-v2-writable-unification.1633595877341

CHANGELOG.md

 # Changelog
 
+## \[2.0.0-beta.20]
+
+- Fix a bug when using blockParent where the children are not getting halt on an explicit halt.
+  - Bumped due to a bump in @effection/core.
+  - [1cd9803](https://github.com/thefrontside/effection/commit/1cd98033d2641989114f9589c7d887954fa66781) Fix halting children for blockParent tasks on 2021-09-30
+
+## \[2.0.0-beta.19]
+
+- Add Stream `toBuffer` and Stream `buffered` so we have both options on either accessing the buffer directly or returning the stream.
+  - Bumped due to a bump in @effection/stream.
+  - [fe60532](https://github.com/thefrontside/effection/commit/fe60532c3f8cfdd8b53c324b7ea8e38e437f080f) Add both toBuffer and buffered to Stream on 2021-09-30
+
+## \[2.0.0-beta.18]
+
+- Stream `buffer` returns the actual buffer and gives direct access to it
+  - Bumped due to a bump in @effection/stream.
+  - [07c8f83](https://github.com/thefrontside/effection/commit/07c8f83b5968f347ca72795c447be411e66274ed) Stream `buffer` returns the actual buffer on 2021-09-30
+
+## \[2.0.0-beta.17]
+
+- - [0248d79](https://github.com/thefrontside/effection/commit/0248d79a33dcfc4200b0832aba975c9cad08981e) Add package readmes on 2021-09-28
+- Remove operation resolutions entirely, use Future instead
+  - [5f67d61](https://github.com/thefrontside/effection/commit/5f67d610324af158eba67be5600d413fc1f2ace1) Add changeset on 2021-09-29
+- Split off `Stream` from subscription package into its own `@effection/stream` package
+  - [248de1d](https://github.com/thefrontside/effection/commit/248de1dd31d172762d9601a2b5acd983dce61ab0) Split `Stream` into its own package on 2021-09-27
+
+## \[2.0.0-beta.16]
+
+- Adjust the propagation of errors for resources to make it possible to catch errors from `init`
+  - Bumped due to a bump in @effection/core.
+  - [75a7248](https://github.com/thefrontside/effection/commit/75a7248ae13d1126bbcaf9b6223f348168e987d0) Catch errors thrown during resource init on 2021-09-21
+- Enable support for resources in higher order operations `all`, `race` and `withTimeout`.
+  - Bumped due to a bump in @effection/core.
+  - [bbe6cdc](https://github.com/thefrontside/effection/commit/bbe6cdc44184a7669278d0d01ad23a2a79a69e52) Enable resource support for higher order operations on 2021-09-09
+
+## \[2.0.0-beta.15]
+
+- Add shortcuts to create resolved/rejected/halted futures via Future.resolve(123), etc...
+  - Bumped due to a bump in @effection/core.
+  - [9599dde](https://github.com/thefrontside/effection/commit/9599dde14e9bc3ba4ac7ea473e8624164727be0c) Add shortcuts for resolves/rejected/halted future on 2021-09-08
+
+## \[2.0.0-beta.14]
+
+- Add @effection/fetch as a dependency and reexport it
+  - Bumped due to a bump in @effection/core.
+  - [5ab5d06](https://github.com/thefrontside/effection/commit/5ab5d0691af75f3583de97402b5aac12325e2918) Reexport @effection/fetch from effection package on 2021-08-26
+- Share internal run loop among task, task future and task controller. Prevents race conditions which cause internal errors.
+  - Bumped due to a bump in @effection/core.
+  - [222d511](https://github.com/thefrontside/effection/commit/222d5116c388c5b597cc3ec5e0fb64b4d22b273a) Share event loop among controller, task and future on 2021-09-01
+- Introduce task scope as an alternative to resources for being able to access the outer scope of an operation
+  - Bumped due to a bump in @effection/core.
+  - [3ed11bd](https://github.com/thefrontside/effection/commit/3ed11bd4f5d980cd130ea894a63acb57450c5aac) Make resource task accessible through init task on 2021-08-27
+- Add `toString()` method to task for nicely formatted rendering of task structure
+  - Bumped due to a bump in @effection/core.
+  - [9a63928](https://github.com/thefrontside/effection/commit/9a6392836704ad527d6da5195f5736462d69bef8) Add toString output for tasks on 2021-08-31
+
+## \[2.0.0-beta.13]
+
+- Allow streams created through `on` and `onEmit` to be named
+  - [cc49d47](https://github.com/thefrontside/effection/commit/cc49d47e337785f6524d79760b0d2dd4d766e6ed) Name event streams on 2021-08-27
+
+## \[2.0.0-beta.12]
+
+- Update core dependency
+  - Bumped due to a bump in @effection/mocha.
+  - [d92eee5](https://github.com/thefrontside/effection/commit/d92eee594fdb8dc6d8ab6a37b6aa362122e63f6e) Update core dependency on 2021-08-16
+
+## \[2.0.0-beta.11]
+
+- Use Object.create to wrap error objects rather than copying properties
+  - Bumped due to a bump in @effection/core.
+  - [a56ae2a](https://github.com/thefrontside/effection/commit/a56ae2af8a6247697b8b6253bd35b6d9e569613d) Use Object.create to create error object with trace on 2021-08-16
+
+## \[2.0.0-beta.10]
+
+- add `Task#spawn` operation to spawn new task with a specific scope
+  - Bumped due to a bump in @effection/core.
+  - [a71d65b](https://github.com/thefrontside/effection/commit/a71d65b77df5c337a78b7934edd181080eacf5bf) Add changefile on 2021-07-27
+
+## \[2.0.0-beta.9]
+
+- Add sideEffects field to package.json
+  - [383141d](https://github.com/thefrontside/effection/commit/383141dc556c6a781d98087f3b68085d5eb31173) Add sideEffects field to package.json ([#470](https://github.com/thefrontside/effection/pull/470)) on 2021-08-05
+
 ## \[2.0.0-beta.8]
@@ -4,0 +88,0 @@

dist-cjs/event-source.d.ts

 /// <reference types="node" />
 import { EventEmitter } from 'events';
 export declare type EventSource = EventEmitterSource | EventTargetSource;
-export declare function addListener(source: EventSource, name: string, listener: () => void): void;
-export declare function removeListener(source: EventSource, name: string, listener: () => void): void;
+export declare function addListener<TArgs extends unknown[] = [unknown]>(source: EventSource, name: string, listener: (...args: TArgs) => void): void;
+export declare function removeListener<TArgs extends unknown[] = [unknown]>(source: EventSource, name: string, listener: (...args: TArgs) => void): void;
 interface EventTargetSource {
@@ -7,0 +7,0 @@ addEventListener: EventTarget["addEventListener"];

dist-cjs/on.d.ts

@@ -1,2 +0,2 @@
-import { Stream } from '@effection/subscription';
+import { Stream } from '@effection/stream';
 import { EventSource } from './event-source';
@@ -16,3 +16,3 @@ /**
  * ```javascript
- * task.spawn(on(document, 'click').forEach(event => {
+ * yield spawn(on(document, 'click').forEach(event => {
  *   console.log(`click at (${event.pageX}, ${event.pageY})`);
@@ -28,7 +28,7 @@ * }));
  * let buffer = '';
- * task.spawn(on(process.stdin, 'data').forEach(data => {
+ * yield spawn(on(process.stdin, 'data').forEach(data => {
  *   buffer += data;
  * }));
  */
-export declare function on<T = unknown>(source: EventSource, name: string): Stream<T, void>;
+export declare function on<T = unknown>(source: EventSource, name: string, streamName?: string): Stream<T, void>;
 /**
@@ -50,3 +50,3 @@ * Exactly like {@link on | on()} except each value produced by the
  *
- * task.spawn(onEmit(emitter, 'multiplication').forEach(([left, right]) => {
+ * yield spawn(onEmit(emitter, 'multiplication').forEach(([left, right]) => {
  *   console.log(`${left} times ${right} = ${left * right}!`);
@@ -62,3 +62,3 @@ * }));
  */
-export declare function onEmit<T extends Array<unknown> = unknown[]>(source: EventSource, name: string): Stream<T, void>;
+export declare function onEmit<T extends Array<unknown> = unknown[]>(source: EventSource, name: string, streamName?: string): Stream<T, void>;
 //# sourceMappingURL=on.d.ts.map

dist-cjs/on.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
 exports.onEmit = exports.on = void 0;
-const subscription_1 = require("@effection/subscription");
+const core_1 = require("@effection/core");
+const stream_1 = require("@effection/stream");
 const event_source_1 = require("./event-source");
@@ -19,3 +20,3 @@ /**
  * ```javascript
- * task.spawn(on(document, 'click').forEach(event => {
+ * yield spawn(on(document, 'click').forEach(event => {
  *   console.log(`click at (${event.pageX}, ${event.pageY})`);
@@ -31,16 +32,17 @@ * }));
  * let buffer = '';
- * task.spawn(on(process.stdin, 'data').forEach(data => {
+ * yield spawn(on(process.stdin, 'data').forEach(data => {
  *   buffer += data;
  * }));
  */
-function on(source, name) {
-    return subscription_1.createStream((publish) => ({
-        name: 'listen',
-        labels: { eventName: name, source: source.toString() },
-        perform() {
-            let listener = (...args) => publish(args[0]);
-            event_source_1.addListener(source, name, listener);
-            return () => event_source_1.removeListener(source, name, listener);
+function on(source, name, streamName = `on('${name}')`) {
+    return stream_1.createStream((publish) => core_1.withLabels(function* (task) {
+        let listener = (value) => task.run(publish(value));
+        event_source_1.addListener(source, name, listener);
+        try {
+            yield;
         }
-    }), `on('${name}')`);
+        finally {
+            event_source_1.removeListener(source, name, listener);
+        }
+    }, { name: 'listen', eventName: name, source: source.toString() }), streamName);
 }
@@ -64,3 +66,3 @@ exports.on = on;
  *
- * task.spawn(onEmit(emitter, 'multiplication').forEach(([left, right]) => {
+ * yield spawn(onEmit(emitter, 'multiplication').forEach(([left, right]) => {
  *   console.log(`${left} times ${right} = ${left * right}!`);
@@ -76,14 +78,15 @@ * }));
  */
-function onEmit(source, name) {
-    return subscription_1.createStream((publish) => ({
-        name: 'listen',
-        labels: { eventName: name, source: source.toString() },
-        perform() {
-            let listener = (...args) => publish(args);
-            event_source_1.addListener(source, name, listener);
-            return () => event_source_1.removeListener(source, name, listener);
+function onEmit(source, name, streamName = `onEmit('${name}')`) {
+    return stream_1.createStream((publish) => core_1.withLabels(function* (task) {
+        let listener = (...args) => task.run(publish(args));
+        event_source_1.addListener(source, name, listener);
+        try {
+            yield;
         }
-    }), `onEmit('${name}')`);
+        finally {
+            event_source_1.removeListener(source, name, listener);
+        }
+    }, { name: 'listen', eventName: name, source: source.toString() }), streamName);
 }
 exports.onEmit = onEmit;
 //# sourceMappingURL=on.js.map

dist-cjs/once.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
 exports.onceEmit = exports.once = void 0;
+const core_1 = require("@effection/core");
 const event_source_1 = require("./event-source");
@@ -28,11 +29,9 @@ /**
 function once(source, eventName) {
-    return {
-        name: `once`,
-        labels: { eventName, source: source.toString() },
-        perform(resolve) {
-            let listener = (...args) => { resolve(args[0]); };
-            event_source_1.addListener(source, eventName, listener);
-            return () => event_source_1.removeListener(source, eventName, listener);
-        }
-    };
+    return core_1.withLabels((task) => {
+        let { future, resolve } = core_1.createFuture();
+        let listener = (...args) => { resolve(args[0]); };
+        event_source_1.addListener(source, eventName, listener);
+        task.consume(() => event_source_1.removeListener(source, eventName, listener));
+        return future;
+    }, { name: `once`, eventName, source: source.toString() });
 }
@@ -64,13 +63,11 @@ exports.once = once;
 function onceEmit(source, eventName) {
-    return {
-        name: `onceEmit`,
-        labels: { eventName, source: source.toString() },
-        perform(resolve) {
-            let listener = (...args) => { resolve(args); };
-            event_source_1.addListener(source, eventName, listener);
-            return () => event_source_1.removeListener(source, eventName, listener);
-        }
-    };
+    return core_1.withLabels((task) => {
+        let { future, resolve } = core_1.createFuture();
+        let listener = (...args) => { resolve(args); };
+        event_source_1.addListener(source, eventName, listener);
+        task.consume(() => event_source_1.removeListener(source, eventName, listener));
+        return future;
+    }, { name: `onceEmit`, eventName, source: source.toString() });
 }
 exports.onceEmit = onceEmit;
 //# sourceMappingURL=once.js.map

dist-esm/event-source.d.ts

 /// <reference types="node" />
 import { EventEmitter } from 'events';
 export declare type EventSource = EventEmitterSource | EventTargetSource;
-export declare function addListener(source: EventSource, name: string, listener: () => void): void;
-export declare function removeListener(source: EventSource, name: string, listener: () => void): void;
+export declare function addListener<TArgs extends unknown[] = [unknown]>(source: EventSource, name: string, listener: (...args: TArgs) => void): void;
+export declare function removeListener<TArgs extends unknown[] = [unknown]>(source: EventSource, name: string, listener: (...args: TArgs) => void): void;
 interface EventTargetSource {
@@ -7,0 +7,0 @@ addEventListener: EventTarget["addEventListener"];

dist-esm/on.d.ts

@@ -1,2 +0,2 @@
-import { Stream } from '@effection/subscription';
+import { Stream } from '@effection/stream';
 import { EventSource } from './event-source';
@@ -16,3 +16,3 @@ /**
  * ```javascript
- * task.spawn(on(document, 'click').forEach(event => {
+ * yield spawn(on(document, 'click').forEach(event => {
  *   console.log(`click at (${event.pageX}, ${event.pageY})`);
@@ -28,7 +28,7 @@ * }));
  * let buffer = '';
- * task.spawn(on(process.stdin, 'data').forEach(data => {
+ * yield spawn(on(process.stdin, 'data').forEach(data => {
  *   buffer += data;
  * }));
  */
-export declare function on<T = unknown>(source: EventSource, name: string): Stream<T, void>;
+export declare function on<T = unknown>(source: EventSource, name: string, streamName?: string): Stream<T, void>;
 /**
@@ -50,3 +50,3 @@ * Exactly like {@link on | on()} except each value produced by the
  *
- * task.spawn(onEmit(emitter, 'multiplication').forEach(([left, right]) => {
+ * yield spawn(onEmit(emitter, 'multiplication').forEach(([left, right]) => {
  *   console.log(`${left} times ${right} = ${left * right}!`);
@@ -62,3 +62,3 @@ * }));
  */
-export declare function onEmit<T extends Array<unknown> = unknown[]>(source: EventSource, name: string): Stream<T, void>;
+export declare function onEmit<T extends Array<unknown> = unknown[]>(source: EventSource, name: string, streamName?: string): Stream<T, void>;
 //# sourceMappingURL=on.d.ts.map

dist-esm/on.js

@@ -1,2 +0,3 @@
-import { createStream } from '@effection/subscription';
+import { withLabels } from '@effection/core';
+import { createStream } from '@effection/stream';
 import { addListener, removeListener } from './event-source';
@@ -16,3 +17,3 @@ /**
  * ```javascript
- * task.spawn(on(document, 'click').forEach(event => {
+ * yield spawn(on(document, 'click').forEach(event => {
  *   console.log(`click at (${event.pageX}, ${event.pageY})`);
@@ -28,16 +29,17 @@ * }));
  * let buffer = '';
- * task.spawn(on(process.stdin, 'data').forEach(data => {
+ * yield spawn(on(process.stdin, 'data').forEach(data => {
  *   buffer += data;
  * }));
  */
-export function on(source, name) {
-    return createStream((publish) => ({
-        name: 'listen',
-        labels: { eventName: name, source: source.toString() },
-        perform() {
-            let listener = (...args) => publish(args[0]);
-            addListener(source, name, listener);
-            return () => removeListener(source, name, listener);
+export function on(source, name, streamName = `on('${name}')`) {
+    return createStream((publish) => withLabels(function* (task) {
+        let listener = (value) => task.run(publish(value));
+        addListener(source, name, listener);
+        try {
+            yield;
         }
-    }), `on('${name}')`);
+        finally {
+            removeListener(source, name, listener);
+        }
+    }, { name: 'listen', eventName: name, source: source.toString() }), streamName);
 }
@@ -60,3 +62,3 @@ /**
  *
- * task.spawn(onEmit(emitter, 'multiplication').forEach(([left, right]) => {
+ * yield spawn(onEmit(emitter, 'multiplication').forEach(([left, right]) => {
  *   console.log(`${left} times ${right} = ${left * right}!`);
@@ -72,13 +74,14 @@ * }));
  */
-export function onEmit(source, name) {
-    return createStream((publish) => ({
-        name: 'listen',
-        labels: { eventName: name, source: source.toString() },
-        perform() {
-            let listener = (...args) => publish(args);
-            addListener(source, name, listener);
-            return () => removeListener(source, name, listener);
+export function onEmit(source, name, streamName = `onEmit('${name}')`) {
+    return createStream((publish) => withLabels(function* (task) {
+        let listener = (...args) => task.run(publish(args));
+        addListener(source, name, listener);
+        try {
+            yield;
         }
-    }), `onEmit('${name}')`);
+        finally {
+            removeListener(source, name, listener);
+        }
+    }, { name: 'listen', eventName: name, source: source.toString() }), streamName);
 }
 //# sourceMappingURL=on.js.map

dist-esm/once.js

@@ -0,1 +1,2 @@
+import { createFuture, withLabels } from '@effection/core';
 import { addListener, removeListener } from './event-source';
@@ -25,11 +26,9 @@ /**
 export function once(source, eventName) {
-    return {
-        name: `once`,
-        labels: { eventName, source: source.toString() },
-        perform(resolve) {
-            let listener = (...args) => { resolve(args[0]); };
-            addListener(source, eventName, listener);
-            return () => removeListener(source, eventName, listener);
-        }
-    };
+    return withLabels((task) => {
+        let { future, resolve } = createFuture();
+        let listener = (...args) => { resolve(args[0]); };
+        addListener(source, eventName, listener);
+        task.consume(() => removeListener(source, eventName, listener));
+        return future;
+    }, { name: `once`, eventName, source: source.toString() });
 }
@@ -60,12 +59,10 @@ /**
 export function onceEmit(source, eventName) {
-    return {
-        name: `onceEmit`,
-        labels: { eventName, source: source.toString() },
-        perform(resolve) {
-            let listener = (...args) => { resolve(args); };
-            addListener(source, eventName, listener);
-            return () => removeListener(source, eventName, listener);
-        }
-    };
+    return withLabels((task) => {
+        let { future, resolve } = createFuture();
+        let listener = (...args) => { resolve(args); };
+        addListener(source, eventName, listener);
+        task.consume(() => removeListener(source, eventName, listener));
+        return future;
+    }, { name: `onceEmit`, eventName, source: source.toString() });
 }
 //# sourceMappingURL=once.js.map

package.json

 {
   "name": "@effection/events",
-  "version": "2.0.0-side-effects.1628189696867",
+  "version": "2.0.0-v2-writable-unification.1633595877341",
   "description": "Helpers for listening to events with effection",
@@ -10,3 +10,3 @@ "main": "dist-cjs/index.js",
   "sideEffects": false,
-  "homepage": "https://github.com/thefrontside/effection",
+  "homepage": "https://frontside.com/effection",
   "repository": {
@@ -32,7 +32,6 @@ "type": "git",
   "dependencies": {
-    "@effection/core": "2.0.0-side-effects.1628189696867",
-    "@effection/subscription": "2.0.0-side-effects.1628189696867"
+    "@effection/core": "2.0.0-v2-writable-unification.1633595877341",
+    "@effection/stream": "2.0.0-v2-writable-unification.1633595877341"
   },
   "devDependencies": {
-    "@effection/mocha": "2.0.0-side-effects.1628189696867",
     "@frontside/tsconfig": "^1.2.0",
@@ -39,0 +38,0 @@ "@types/node": "16.3.2",

README.md

 # @effection/events
+[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
+[![Created by Frontside](https://img.shields.io/badge/created%20by-frontside-26abe8.svg)](https://frontside.com)
+[![Chat on Discord](https://img.shields.io/discord/700803887132704931?Label=Discord)](https://discord.gg/Ug5nWH8)
 
-Helpers for working with events in Effection. These helpers work both with
-Node.js style EventEmitters (`on`/`off`) and with browser style EventTarget
-(`addEventListeners`/`removeEventListener`).
+[Effection][] is the structured concurrency toolkit for JavaScript. You can find
+detailed information about using it to handle events at [https://frontside.com/effection/docs/guides/events](https://frontside.com/effection/docs/guides/events)
 
-## Usage
-
-The `once` operation can be used to wait for an event to occur exactly once.
-
-``` typescript
-import { main } from '@effection/main';
-import { once } from '@effection/events';
-
-main(function* sayHello() {
-  yield once(window, "DOMContentLoaded");
-  console.log('Hello World!');
-});
-```
-
-The `on` operation returns a subscription which can be used to loop over events:
-
-``` typescript
-import { main } from '@effection/main';
-import { on } from '@effection/events';
-
-main(function* sayHello() {
-  let messages = yield on(window, "message");
-
-  while(true) {
-    let message = yield messages.next();
-    console.log('Got message:', message);
-  }
-});
-```
-
-It is a common pattern, primarily in Node to have an event called `error` which
-has an error as its first argument. `@effection/events` includes a helper to
-make handling this situation convenient:
-
-``` typescript
-import { main } from '@effection/main';
-import { once, throwOnErrorEvent } from '@effection/events';
-
-main(function* sayHello() {
-  let someSocket = createSocket();
-
-  yield throwOnErrorEvent(someSocket);
-
-  yield once(someSocket, "open");
-
-  let messages = yield once(someSocket, "message");
-
-  while(true) {
-    let message = yield messages.next();
-    console.log('Got message:', message);
-  }
-});
-```
+[Effection]: https://frontside.com/effection

src/event-source.ts

@@ -9,15 +9,15 @@ import { EventEmitter } from 'events';
 
-export function addListener(source: EventSource, name: string, listener: () => void): void {
+export function addListener<TArgs extends unknown[] = [unknown]>(source: EventSource, name: string, listener: (...args: TArgs) => void): void {
   if(isEventTarget(source)) {
-    source.addEventListener(name, listener);
+    source.addEventListener(name, listener as () => void);
   } else {
-    source.on(name, listener);
+    source.on(name, listener as () => void);
   }
 }
 
-export function removeListener(source: EventSource, name: string, listener: () => void): void {
+export function removeListener<TArgs extends unknown[] = [unknown]>(source: EventSource, name: string, listener: (...args: TArgs) => void): void {
   if(isEventTarget(source)) {
-    source.removeEventListener(name, listener);
+    source.removeEventListener(name, listener as () => void);
   } else {
-    source.off(name, listener);
+    source.off(name, listener as () => void);
   }
@@ -24,0 +24,0 @@ }

src/on.ts

@@ -1,2 +0,3 @@
-import { createStream, Stream } from '@effection/subscription';
+import { withLabels } from '@effection/core';
+import { createStream, Stream } from '@effection/stream';
 import { EventSource, addListener, removeListener } from './event-source';
@@ -18,3 +19,3 @@
  * ```javascript
- * task.spawn(on(document, 'click').forEach(event => {
+ * yield spawn(on(document, 'click').forEach(event => {
  *   console.log(`click at (${event.pageX}, ${event.pageY})`);
@@ -30,16 +31,16 @@ * }));
  * let buffer = '';
- * task.spawn(on(process.stdin, 'data').forEach(data => {
+ * yield spawn(on(process.stdin, 'data').forEach(data => {
  *   buffer += data;
  * }));
  */
-export function on<T = unknown>(source: EventSource, name: string): Stream<T, void> {
-  return createStream((publish) => ({
-    name: 'listen',
-    labels: { eventName: name, source: source.toString() },
-    perform() {
-      let listener = (...args: T[]) => publish(args[0]);
-      addListener(source, name, listener);
-      return () => removeListener(source, name, listener);
+export function on<T = unknown>(source: EventSource, name: string, streamName = `on('${name}')`): Stream<T, void> {
+  return createStream((publish) => withLabels(function*(task) {
+    let listener = (value: T) => task.run(publish(value));
+    addListener(source, name, listener);
+    try {
+      yield;
+    } finally {
+      removeListener(source, name, listener);
     }
-  }), `on('${name}')`);
+  }, { name: 'listen', eventName: name, source: source.toString() }), streamName);
 }
@@ -63,3 +64,3 @@
  *
- * task.spawn(onEmit(emitter, 'multiplication').forEach(([left, right]) => {
+ * yield spawn(onEmit(emitter, 'multiplication').forEach(([left, right]) => {
  *   console.log(`${left} times ${right} = ${left * right}!`);
@@ -75,12 +76,12 @@ * }));
  */
-export function onEmit<T extends Array<unknown> = unknown[]>(source: EventSource, name: string): Stream<T, void> {
-    return createStream((publish) => ({
-    name: 'listen',
-    labels: { eventName: name, source: source.toString() },
-    perform() {
-      let listener = (...args: T) => publish(args);
-      addListener(source, name, listener);
-      return () => removeListener(source, name, listener);
+export function onEmit<T extends Array<unknown> = unknown[]>(source: EventSource, name: string, streamName = `onEmit('${name}')`): Stream<T, void> {
+  return createStream((publish) => withLabels(function*(task) {
+    let listener = (...args: T) => task.run(publish(args));
+    addListener(source, name, listener);
+    try {
+      yield;
+    } finally {
+      removeListener(source, name, listener);
     }
-  }), `onEmit('${name}')`);
+  }, { name: 'listen', eventName: name, source: source.toString() }), streamName);
 }

src/once.ts

@@ -1,2 +0,2 @@
-import { Operation } from '@effection/core';
+import { Operation, createFuture, withLabels } from '@effection/core';
 import { EventSource, addListener, removeListener } from './event-source';
@@ -28,11 +28,9 @@
 export function once<T = unknown>(source: EventSource, eventName: string): Operation<T> {
-  return {
-    name: `once`,
-    labels: { eventName, source: source.toString() },
-    perform(resolve) {
-      let listener = (...args: T[]) => { resolve(args[0]) };
-      addListener(source, eventName, listener);
-      return () => removeListener(source, eventName, listener);
-    }
-  };
+  return withLabels((task) => {
+    let { future, resolve } = createFuture<T>();
+    let listener = (...args: T[]) => { resolve(args[0]) };
+    addListener(source, eventName, listener);
+    task.consume(() => removeListener(source, eventName, listener));
+    return future;
+  }, { name: `once`, eventName, source: source.toString() });
 }
@@ -64,11 +62,9 @@
 export function onceEmit<TArgs extends unknown[] = unknown[]>(source: EventSource, eventName: string): Operation<TArgs> {
-  return {
-    name: `onceEmit`,
-    labels: { eventName, source: source.toString() },
-    perform(resolve) {
-      let listener = (...args: unknown[]) => { resolve(args as TArgs) };
-      addListener(source, eventName, listener);
-      return () => removeListener(source, eventName, listener);
-    }
-  };
+  return withLabels((task) => {
+    let { future, resolve } = createFuture<TArgs>();
+    let listener = (...args: unknown[]) => { resolve(args as TArgs) };
+    addListener(source, eventName, listener);
+    task.consume(() => removeListener(source, eventName, listener));
+    return future;
+  }, { name: `onceEmit`, eventName, source: source.toString() });
 }

No alert changes

Improved metrics

Total package byte prevSize

58821

increased by12.52%

Dev dependency count

6

decreased by-14.29%

Worsened metrics

Lines of code

816

decreased by-0.37%

Number of lines in readme file

10

decreased by-83.33%

Dependency changes

• + Added@effection/stream@2.0.0-v2-writable-unification.1633595877341

• + Added@effection/core@2.0.0-v2-writable-unification.1633595877341(transitive)

• + Added@effection/stream@2.0.0-v2-writable-unification.1633595877341(transitive)

• + Added@effection/subscription@2.0.0-v2-writable-unification.1633595877341(transitive)

• - Removed@effection/subscription@2.0.0-side-effects.1628189696867

• - Removed@effection/core@2.0.0-side-effects.1628189696867(transitive)

• - Removed@effection/subscription@2.0.0-side-effects.1628189696867(transitive)

• Updated@effection/core@2.0.0-v2-writable-unification.1633595877341