From 8a2f1fcf08d7b452abfba557ad455273199b4cbe Mon Sep 17 00:00:00 2001
From: Leo Conforti <leo@leoconforti.us>
Date: Tue, 7 May 2024 04:18:02 +0000
Subject: [PATCH] fix demux function types

---
 docs/modules/Demux.ts.md |  60 +++++++++++-----------
 src/Demux.ts             | 105 +++++++++++++++++++++------------------
 2 files changed, 88 insertions(+), 77 deletions(-)

diff --git a/docs/modules/Demux.ts.md b/docs/modules/Demux.ts.md
index 48e2b090..8686838a 100644
--- a/docs/modules/Demux.ts.md
+++ b/docs/modules/Demux.ts.md
@@ -71,17 +71,19 @@ the correct sink.
 **Signature**
 
 ```ts
-export declare const demuxMultiplexedSocket: (<E1, E2, E3>(
+export declare const demuxMultiplexedSocket: (<A1, A2, E1, E2, E3>(
   source: Stream.Stream<Uint8Array, E1, never>,
-  sink1: Sink.Sink<void, string | Uint8Array, never, E2, never>,
-  sink2: Sink.Sink<void, string | Uint8Array, never, E3, never>
-) => (socket: MultiplexedStreamSocket) => Effect.Effect<void, Socket.SocketError | E1 | E2 | E3, never>) &
-  (<E1, E2, E3>(
+  sink1: Sink.Sink<A1, string | Uint8Array, never, E2, never>,
+  sink2: Sink.Sink<A2, string | Uint8Array, never, E3, never>
+) => (
+  socket: MultiplexedStreamSocket
+) => Effect.Effect<readonly [stdout: A1, stderr: A2], Socket.SocketError | E1 | E2 | E3, never>) &
+  (<A1, A2, E1, E2, E3>(
     socket: MultiplexedStreamSocket,
     source: Stream.Stream<Uint8Array, E1, never>,
-    sink1: Sink.Sink<void, string | Uint8Array, never, E2, never>,
-    sink2: Sink.Sink<void, string | Uint8Array, never, E3, never>
-  ) => Effect.Effect<void, Socket.SocketError | E1 | E2 | E3, never>)
+    sink1: Sink.Sink<A1, string | Uint8Array, never, E2, never>,
+    sink2: Sink.Sink<A2, string | Uint8Array, never, E3, never>
+  ) => Effect.Effect<readonly [stdout: A1, stderr: A2], Socket.SocketError | E1 | E2 | E3, never>)
 ```
 
 Added in v1.0.0
@@ -97,15 +99,15 @@ combined on the same sink.
 **Signature**
 
 ```ts
-export declare const demuxRawSocket: (<E1, E2>(
+export declare const demuxRawSocket: (<A, E1, E2>(
   source: Stream.Stream<Uint8Array, E1, never>,
-  sink: Sink.Sink<void, string | Uint8Array, never, E2, never>
-) => (socket: RawStreamSocket) => Effect.Effect<void, Socket.SocketError | E1 | E2, never>) &
-  (<E1, E2>(
+  sink: Sink.Sink<A, string | Uint8Array, never, E2, never>
+) => (socket: RawStreamSocket) => Effect.Effect<A, Socket.SocketError | E1 | E2, never>) &
+  (<A, E1, E2>(
     socket: RawStreamSocket,
     source: Stream.Stream<Uint8Array, E1, never>,
-    sink: Sink.Sink<void, string | Uint8Array, never, E2, never>
-  ) => Effect.Effect<void, Socket.SocketError | E1 | E2, never>)
+    sink: Sink.Sink<A, string | Uint8Array, never, E2, never>
+  ) => Effect.Effect<A, Socket.SocketError | E1 | E2, never>)
 ```
 
 Added in v1.0.0
