diff --git a/dist/unstable/rpc/RpcClient.d.ts b/dist/unstable/rpc/RpcClient.d.tsindex 7c5970042df67a25b4145df017960b3e6febed9c..7af3374632d53ad59f24e80835aa6741853b1f2b 100644--- a/dist/unstable/rpc/RpcClient.d.ts+++ b/dist/unstable/rpc/RpcClient.d.ts@@ -2,6 +2,7 @@ import * as Cause from "../../Cause.ts"; import * as Context from "../../Context.ts"; import type * as Duration from "../../Duration.ts"; import * as Effect from "../../Effect.ts";+import * as Exit from "../../Exit.ts"; import * as Layer from "../../Layer.ts"; import * as Queue from "../../Queue.ts"; import * as Schedule from "../../Schedule.ts";@@ -192,9 +193,40 @@ export declare const layerProtocolWorker: (options: {     readonly targetUtilization?: number | undefined;     readonly timeToLive: Duration.Input; }) => Layer.Layer<Protocol, WorkerError, Worker.WorkerPlatform | Worker.Spawner>;+declare const RequestHooks_base: Context.ServiceClass<RequestHooks, "effect/rpc/RpcClient/RequestHooks", {+    readonly onRequestStart?: ((info: {+        readonly id: RequestId;+        readonly tag: string;+        readonly stream: boolean;+    }) => Effect.Effect<void>) | undefined;+    readonly onRequestChunk?: ((info: {+        readonly id: RequestId;+        readonly tag: string;+        readonly chunkCount: number;+    }) => Effect.Effect<void>) | undefined;+    readonly onRequestExit?: ((info: {+        readonly id: RequestId;+        readonly tag: string;+        readonly stream: boolean;+        readonly exit: Exit.Exit<unknown, unknown>;+    }) => Effect.Effect<void>) | undefined;+    readonly onRequestInterrupt?: ((info: {+        readonly id: RequestId;+        readonly tag?: string | undefined;+    }) => Effect.Effect<void>) | undefined;+}>;+/**+ * @since 4.0.0+ * @category RequestHooks+ */+export declare class RequestHooks extends RequestHooks_base {+} declare const ConnectionHooks_base: Context.ServiceClass<ConnectionHooks, "effect/rpc/RpcClient/ConnectionHooks", {     readonly onConnect: Effect.Effect<void>;     readonly onDisconnect: Effect.Effect<void>;+    readonly onPing?: Effect.Effect<void> | undefined;+    readonly onPong?: Effect.Effect<void> | undefined;+    readonly onPingTimeout?: Effect.Effect<void> | undefined; }>; /**  * @since 4.0.0diff --git a/dist/unstable/rpc/RpcClient.js b/dist/unstable/rpc/RpcClient.jsindex 951dfc2d667a9136cfa714dcd3195e6575917fde..599c0345873bedd72bc93afe80eab0c676319421 100644--- a/dist/unstable/rpc/RpcClient.js+++ b/dist/unstable/rpc/RpcClient.js@@ -42,6 +42,7 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro   const services = yield* Effect.context();   const scope = Context.get(services, Scope.Scope);   const entries = new Map();+  const hooks = yield* Effect.serviceOption(RequestHooks);   let isShutdown = false;   yield* Scope.addFinalizer(scope, Effect.withFiber(parent => {     isShutdown = true;@@ -79,6 +80,11 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro       return Effect.interrupt;     }     const id = generateRequestId();+    const onStart = Option.isSome(hooks) && hooks.value.onRequestStart ? hooks.value.onRequestStart({+      id,+      tag: rpc._tag,+      stream: false+    }) : Effect.void;     const send = middleware(message => options.onFromClient({       message,       context,@@ -96,7 +102,7 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro       headers: Headers.merge(parentFiber.getRef(CurrentHeaders), headers)     });     if (discard) {-      return send;+      return onStart.pipe(Effect.andThen(send));     }     let fiber;     return Effect.onInterrupt(Effect.callback(resume => {@@ -114,7 +120,7 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro         }       };       entries.set(id, entry);-      fiber = send.pipe(span ? Effect.withParentSpan(span, {+      fiber = onStart.pipe(Effect.andThen(send), span ? Effect.withParentSpan(span, {         captureStackTrace: false       }) : identity, Effect.runForkWith(parentFiber.context));       fiber.addObserver(exit => {@@ -123,8 +129,9 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro         }       });     }), interruptors => {+      const entry = entries.get(id);       entries.delete(id);-      return Effect.andThen(Fiber.interrupt(fiber), sendInterrupt(id, Array.from(interruptors), context));+      return Effect.andThen(Fiber.interrupt(fiber), sendInterrupt(id, Array.from(interruptors), context, entry?.rpc._tag));     });   });   const onStreamRequest = Effect.fnUntraced(function* (rpc, middleware, payload, headers, streamBufferSize, context) {@@ -136,11 +143,17 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro     });     const fiber = Fiber.getCurrent();     const id = generateRequestId();+    const onStart = Option.isSome(hooks) && hooks.value.onRequestStart ? hooks.value.onRequestStart({+      id,+      tag: rpc._tag,+      stream: true+    }) : Effect.void;     const scope = Context.getUnsafe(fiber.context, Scope.Scope);     yield* Scope.addFinalizerExit(scope, exit => {-      if (!entries.has(id)) return Effect.void;+      const entry = entries.get(id);+      if (!entry) return Effect.void;       entries.delete(id);-      return sendInterrupt(id, Exit.isFailure(exit) ? Array.from(Cause.interruptors(exit.cause)) : [], context);+      return sendInterrupt(id, Exit.isFailure(exit) ? Array.from(Cause.interruptors(exit.cause)) : [], context, entry.rpc._tag);     });     const queue = yield* Queue.bounded(streamBufferSize);     entries.set(id, {@@ -148,9 +161,10 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro       rpc,       queue,       scope,-      context+      context,+      chunkCount: 0     });-    yield* middleware(message => options.onFromClient({+    yield* onStart.pipe(Effect.andThen(middleware(message => options.onFromClient({       message,       context,       discard: false@@ -165,7 +179,7 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro         sampled: span.sampled       } : {}),       headers: Headers.merge(fiber.getRef(CurrentHeaders), headers)-    }).pipe(span ? Effect.withParentSpan(span, {+    }))).pipe(span ? Effect.withParentSpan(span, {       captureStackTrace: false     }) : identity, Effect.catchCause(error => Queue.failCause(queue, error)), Effect.interruptible, Effect.forkIn(scope, {       startImmediately: true@@ -195,7 +209,10 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro       });     };   };-  const sendInterrupt = (requestId, interruptors, context) => Effect.callback(resume => {+  const sendInterrupt = (requestId, interruptors, context, tag) => (Option.isSome(hooks) && hooks.value.onRequestInterrupt ? hooks.value.onRequestInterrupt({+    id: requestId,+    tag+  }) : Effect.void).pipe(Effect.andThen(Effect.callback(resume => {     const parentFiber = Fiber.getCurrent();     const fiber = options.onFromClient({       message: {@@ -209,7 +226,7 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro     fiber.addObserver(() => {       resume(Effect.void);     });-  });+  })));   const write = message => {     switch (message._tag) {       case "Chunk":@@ -217,7 +234,13 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro           const requestId = message.requestId;           const entry = entries.get(requestId);           if (!entry || entry._tag !== "Queue") return Effect.void;-          return Queue.offerAll(entry.queue, message.values).pipe(supportsAck ? Effect.flatMap(() => options.onFromClient({+          const chunkCount = entry.chunkCount += message.values.length;+          const onChunk = Option.isSome(hooks) && hooks.value.onRequestChunk ? hooks.value.onRequestChunk({+            id: requestId,+            tag: entry.rpc._tag,+            chunkCount+          }) : Effect.void;+          return Queue.offerAll(entry.queue, message.values).pipe(Effect.andThen(onChunk), supportsAck ? Effect.flatMap(() => options.onFromClient({             message: {               _tag: "Ack",               requestId: message.requestId@@ -232,11 +255,18 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro           const entry = entries.get(requestId);           if (!entry) return Effect.void;           entries.delete(requestId);+          const onExit = Option.isSome(hooks) && hooks.value.onRequestExit ? hooks.value.onRequestExit({+            id: requestId,+            tag: entry.rpc._tag,+            stream: entry._tag === "Queue",+            exit: message.exit+          }) : Effect.void;           if (entry._tag === "Effect") {-            entry.resume(message.exit);-            return Effect.void;+            return onExit.pipe(Effect.andThen(Effect.sync(() => {+              entry.resume(message.exit);+            })));           }-          return message.exit._tag === "Success" ? Queue.end(entry.queue) : Queue.failCause(entry.queue, message.exit.cause);+          return onExit.pipe(Effect.andThen(message.exit._tag === "Success" ? Queue.end(entry.queue) : Queue.failCause(entry.queue, message.exit.cause)));         }       case "Defect":         {@@ -530,7 +560,7 @@ export const makeProtocolSocket = options => Protocol.make(Effect.fnUntraced(fun   const requestClientMap = new Map();   const write = yield* socket.writer;   let parser = serialization.makeUnsafe();-  const pinger = yield* makePinger(write(parser.encode(constPing)));+  const pinger = yield* makePinger(write(parser.encode(constPing)), Option.isSome(hooks) ? hooks.value : undefined);   let currentError;   const onOpen = Effect.suspend(() => {     currentError = undefined;@@ -550,8 +580,7 @@ export const makeProtocolSocket = options => Protocol.make(Effect.fnUntraced(fun           body: () => {             const response = responses[i++];             if (response._tag === "Pong") {-              pinger.onPong();-              return Effect.void;+              return pinger.onPong;             }             if ("requestId" in response) {               const clientId = requestClientMap.get(response.requestId);@@ -579,12 +608,12 @@ export const makeProtocolSocket = options => Protocol.make(Effect.fnUntraced(fun       }     }, {       onOpen-    }).pipe(Effect.raceFirst(Effect.flatMap(pinger.timeout, () => Effect.fail(new Socket.SocketError({+    }).pipe(Effect.raceFirst(Effect.flatMap(pinger.timeout, () => (Option.isSome(hooks) && hooks.value.onPingTimeout ? hooks.value.onPingTimeout : Effect.void).pipe(Effect.andThen(Effect.fail(new Socket.SocketError({       reason: new Socket.SocketOpenError({         kind: "Timeout",         cause: new Error("ping timeout")       })-    })))));+    })))))));   }).pipe(Effect.flatMap(() => Effect.fail(new Socket.SocketError({     reason: new Socket.SocketCloseError({       code: 1000@@ -626,20 +655,20 @@ export const makeProtocolSocket = options => Protocol.make(Effect.fnUntraced(fun   }; })); const defaultRetryPolicy = /*#__PURE__*/Schedule.exponential(500, 1.5).pipe(/*#__PURE__*/Schedule.either(/*#__PURE__*/Schedule.spaced(5000)));-const makePinger = /*#__PURE__*/Effect.fnUntraced(function* (writePing) {+const makePinger = /*#__PURE__*/Effect.fnUntraced(function* (writePing, hooks) {   let recievedPong = true;   const latch = Latch.makeUnsafe();   const reset = () => {     recievedPong = true;     latch.closeUnsafe();   };-  const onPong = () => {+  const onPong = Effect.sync(() => {     recievedPong = true;-  };+  }).pipe(Effect.andThen(hooks?.onPong ?? Effect.void));   yield* Effect.suspend(() => {     if (!recievedPong) return latch.open;     recievedPong = false;-    return writePing;+    return (hooks?.onPing ?? Effect.void).pipe(Effect.andThen(writePing));   }).pipe(Effect.delay("5 seconds"), Effect.ignore, Effect.forever, Effect.interruptible, Effect.forkScoped);   return {     timeout: latch.await,@@ -773,6 +802,11 @@ export const makeProtocolWorker = options => Protocol.make(Effect.fnUntraced(fun  * @category protocol  */ export const layerProtocolWorker = /*#__PURE__*/flow(makeProtocolWorker, /*#__PURE__*/Layer.effect(Protocol));+/**+ * @since 4.0.0+ * @category RequestHooks+ */+export class RequestHooks extends /*#__PURE__*/Context.Service()("effect/rpc/RpcClient/RequestHooks") {} /**  * @since 4.0.0  * @category ConnectionHooks