allow Stream.split to use refinement for better type inference (#3917)

Co-authored-by: Kyle Sutton <[email protected]>

Author: SuttonKyle

.changeset/warm-meals-pretend.md

@@ -0,0 +1,5 @@
+---
+"effect": patch
+---
+
+Allow Stream.split to use refinement for better type inference

packages/effect/dtslint/Stream.ts

@@ -177,7 +177,7 @@ Stream.dropUntil(numbersOrStrings, Predicate.isNumber)
 pipe(numbersOrStrings, Stream.dropUntil(Predicate.isNumber))
 
 // -------------------------------------------------------------------------------------
-// splitWhere
+// split
 // -------------------------------------------------------------------------------------
 
 // $ExpectType Stream<Chunk<number>, never, never>
@@ -186,10 +186,10 @@ Stream.split(numbers, predicateNumbersOrStrings)
 // $ExpectType Stream<Chunk<number>, never, never>
 pipe(numbers, Stream.split(predicateNumbersOrStrings))
 
-// $ExpectType Stream<Chunk<string | number>, never, never>
+// $ExpectType Stream<Chunk<string>, never, never>
 Stream.split(numbersOrStrings, Predicate.isNumber)
 
-// $ExpectType Stream<Chunk<string | number>, never, never>
+// $ExpectType Stream<Chunk<string>, never, never>
 pipe(numbersOrStrings, Stream.split(Predicate.isNumber))
 
 // -------------------------------------------------------------------------------------

packages/effect/src/Stream.ts

@@ -4548,7 +4548,7 @@ export const someOrFail: {
 } = internal.someOrFail
 
 /**
- * Splits elements based on a predicate.
+ * Splits elements based on a predicate or refinement.
  *
  * ```ts
  * import * as Stream from "./Stream"
@@ -4566,7 +4566,11 @@ export const someOrFail: {
  * @category utils
  */
 export const split: {
+  <A, B extends A>(
+    refinement: Refinement<NoInfer<A>, B>
+  ): <E, R>(self: Stream<A, E, R>) => Stream<Chunk.Chunk<Exclude<A, B>>, E, R>
   <A>(predicate: Predicate<NoInfer<A>>): <E, R>(self: Stream<A, E, R>) => Stream<Chunk.Chunk<A>, E, R>
+  <A, E, R, B extends A>(self: Stream<A, E, R>, refinement: Refinement<A, B>): Stream<Chunk.Chunk<Exclude<A, B>>, E, R>
   <A, E, R>(self: Stream<A, E, R>, predicate: Predicate<A>): Stream<Chunk.Chunk<A>, E, R>
 } = internal.split
 

packages/effect/src/internal/stream.ts

@@ -6189,44 +6189,59 @@ export const slidingSize = dual<
 )
 
 /** @internal */