@@ -122,26 +124,28 @@ and one for stderr.
 
 ```ts
 export declare const demuxSocket: {
-  <E1, E2>(
+  <A1, E1, E2>(
     socket: RawStreamSocket,
     source: Stream.Stream<Uint8Array, E1, never>,
-    sink: Sink.Sink<void, string | Uint8Array, never, E2, never>
-  ): Effect.Effect<void, Socket.SocketError | E1 | E2, never>
-  <E1, E2>(
+    sink: Sink.Sink<A1, string | Uint8Array, never, E2, never>
+  ): Effect.Effect<A1, Socket.SocketError | E1 | E2, never>
+  <A1, E1, E2>(
     source: Stream.Stream<Uint8Array, E1, never>,
-    sink: Sink.Sink<void, string | Uint8Array, never, E2, never>
-  ): (socket: RawStreamSocket) => Effect.Effect<void, Socket.SocketError | E1 | E2, never>
-  <E1, E2, E3>(
+    sink: Sink.Sink<A1, string | Uint8Array, never, E2, never>
+  ): (socket: RawStreamSocket) => Effect.Effect<A1, Socket.SocketError | E1 | E2, never>
+  <A1, A2, E1, E2, E3>(
     socket: MultiplexedStreamSocket,
     source: Stream.Stream<Uint8Array, E1, never>,
-    sink1: Sink.Sink<void, string | Uint8Array, never, E2, never>,
-    sink2: Sink.Sink<void, string | Uint8Array, never, E3, never>
-  ): Effect.Effect<void, Socket.SocketError | E1 | E2 | E3, never>
-  <E1, E2, E3>(
+    sink1: Sink.Sink<A1, string | Uint8Array, never, E2, never>,
+    sink2: Sink.Sink<A2, string | Uint8Array, never, E3, never>
+  ): Effect.Effect<readonly [stdout: A1, stderr: A2], Socket.SocketError | E1 | E2 | E3, never>
+  <A1, A2, E1, E2, E3>(
     source: Stream.Stream<Uint8Array, E1, never>,
-    sink1: Sink.Sink<void, string | Uint8Array, never, E2, never>,
-    sink2: Sink.Sink<void, string | Uint8Array, never, E3, never>
-  ): (socket: MultiplexedStreamSocket) => Effect.Effect<void, Socket.SocketError | E1 | E2 | E3, never>
+    sink1: Sink.Sink<A1, string | Uint8Array, never, E2, never>,
+    sink2: Sink.Sink<A2, string | Uint8Array, never, E3, never>
+  ): (
+    socket: MultiplexedStreamSocket
+  ) => Effect.Effect<readonly [stdout: A1, stderr: A2], Socket.SocketError | E1 | E2 | E3, never>
 }
 ```
 
diff --git a/src/Demux.ts b/src/Demux.ts
index 47860950..fdb05d9a 100644
--- a/src/Demux.ts
+++ b/src/Demux.ts
@@ -15,6 +15,7 @@ import * as Effect from "effect/Effect";
 import * as Function from "effect/Function";
 import * as Sink from "effect/Sink";
 import * as Stream from "effect/Stream";
+import * as Tuple from "effect/Tuple";
 
 import { IExposeSocketOnEffectClientResponse } from "./Requests.js";
 
