switch to object options for Stream.async apis (#3213)

.changeset/chilled-ducks-sniff.md

@@ -0,0 +1,24 @@
+---
+"effect": major
+---
+
+Use object options for Stream.async apis
+
+Instead of:
+
+```ts
+Stream.async((emit) => {
+  //...
+}, 16);
+```
+
+You can now write:
+
+```ts
+Stream.async(
+  (emit) => {
+    //...
+  },
+  { bufferSize: 16 },
+);
+```

packages/effect/src/Stream.ts

@@ -314,7 +314,7 @@ export const as: {
 
 const _async: <A, E = never, R = never>(
   register: (emit: Emit.Emit<R, E, A, void>) => Effect.Effect<void, never, R> | void,
-  bufferSize?: number | "unbounded" | {
+  options?: { readonly bufferSize: "unbounded" } | {
     readonly bufferSize?: number | undefined
     readonly strategy?: "dropping" | "sliding" | "suspend" | undefined
   } | undefined
@@ -369,7 +369,7 @@ export {
  */
 export const asyncEffect: <A, E = never, R = never>(
   register: (emit: Emit.Emit<R, E, A, void>) => Effect.Effect<unknown, E, R>,
-  bufferSize?: number | "unbounded" | {
+  options?: { readonly bufferSize: "unbounded" } | {
     readonly bufferSize?: number | undefined
     readonly strategy?: "dropping" | "sliding" | "suspend" | undefined
   } | undefined
@@ -426,7 +426,7 @@ export const asyncPush: <A, E = never, R = never>(
  */
 export const asyncScoped: <A, E = never, R = never>(
   register: (emit: Emit.Emit<R, E, A, void>) => Effect.Effect<unknown, E, R | Scope.Scope>,
-  bufferSize?: number | "unbounded" | {
+  options?: { readonly bufferSize: "unbounded" } | {
     readonly bufferSize?: number | undefined
     readonly strategy?: "dropping" | "sliding" | "suspend" | undefined
   } | undefined

packages/effect/src/internal/stream.ts

@@ -466,23 +466,23 @@ export const as = dual<
 >(2, <A, E, R, B>(self: Stream.Stream<A, E, R>, value: B): Stream.Stream<B, E, R> => map(self, () => value))
 
 const queueFromBufferOptions = <A, E>(
-  bufferSize?: number | "unbounded" | {
+  options?: {
+    readonly bufferSize: "unbounded"
+  } | {
     readonly bufferSize?: number | undefined
     readonly strategy?: "dropping" | "sliding" | "suspend" | undefined
   } | undefined
 ): Effect.Effect<Queue.Queue<Take.Take<A, E>>> => {
-  if (bufferSize === "unbounded") {
+  if (options?.bufferSize === "unbounded") {
     return Queue.unbounded()
-  } else if (typeof bufferSize === "number" || bufferSize === undefined) {
-    return Queue.bounded(bufferSize ?? 16)
   }
-  switch (bufferSize.strategy) {
+  switch (options?.strategy) {
     case "dropping":
-      return Queue.dropping(bufferSize.bufferSize ?? 16)
+      return Queue.dropping(options.bufferSize ?? 16)
     case "sliding":
-      return Queue.sliding(bufferSize.bufferSize ?? 16)
+      return Queue.sliding(options.bufferSize ?? 16)
     default:
-      return Queue.bounded(bufferSize.bufferSize ?? 16)
+      return Queue.bounded(options?.bufferSize ?? 16)
   }
 }
 
@@ -491,13 +491,15 @@ export const _async = <A, E = never, R = never>(
   register: (
     emit: Emit.Emit<R, E, A, void>
   ) => Effect.Effect<void, never, R> | void,
-  bufferSize?: number | "unbounded" | {
+  options?: {
+    readonly bufferSize: "unbounded"
+  } | {
     readonly bufferSize?: number | undefined
     readonly strategy?: "dropping" | "sliding" | "suspend" | undefined
   } | undefined
 ): Stream.Stream<A, E, R> =>
   Effect.acquireRelease(
-    queueFromBufferOptions<A, E>(bufferSize),
+    queueFromBufferOptions<A, E>(options),
     (queue) => Queue.shutdown(queue)
   ).pipe(
     Effect.flatMap((output) =>
@@ -546,14 +548,16 @@ export const _async = <A, E = never, R = never>(
 /** @internal */
 export const asyncEffect = <A, E = never, R = never>(
   register: (emit: Emit.Emit<R, E, A, void>) => Effect.Effect<unknown, E, R>,
-  bufferSize?: number | "unbounded" | {
+  options?: {
+    readonly bufferSize: "unbounded"
+  } | {
     readonly bufferSize?: number | undefined
     readonly strategy?: "dropping" | "sliding" | "suspend" | undefined
   } | undefined
 ): Stream.Stream<A, E, R> =>
   pipe(
     Effect.acquireRelease(
-      queueFromBufferOptions<A, E>(bufferSize),
+      queueFromBufferOptions<A, E>(options),
       (queue) => Queue.shutdown(queue)
     ),
     Effect.flatMap((output) =>
@@ -649,14 +653,16 @@ export const asyncPush = <A, E = never, R = never>(
 /** @internal */
 export const asyncScoped = <A, E = never, R = never>(
   register: (emit: Emit.Emit<R, E, A, void>) => Effect.Effect<unknown, E, R | Scope.Scope>,
-  bufferSize?: number | "unbounded" | {
+  options?: {
+    readonly bufferSize: "unbounded"
+  } | {
     readonly bufferSize?: number | undefined
     readonly strategy?: "dropping" | "sliding" | "suspend" | undefined
   } | undefined
 ): Stream.Stream<A, E, Exclude<R, Scope.Scope>> =>
   pipe(
     Effect.acquireRelease(
-      queueFromBufferOptions<A, E>(bufferSize),
+      queueFromBufferOptions<A, E>(options),
       (queue) => Queue.shutdown(queue)
     ),
     Effect.flatMap((output) =>

packages/effect/test/Stream/async.test.ts

@@ -110,7 +110,7 @@ describe("Stream", () => {
           )
         )
         return Effect.void
-      }, 5)
+      }, { bufferSize: 5 })
       const sink = pipe(Sink.take<number>(1), Sink.zipRight(Sink.never))
       const fiber = yield* $(stream, Stream.run(sink), Effect.fork)
       yield* $(Ref.get(refCount), Effect.repeat({ while: (n) => n !== 7 }))
@@ -205,7 +205,7 @@ describe("Stream", () => {
           )
         )
         return Effect.void
-      }, 5)
+      }, { bufferSize: 5 })
       const sink = pipe(Sink.take<number>(1), Sink.zipRight(Sink.never))
       const fiber = yield* $(stream, Stream.run(sink), Effect.fork)
       yield* $(Ref.get(refCount), Effect.repeat({ while: (n) => n !== 7 }))
@@ -398,7 +398,7 @@ describe("Stream", () => {
           )
         )
         return Effect.void
-      }, 5)
+      }, { bufferSize: 5 })
       const sink = pipe(Sink.take<number>(1), Sink.zipRight(Sink.never))
       const fiber = yield* $(stream, Stream.run(sink), Effect.fork)
       yield* $(Ref.get(refCount), Effect.repeat({ while: (n) => n !== 7 }))