-export const split = dual<
-  <A>(predicate: Predicate<NoInfer<A>>) => <E, R>(self: Stream.Stream<A, E, R>) => Stream.Stream<Chunk.Chunk<A>, E, R>,
-  <A, E, R>(self: Stream.Stream<A, E, R>, predicate: Predicate<A>) => Stream.Stream<Chunk.Chunk<A>, E, R>
->(2, <A, E, R>(self: Stream.Stream<A, E, R>, predicate: Predicate<A>): Stream.Stream<Chunk.Chunk<A>, E, R> => {
-  const split = (
-    leftovers: Chunk.Chunk<A>,
-    input: Chunk.Chunk<A>
-  ): Channel.Channel<Chunk.Chunk<Chunk.Chunk<A>>, Chunk.Chunk<A>, E, E, unknown, unknown, R> => {
-    const [chunk, remaining] = pipe(leftovers, Chunk.appendAll(input), Chunk.splitWhere(predicate))
-    if (Chunk.isEmpty(chunk) || Chunk.isEmpty(remaining)) {
-      return loop(pipe(chunk, Chunk.appendAll(pipe(remaining, Chunk.drop(1)))))
+export const split: {
+  <A, B extends A>(
+    refinement: Refinement<NoInfer<A>, B>
+  ): <E, R>(self: Stream.Stream<A, E, R>) => Stream.Stream<Chunk.Chunk<Exclude<A, B>>, E, R>
+  <A>(
+    predicate: Predicate<NoInfer<A>>
+  ): <E, R>(self: Stream.Stream<A, E, R>) => Stream.Stream<Chunk.Chunk<A>, E, R>
+  <A, E, R, B extends A>(
+    self: Stream.Stream<A, E, R>,
+    refinement: Refinement<A, B>
+  ): Stream.Stream<Chunk.Chunk<Exclude<A, B>>, E, R>
+  <A, E, R>(self: Stream.Stream<A, E, R>, predicate: Predicate<A>): Stream.Stream<Chunk.Chunk<A>, E, R>
+} = dual(
+  2,
+  <A, E, R>(
+    self: Stream.Stream<A, E, R>,
+    predicate: Predicate<A>
+  ): Stream.Stream<Chunk.Chunk<A>, E, R> => {
+    const split = (
+      leftovers: Chunk.Chunk<A>,
+      input: Chunk.Chunk<A>
+    ): Channel.Channel<Chunk.Chunk<Chunk.Chunk<A>>, Chunk.Chunk<A>, E, E, unknown, unknown, R> => {
+      const [chunk, remaining] = pipe(leftovers, Chunk.appendAll(input), Chunk.splitWhere(predicate))
+      if (Chunk.isEmpty(chunk) || Chunk.isEmpty(remaining)) {
+        return loop(pipe(chunk, Chunk.appendAll(pipe(remaining, Chunk.drop(1)))))
+      }
+      return pipe(
+        core.write(Chunk.of(chunk)),
+        core.flatMap(() => split(Chunk.empty(), pipe(remaining, Chunk.drop(1))))
+      )
     }
-    return pipe(
-      core.write(Chunk.of(chunk)),
-      core.flatMap(() => split(Chunk.empty(), pipe(remaining, Chunk.drop(1))))
-    )
-  }
-  const loop = (
-    leftovers: Chunk.Chunk<A>
-  ): Channel.Channel<Chunk.Chunk<Chunk.Chunk<A>>, Chunk.Chunk<A>, E, E, unknown, unknown, R> =>
-    core.readWith({
-      onInput: (input: Chunk.Chunk<A>) => split(leftovers, input),
-      onFailure: core.fail,
-      onDone: () => {
-        if (Chunk.isEmpty(leftovers)) {
-          return core.void
-        }
-        if (Option.isNone(pipe(leftovers, Chunk.findFirst(predicate)))) {
-          return channel.zipRight(core.write(Chunk.of(leftovers)), core.void)
+    const loop = (
+      leftovers: Chunk.Chunk<A>
+    ): Channel.Channel<Chunk.Chunk<Chunk.Chunk<A>>, Chunk.Chunk<A>, E, E, unknown, unknown, R> =>
+      core.readWith({
+        onInput: (input: Chunk.Chunk<A>) => split(leftovers, input),
+        onFailure: core.fail,
+        onDone: () => {
+          if (Chunk.isEmpty(leftovers)) {
+            return core.void
+          }
+          if (Option.isNone(pipe(leftovers, Chunk.findFirst(predicate)))) {
+            return channel.zipRight(core.write(Chunk.of(leftovers)), core.void)
+          }
+          return channel.zipRight(
+            split(Chunk.empty(), leftovers),
+            core.void
+          )
         }
-        return channel.zipRight(
-          split(Chunk.empty(), leftovers),
-          core.void
-        )
-      }
-    })
-  return new StreamImpl(pipe(toChannel(self), core.pipeTo(loop(Chunk.empty()))))
-})
+      })
+    return new StreamImpl(pipe(toChannel(self), core.pipeTo(loop(Chunk.empty()))))
+  }
+)
 
 /** @internal */
 export const splitOnChunk = dual<