diff --git a/dist/unstable/rpc/RpcClient.d.ts b/dist/unstable/rpc/RpcClient.d.tsindex b0f61df..ead663e 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";@@ -260,9 +261,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; }>; /** * Represents optional client protocol hooks that run when a transport connects@@ -279,4 +311,4 @@ declare const ConnectionHooks_base: Context.ServiceClass<ConnectionHooks, "effec export declare class ConnectionHooks extends ConnectionHooks_base { } export {};-//# sourceMappingURL=RpcClient.d.ts.map\ No newline at end of file+//# sourceMappingURL=RpcClient.d.ts.mapdiff --git a/dist/unstable/rpc/RpcClient.js b/dist/unstable/rpc/RpcClient.jsindex 811a8a8..f9cf2eb 100644--- a/dist/unstable/rpc/RpcClient.js+++ b/dist/unstable/rpc/RpcClient.js@@ -46,6 +46,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;@@ -83,6 +84,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,@@ -100,7 +106,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 => {@@ -118,7 +124,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 => {@@ -127,8 +133,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) {@@ -140,11 +147,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, {@@ -152,9 +165,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@@ -169,7 +183,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@@ -199,7 +213,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: {@@ -213,7 +230,7 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro fiber.addObserver(() => { resume(Effect.void); });- });+ }))); const write = message => { switch (message._tag) { case "Chunk":@@ -221,7 +238,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@@ -236,11 +259,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": {@@ -568,7 +598,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;@@ -588,8 +618,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);@@ -617,12 +646,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@@ -664,20 +693,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,@@ -820,6 +849,11 @@ export const makeProtocolWorker = options => Protocol.make(Effect.fnUntraced(fun * @since 4.0.0 */ 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") {} /** * Represents optional client protocol hooks that run when a transport connects * and disconnects.@@ -835,4 +869,4 @@ export const layerProtocolWorker = /*#__PURE__*/flow(makeProtocolWorker, /*#__PU export class ConnectionHooks extends /*#__PURE__*/Context.Service()("effect/rpc/RpcClient/ConnectionHooks") {} // internal const decodeDefect = /*#__PURE__*/Schema.decodeSync(Schema.Defect);-//# sourceMappingURL=RpcClient.js.map\ No newline at end of file+//# sourceMappingURL=RpcClient.js.map
diff --git a/dist/unstable/rpc/RpcClient.d.ts b/dist/unstable/rpc/RpcClient.d.ts
index b0f61df..ead663e 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";
@@ -260,9 +261,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;
}>;
/**
* Represents optional client protocol hooks that run when a transport connects
@@ -279,4 +311,4 @@ declare const ConnectionHooks_base: Context.ServiceClass<ConnectionHooks, "effec
export declare class ConnectionHooks extends ConnectionHooks_base {
}
export {};
-//# sourceMappingURL=RpcClient.d.ts.map
\ No newline at end of file
+//# sourceMappingURL=RpcClient.d.ts.map
diff --git a/dist/unstable/rpc/RpcClient.js b/dist/unstable/rpc/RpcClient.js
index 811a8a8..f9cf2eb 100644
--- a/dist/unstable/rpc/RpcClient.js
+++ b/dist/unstable/rpc/RpcClient.js
@@ -46,6 +46,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;
@@ -83,6 +84,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,
@@ -100,7 +106,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 => {
@@ -118,7 +124,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 => {
@@ -127,8 +133,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) {
@@ -140,11 +147,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, {
@@ -152,9 +165,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
@@ -169,7 +183,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
@@ -199,7 +213,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: {
@@ -213,7 +230,7 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro
fiber.addObserver(() => {
resume(Effect.void);
});
- });
+ })));
const write = message => {
switch (message._tag) {
case "Chunk":
@@ -221,7 +238,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
@@ -236,11 +259,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":
{
@@ -568,7 +598,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;
@@ -588,8 +618,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);
@@ -617,12 +646,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
@@ -664,20 +693,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,
@@ -820,6 +849,11 @@ export const makeProtocolWorker = options => Protocol.make(Effect.fnUntraced(fun
* @since 4.0.0
*/
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") {}
/**
* Represents optional client protocol hooks that run when a transport connects
* and disconnects.
@@ -835,4 +869,4 @@ export const layerProtocolWorker = /*#__PURE__*/flow(makeProtocolWorker, /*#__PU
export class ConnectionHooks extends /*#__PURE__*/Context.Service()("effect/rpc/RpcClient/ConnectionHooks") {}
// internal
const decodeDefect = /*#__PURE__*/Schema.decodeSync(Schema.Defect);
-//# sourceMappingURL=RpcClient.js.map
\ No newline at end of file
+//# sourceMappingURL=RpcClient.js.map