added Stream.fromEventListener (#2513)

Co-authored-by: Tim <hello@timsmart.co>

.changeset/green-lions-jog.md

@@ -0,0 +1,6 @@
+---
+"effect": minor
+"@effect/platform-browser": patch
+---
+
+added Stream.fromEventListener, and BrowserStream.{fromEventListenerWindow, fromEventListenerDocument} for constructing a stream from addEventListener

packages/effect/src/Stream.ts

@@ -4558,3 +4558,13 @@ export const decodeText: {
  * @category encoding
  */
 export const encodeText: <E, R>(self: Stream<string, E, R>) => Stream<Uint8Array, E, R> = internal.encodeText
+
+/**
+ * Creates a `Stream` using addEventListener.
+ * @since 3.1.0
+ */
+export const fromEventListener: <A = Event>(
+  target: EventTarget,
+  type: string,
+  options?: boolean | Omit<AddEventListenerOptions, "signal">
+) => Stream<A> = internal.fromEventListener

packages/effect/src/internal/stream.ts

@@ -8048,3 +8048,28 @@ export const encodeText = <E, R>(self: Stream.Stream<string, E, R>): Stream.Stre
     const encoder = new TextEncoder()
     return map(self, (s) => encoder.encode(s))
   })
+
+/** @internal */
+export const fromEventListener = <A = Event>(
+  target: EventTarget,
+  type: string,
+  options?: boolean | Omit<AddEventListenerOptions, "signal">
+): Stream.Stream<A> =>
+  _async<A>((emit) => {
+    let batch: Array<A> = []
+    let taskRunning = false
+    function cb(e: A) {
+      batch.push(e)
+      if (!taskRunning) {
+        taskRunning = true
+        queueMicrotask(() => {
+          const events = batch
+          batch = []
+          taskRunning = false
+          emit.chunk(Chunk.unsafeFromArray(events))
+        })
+      }
+    }
+    target.addEventListener(type, cb as any, options)
+    return Effect.sync(() => target.removeEventListener(type, cb as any, options))
+  })

packages/effect/test/Stream/fromEventListener.test.ts

@@ -0,0 +1,23 @@
+import { Effect, Stream } from "effect"
+import * as it from "effect-test/utils/extend"
+import { describe } from "vitest"
+
+class TestTarget extends EventTarget {
+  emit() {
+    this.dispatchEvent(new Event("test-event"))
+  }
+}
+
+describe("Stream.fromEventListener", () => {
+  it.effect("emitted count", (ctx) =>
+    Effect.gen(function*(_) {
+      const target = new TestTarget()
+
+      const count = yield* _(
+        Stream.fromEventListener(target, "test-event"),
+        Stream.interruptWhen(Effect.sync(() => target.emit()).pipe(Effect.repeatN(2))),
+        Stream.runCount
+      )
+      ctx.expect(count).toEqual(3)
+    }))
+})

packages/platform-browser/src/BrowserStream.ts

@@ -0,0 +1,24 @@
+/**
+ * @since 1.0.0
+ */
+
+import type * as Stream from "effect/Stream"
+import * as internal from "./internal/stream.js"
+
+/**
+ * Creates a `Stream` from window.addEventListener.
+ * @since 1.0.0
+ */
+export const fromEventListenerWindow: <K extends keyof WindowEventMap>(
+  type: K,
+  options?: boolean | Omit<AddEventListenerOptions, "signal">
+) => Stream.Stream<WindowEventMap[K]> = internal.fromEventListenerWindow
+
+/**
+ * Creates a `Stream` from document.addEventListener.
+ * @since 1.0.0
+ */
+export const fromEventListenerDocument: <K extends keyof DocumentEventMap>(
+  type: K,
+  options?: boolean | Omit<AddEventListenerOptions, "signal">
+) => Stream.Stream<DocumentEventMap[K]> = internal.fromEventListenerDocument

packages/platform-browser/src/index.ts

@@ -13,6 +13,11 @@ export * as BrowserKeyValueStore from "./BrowserKeyValueStore.js"
  */
 export * as BrowserRuntime from "./BrowserRuntime.js"
 
+/**
+ * @since 1.0.0
+ */
+export * as BrowserStream from "./BrowserStream.js"
+
 /**
  * @since 1.0.0
  */

packages/platform-browser/src/internal/stream.ts

@@ -0,0 +1,17 @@
+/**
+ * @since 1.0.0
+ */
+
+import * as Stream from "effect/Stream"
+
+/** @internal */
+export const fromEventListenerWindow = <K extends keyof WindowEventMap>(
+  type: K,
+  options?: boolean | Omit<AddEventListenerOptions, "signal">
+) => Stream.fromEventListener<WindowEventMap[K]>(window, type, options)
+
+/** @internal */
+export const fromEventListenerDocument = <K extends keyof DocumentEventMap>(
+  type: K,
+  options?: boolean | Omit<AddEventListenerOptions, "signal">
+) => Stream.fromEventListener<DocumentEventMap[K]>(document, type, options)