import { WsRpcGroup } from "@t3tools/contracts";import { Duration, Effect, Layer, Schedule } from "effect";import { RpcClient, RpcSerialization } from "effect/unstable/rpc";import * as Socket from "effect/unstable/socket/Socket"; import { resolveServerUrl } from "../lib/utils";import { acknowledgeRpcRequest, clearAllTrackedRpcRequests, trackRpcRequestSent,} from "./requestLatencyState";import { getWsReconnectDelayMsForRetry, recordWsConnectionAttempt, recordWsConnectionClosed, recordWsConnectionErrored, recordWsConnectionOpened, WS_RECONNECT_MAX_RETRIES,} from "./wsConnectionState"; export const makeWsRpcProtocolClient = RpcClient.make(WsRpcGroup); type RpcClientFactory = typeof makeWsRpcProtocolClient;export type WsRpcProtocolClient = RpcClientFactory extends Effect.Effect<infer Client, any, any> ? Client : never; export function createWsRpcProtocolLayer(url?: string) { const resolvedUrl = resolveServerUrl({ url, protocol: window.location.protocol === "https:" ? "wss" : "ws", pathname: "/ws", }); const trackingWebSocketConstructorLayer = Layer.succeed( Socket.WebSocketConstructor, (socketUrl, protocols) => { recordWsConnectionAttempt(socketUrl); const socket = new globalThis.WebSocket(socketUrl, protocols); socket.addEventListener( "open", () => { recordWsConnectionOpened(); }, { once: true }, ); socket.addEventListener( "error", () => { clearAllTrackedRpcRequests(); recordWsConnectionErrored("Unable to connect to the T3 server WebSocket."); }, { once: true }, ); socket.addEventListener( "close", (event) => { clearAllTrackedRpcRequests(); recordWsConnectionClosed({ code: event.code, reason: event.reason, }); }, { once: true }, ); return socket; }, ); const socketLayer = Socket.layerWebSocket(resolvedUrl).pipe( Layer.provide(trackingWebSocketConstructorLayer), ); const retryPolicy = Schedule.addDelay(Schedule.recurs(WS_RECONNECT_MAX_RETRIES), (retryCount) => Effect.succeed(Duration.millis(getWsReconnectDelayMsForRetry(retryCount) ?? 0)), ); const protocolLayer = Layer.effect( RpcClient.Protocol, Effect.map( RpcClient.makeProtocolSocket({ retryPolicy, retryTransientErrors: true, }), (protocol) => ({ ...protocol, run: (writeResponse) => protocol.run((response) => { if (response._tag === "Chunk" || response._tag === "Exit") { acknowledgeRpcRequest(response.requestId); } else if (response._tag === "ClientProtocolError" || response._tag === "Defect") { clearAllTrackedRpcRequests(); } return writeResponse(response); }), send: (request, transferables) => { if (request._tag === "Request") { trackRpcRequestSent(request.id, request.tag); } return protocol.send(request, transferables); }, }), ), ); return protocolLayer.pipe(Layer.provide(Layer.mergeAll(socketLayer, RpcSerialization.layerJson)));}
import { WsRpcGroup } from "@t3tools/contracts";
import { Duration, Effect, Layer, Schedule } from "effect";
import { RpcClient, RpcSerialization } from "effect/unstable/rpc";
import * as Socket from "effect/unstable/socket/Socket";
import { resolveServerUrl } from "../lib/utils";
import {
acknowledgeRpcRequest,
clearAllTrackedRpcRequests,
trackRpcRequestSent,
} from "./requestLatencyState";
import {
getWsReconnectDelayMsForRetry,
recordWsConnectionAttempt,
recordWsConnectionClosed,
recordWsConnectionErrored,
recordWsConnectionOpened,
WS_RECONNECT_MAX_RETRIES,
} from "./wsConnectionState";
export const makeWsRpcProtocolClient = RpcClient.make(WsRpcGroup);
type RpcClientFactory = typeof makeWsRpcProtocolClient;
export type WsRpcProtocolClient =
RpcClientFactory extends Effect.Effect<infer Client, any, any> ? Client : never;
export function createWsRpcProtocolLayer(url?: string) {
const resolvedUrl = resolveServerUrl({
url,
protocol: window.location.protocol === "https:" ? "wss" : "ws",
pathname: "/ws",
});
const trackingWebSocketConstructorLayer = Layer.succeed(
Socket.WebSocketConstructor,
(socketUrl, protocols) => {
recordWsConnectionAttempt(socketUrl);
const socket = new globalThis.WebSocket(socketUrl, protocols);
socket.addEventListener(
"open",
() => {
recordWsConnectionOpened();
},
{ once: true },
);
socket.addEventListener(
"error",
() => {
clearAllTrackedRpcRequests();
recordWsConnectionErrored("Unable to connect to the T3 server WebSocket.");
},
{ once: true },
);
socket.addEventListener(
"close",
(event) => {
clearAllTrackedRpcRequests();
recordWsConnectionClosed({
code: event.code,
reason: event.reason,
});
},
{ once: true },
);
return socket;
},
);
const socketLayer = Socket.layerWebSocket(resolvedUrl).pipe(
Layer.provide(trackingWebSocketConstructorLayer),
);
const retryPolicy = Schedule.addDelay(Schedule.recurs(WS_RECONNECT_MAX_RETRIES), (retryCount) =>
Effect.succeed(Duration.millis(getWsReconnectDelayMsForRetry(retryCount) ?? 0)),
);
const protocolLayer = Layer.effect(
RpcClient.Protocol,
Effect.map(
RpcClient.makeProtocolSocket({
retryPolicy,
retryTransientErrors: true,
}),
(protocol) => ({
...protocol,
run: (writeResponse) =>
protocol.run((response) => {
if (response._tag === "Chunk" || response._tag === "Exit") {
acknowledgeRpcRequest(response.requestId);
} else if (response._tag === "ClientProtocolError" || response._tag === "Defect") {
clearAllTrackedRpcRequests();
}
return writeResponse(response);
}),
send: (request, transferables) => {
if (request._tag === "Request") {
trackRpcRequestSent(request.id, request.tag);
}
return protocol.send(request, transferables);
},
}),
),
);
return protocolLayer.pipe(Layer.provide(Layer.mergeAll(socketLayer, RpcSerialization.layerJson)));
}