@@ -147,22 +148,22 @@ export enum MultiplexedStreamSocketHeaderType {
  * @see https://docs.docker.com/engine/api/v1.43/#tag/Container/operation/ContainerAttach
  */
 export const demuxRawSocket = Function.dual<
-    <E1, E2>(
+    <A, E1, E2>(
         source: Stream.Stream<Uint8Array, E1, never>,
-        sink: Sink.Sink<void, string | Uint8Array, never, E2, never>
-    ) => (socket: RawStreamSocket) => Effect.Effect<void, E1 | E2 | Socket.SocketError, never>,
-    <E1, E2>(
+        sink: Sink.Sink<A, string | Uint8Array, never, E2, never>
+    ) => (socket: RawStreamSocket) => Effect.Effect<A, E1 | E2 | Socket.SocketError, never>,
+    <A, E1, E2>(
         socket: RawStreamSocket,
         source: Stream.Stream<Uint8Array, E1, never>,
-        sink: Sink.Sink<void, string | Uint8Array, never, E2, never>
-    ) => Effect.Effect<void, E1 | E2 | Socket.SocketError, never>
+        sink: Sink.Sink<A, string | Uint8Array, never, E2, never>
+    ) => Effect.Effect<A, E1 | E2 | Socket.SocketError, never>
 >(
     3,
-    <E1, E2>(
+    <A, E1, E2>(
         socket: RawStreamSocket,
         source: Stream.Stream<Uint8Array, E1, never>,
-        sink: Sink.Sink<void, string | Uint8Array, never, E2, never>
-    ): Effect.Effect<void, E1 | E2 | Socket.SocketError, never> =>
+        sink: Sink.Sink<A, string | Uint8Array, never, E2, never>
+    ): Effect.Effect<A, E1 | E2 | Socket.SocketError, never> =>
         Function.pipe(source, Stream.pipeThroughChannel(Socket.toChannel(socket)), Stream.run(sink))
 );
 
@@ -178,25 +179,27 @@ export const demuxRawSocket = Function.dual<
  * @see https://docs.docker.com/engine/api/v1.43/#tag/Container/operation/ContainerAttach
  */
 export const demuxMultiplexedSocket = Function.dual<
-    <E1, E2, E3>(
+    <A1, A2, E1, E2, E3>(
         source: Stream.Stream<Uint8Array, E1, never>,
-        sink1: Sink.Sink<void, string | Uint8Array, never, E2, never>,
-        sink2: Sink.Sink<void, string | Uint8Array, never, E3, never>
-    ) => (socket: MultiplexedStreamSocket) => Effect.Effect<void, E1 | E2 | E3 | Socket.SocketError, never>,
-    <E1, E2, E3>(
+        sink1: Sink.Sink<A1, string | Uint8Array, never, E2, never>,
+        sink2: Sink.Sink<A2, string | Uint8Array, never, E3, never>
+    ) => (
+        socket: MultiplexedStreamSocket
+    ) => Effect.Effect<readonly [stdout: A1, stderr: A2], E1 | E2 | E3 | Socket.SocketError, never>,
+    <A1, A2, E1, E2, E3>(
         socket: MultiplexedStreamSocket,
         source: Stream.Stream<Uint8Array, E1, never>,
-        sink1: Sink.Sink<void, string | Uint8Array, never, E2, never>,
-        sink2: Sink.Sink<void, string | Uint8Array, never, E3, never>
-    ) => Effect.Effect<void, E1 | E2 | E3 | Socket.SocketError, never>
+        sink1: Sink.Sink<A1, string | Uint8Array, never, E2, never>,
+        sink2: Sink.Sink<A2, string | Uint8Array, never, E3, never>
+    ) => Effect.Effect<readonly [stdout: A1, stderr: A2], E1 | E2 | E3 | Socket.SocketError, never>
 >(
     4,
-    <E1, E2, E3>(
+    <A1, A2, E1, E2, E3>(
         socket: MultiplexedStreamSocket,
         source: Stream.Stream<Uint8Array, E1, never>,
-        sink1: Sink.Sink<void, string | Uint8Array, never, E2, never>,
-        sink2: Sink.Sink<void, string | Uint8Array, never, E3, never>
-    ): Effect.Effect<void, E1 | E2 | E3 | Socket.SocketError, never> =>
+        sink1: Sink.Sink<A1, string | Uint8Array, never, E2, never>,
+        sink2: Sink.Sink<A2, string | Uint8Array, never, E3, never>
+    ): Effect.Effect<readonly [stdout: A1, stderr: A2], E1 | E2 | E3 | Socket.SocketError, never> =>
         Function.pipe(
             source,
             Stream.pipeThroughChannel(Socket.toChannel(socket)),
@@ -221,16 +224,18 @@ export const demuxMultiplexedSocket = Function.dual<
                     type === MultiplexedStreamSocketHeaderType.Stderr
             ),
             Stream.partition(({ type }) => type === MultiplexedStreamSocketHeaderType.Stderr),
-            Effect.map(([stdout, stderr]) => [
-                stdout.pipe(
-                    Stream.map(({ contents }) => contents),
-                    Stream.run(sink1)
-                ),
-                stderr.pipe(
-                    Stream.map(({ contents }) => contents),
-                    Stream.run(sink2)
-                ),
-            ]),
+            Effect.map(
+                Tuple.mapBoth({
+                    onFirst: Function.compose(
+                        Stream.map(({ contents }) => contents),
+                        Stream.run(sink1)
+                    ),
+                    onSecond: Function.compose(
+                        Stream.map(({ contents }) => contents),
+                        Stream.run(sink2)
+                    ),
+                })
+            ),
             Effect.map(Effect.all),
             Effect.flatten,
             Effect.scoped
@@ -248,34 +253,36 @@ export const demuxMultiplexedSocket = Function.dual<
  * @category Demux
  */
 export const demuxSocket: {
-    <E1, E2>(
+    <A1, E1, E2>(
         socket: RawStreamSocket,
         source: Stream.Stream<Uint8Array, E1, never>,
-        sink: Sink.Sink<void, string | Uint8Array, never, E2, never>
-    ): Effect.Effect<void, E1 | E2 | Socket.SocketError, never>;
-    <E1, E2>(
+        sink: Sink.Sink<A1, string | Uint8Array, never, E2, never>
+    ): Effect.Effect<A1, E1 | E2 | Socket.SocketError, never>;
+    <A1, E1, E2>(
         source: Stream.Stream<Uint8Array, E1, never>,
-        sink: Sink.Sink<void, string | Uint8Array, never, E2, never>
-    ): (socket: RawStreamSocket) => Effect.Effect<void, E1 | E2 | Socket.SocketError, never>;
-    <E1, E2, E3>(
+        sink: Sink.Sink<A1, string | Uint8Array, never, E2, never>
+    ): (socket: RawStreamSocket) => Effect.Effect<A1, E1 | E2 | Socket.SocketError, never>;
+    <A1, A2, E1, E2, E3>(
         socket: MultiplexedStreamSocket,
         source: Stream.Stream<Uint8Array, E1, never>,
-        sink1: Sink.Sink<void, string | Uint8Array, never, E2, never>,
-        sink2: Sink.Sink<void, string | Uint8Array, never, E3, never>
-    ): Effect.Effect<void, E1 | E2 | E3 | Socket.SocketError, never>;
-    <E1, E2, E3>(
+        sink1: Sink.Sink<A1, string | Uint8Array, never, E2, never>,
+        sink2: Sink.Sink<A2, string | Uint8Array, never, E3, never>
+    ): Effect.Effect<readonly [stdout: A1, stderr: A2], E1 | E2 | E3 | Socket.SocketError, never>;
+    <A1, A2, E1, E2, E3>(
         source: Stream.Stream<Uint8Array, E1, never>,
-        sink1: Sink.Sink<void, string | Uint8Array, never, E2, never>,
-        sink2: Sink.Sink<void, string | Uint8Array, never, E3, never>
-    ): (socket: MultiplexedStreamSocket) => Effect.Effect<void, E1 | E2 | E3 | Socket.SocketError, never>;
+        sink1: Sink.Sink<A1, string | Uint8Array, never, E2, never>,
+        sink2: Sink.Sink<A2, string | Uint8Array, never, E3, never>
+    ): (
+        socket: MultiplexedStreamSocket
+    ) => Effect.Effect<readonly [stdout: A1, stderr: A2], E1 | E2 | E3 | Socket.SocketError, never>;
 } = Function.dual(
     (arguments_) => arguments_[0][Socket.TypeId],
-    <E1, E2, E3>(
+    <A1, A2, E1, E2, E3>(
         socket: RawStreamSocket | MultiplexedStreamSocket,
         source: Stream.Stream<Uint8Array, E1, never>,
-        sink1: Sink.Sink<void, string | Uint8Array, never, E2, never>,
-        sink2: Sink.Sink<void, string | Uint8Array, never, E3, never>
-    ): Effect.Effect<void, E1 | E2 | E3 | Socket.SocketError, never> => {
+        sink1: Sink.Sink<A1, string | Uint8Array, never, E2, never>,
+        sink2: Sink.Sink<A2, string | Uint8Array, never, E3, never>
+    ): Effect.Effect<A1 | readonly [stdout: A1, stderr: A2], E1 | E2 | E3 | Socket.SocketError, never> => {
         switch (socket["content-type"]) {
             case "application/vnd.docker.raw-stream": {
                 return demuxRawSocket(socket, source, sink1);