consolidate GroupBy.evaluate & evaluateBuffer

Effect-TS · Sep 14, 2023 · 23a57ec · 23a57ec
1 parent a6973e4
commit 23a57ec
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 40 deletions.
diff --git a/.changeset/warm-walls-accept.md b/.changeset/warm-walls-accept.md
@@ -0,0 +1,5 @@
+---
+"@effect/stream": minor
+---
+
+consolidate GroupBy.evaluate & evaluateBuffer
diff --git a/src/GroupBy.ts b/src/GroupBy.ts
@@ -58,32 +58,16 @@ export declare namespace GroupBy {
  * @category destructors
  */
 export const evaluate: {
-  <K, E, V, R2, E2, A>(
-    f: (key: K, stream: Stream.Stream<never, E, V>) => Stream.Stream<R2, E2, A>
-  ): <R>(self: GroupBy<R, E, K, V>) => Stream.Stream<R2 | R, E | E2, A>
-  <R, K, E, V, R2, E2, A>(
-    self: GroupBy<R, E, K, V>,
-    f: (key: K, stream: Stream.Stream<never, E, V>) => Stream.Stream<R2, E2, A>
-  ): Stream.Stream<R | R2, E | E2, A>
-} = internal.evaluate
-
-/**
- * Like `evaluate`, but with a configurable `bufferSize` parameter.
- *
- * @since 1.0.0
- * @category destructors
- */
-export const evaluateBuffer: {
   <K, E, V, R2, E2, A>(
     f: (key: K, stream: Stream.Stream<never, E, V>) => Stream.Stream<R2, E2, A>,
-    bufferSize: number
+    options?: { readonly bufferSize?: number }
   ): <R>(self: GroupBy<R, E, K, V>) => Stream.Stream<R2 | R, E | E2, A>
   <R, K, E, V, R2, E2, A>(
     self: GroupBy<R, E, K, V>,
     f: (key: K, stream: Stream.Stream<never, E, V>) => Stream.Stream<R2, E2, A>,
-    bufferSize: number
+    options?: { readonly bufferSize?: number }
   ): Stream.Stream<R | R2, E | E2, A>
-} = internal.evaluateBuffer
+} = internal.evaluate
 
 /**
  * Filter the groups to be processed.

diff --git a/src/internal/groupBy.ts b/src/internal/groupBy.ts
@@ -37,43 +37,26 @@ const groupByVariance = {
 
 /** @internal */
 export const evaluate = dual<
-  <K, E, V, R2, E2, A>(
-    f: (key: K, stream: Stream.Stream<never, E, V>) => Stream.Stream<R2, E2, A>
-  ) => <R>(self: GroupBy.GroupBy<R, E, K, V>) => Stream.Stream<R2 | R, E | E2, A>,
-  <R, K, E, V, R2, E2, A>(
-    self: GroupBy.GroupBy<R, E, K, V>,
-    f: (key: K, stream: Stream.Stream<never, E, V>) => Stream.Stream<R2, E2, A>
-  ) => Stream.Stream<R2 | R, E | E2, A>
->(
-  2,
-  <R, K, E, V, R2, E2, A>(
-    self: GroupBy.GroupBy<R, E, K, V>,
-    f: (key: K, stream: Stream.Stream<never, E, V>) => Stream.Stream<R2, E2, A>
-  ): Stream.Stream<R | R2, E | E2, A> => evaluateBuffer(self, f, 16)
-)
-
-/** @internal */
-export const evaluateBuffer = dual<
   <K, E, V, R2, E2, A>(
     f: (key: K, stream: Stream.Stream<never, E, V>) => Stream.Stream<R2, E2, A>,
-    bufferSize: number
+    options?: { readonly bufferSize?: number }
   ) => <R>(self: GroupBy.GroupBy<R, E, K, V>) => Stream.Stream<R2 | R, E | E2, A>,
   <R, K, E, V, R2, E2, A>(
     self: GroupBy.GroupBy<R, E, K, V>,
     f: (key: K, stream: Stream.Stream<never, E, V>) => Stream.Stream<R2, E2, A>,
-    bufferSize: number
+    options?: { readonly bufferSize?: number }
   ) => Stream.Stream<R2 | R, E | E2, A>
 >(
   3,
   <R, K, E, V, R2, E2, A>(
     self: GroupBy.GroupBy<R, E, K, V>,
     f: (key: K, stream: Stream.Stream<never, E, V>) => Stream.Stream<R2, E2, A>,
-    bufferSize: number
+    options?: { readonly bufferSize?: number }
   ): Stream.Stream<R | R2, E | E2, A> =>
     stream.flatMap(
       self.grouped,
       ([key, queue]) => f(key, stream.flattenTake(stream.fromQueue(queue, { shutdown: true }))),
-      { concurrency: "unbounded", bufferSize }
+      { concurrency: "unbounded", bufferSize: options?.bufferSize ?? 16 }
     )
 )