added Stream.fromEventListener

.changeset/green-lions-jog.md

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

packages/effect/src/Stream.ts

@@ -4555,3 +4555,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 1.0.0
+ */
+export const fromEventListener: (
+  target: EventTarget,
+  type: string,
+  options?: boolean | Omit<AddEventListenerOptions, "signal">
+) => Stream<Event> = internal.fromEventListener

packages/effect/src/internal/stream.ts

@@ -8045,3 +8045,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 = (
+  target: EventTarget,
+  type: string,
+  options?: boolean | Omit<AddEventListenerOptions, "signal">
+): Stream.Stream<Event> =>
+  _async<Event>((emit) => {
+    let batch: Array<Event> = []
+    let taskRunning = false
+    function cb(e: Event) {
+      batch.push(e)
+      if (!taskRunning) {
+        taskRunning = true
+        queueMicrotask(() => {
+          const events = batch
+          batch = []
+          taskRunning = false
+          emit.chunk(Chunk.unsafeFromArray(events))
+        })
+      }
+    }
+    target.addEventListener(type, cb, options)
+    return Effect.sync(() => target.removeEventListener(type, cb, options))
+  })

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

@@ -0,0 +1,27 @@
+/**
+ * @since 1.0.0
+ */
+
+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,28 @@
+/**
+ * @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
+ * @example
+ *
+ * import { BrowserStream } from "@effect/platform-browser";
+ * BrowserStream.fromEventListenerWindow("keypress") // Stream<KeyboardEvent>
+ */
+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/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.Stream<WindowEventMap[K]> => Stream.fromEventListener(window, type, options) as any
+
+/** @internal */
+export const fromEventListenerDocument = <K extends keyof DocumentEventMap>(
+  type: K,
+  options?: boolean | Omit<AddEventListenerOptions, "signal">
+): Stream.Stream<DocumentEventMap[K]> => Stream.fromEventListener(document, type, options) as any