| 1 | /** |
| 2 | * Node TCP socket transport for Effect Cluster runner-to-runner RPC. |
| 3 | * |
| 4 | * This module provides the shared Node layers used by socket-based cluster |
| 5 | * transports. `layerClientProtocol` opens TCP sockets to peer runner addresses |
| 6 | * and wraps them in the current RPC serialization protocol. `layerSocketServer` |
| 7 | * exposes the socket server that receives incoming runner RPC traffic. |
| 8 | * |
| 9 | * @since 4.0.0 |
| 10 | */ |
| 11 | import * as Runners from "effect/cluster/Runners" |
| 12 | import * as ShardingConfig from "effect/cluster/ShardingConfig" |
| 13 | import * as Effect from "effect/Effect" |
| 14 | import * as Layer from "effect/Layer" |
| 15 | import * as Option from "effect/Option" |
| 16 | import * as RpcClient from "effect/rpc/RpcClient" |
| 17 | import * as RpcSerialization from "effect/rpc/RpcSerialization" |
| 18 | import { Socket } from "effect/socket/Socket" |
| 19 | import type * as SocketServer from "effect/socket/SocketServer" |
| 20 | import * as NodeSocket from "./NodeSocket.ts" |
| 21 | import * as NodeSocketServer from "./NodeSocketServer.ts" |
| 22 | |
| 23 | /** |
| 24 | * Provides the cluster `RpcClientProtocol` by opening TCP sockets to runner |
| 25 | * addresses and using the current RPC serialization service. |
| 26 | * |
| 27 | * @category layers |
| 28 | * @since 4.0.0 |
| 29 | */ |
| 30 | export const layerClientProtocol: Layer.Layer< |
| 31 | Runners.RpcClientProtocol, |
| 32 | never, |
| 33 | RpcSerialization.RpcSerialization |
| 34 | > = Layer.effect(Runners.RpcClientProtocol)( |
| 35 | Effect.gen(function*() { |
| 36 | const serialization = yield* RpcSerialization.RpcSerialization |
| 37 | return { |
| 38 | codecFor: serialization.codecFor, |
| 39 | make: Effect.fnUntraced(function*(address) { |
| 40 | const socket = yield* NodeSocket.makeNet({ |
| 41 | openTimeout: 1000, |
| 42 | timeout: 5500, |
| 43 | host: address.host, |
| 44 | port: address.port |
| 45 | }) |
| 46 | return yield* RpcClient.makeProtocolSocket().pipe( |
| 47 | Effect.provideService(Socket, socket), |
| 48 | Effect.provideService(RpcSerialization.RpcSerialization, serialization) |
| 49 | ) |
| 50 | }, Effect.orDie) |
| 51 | } |
| 52 | }) |
| 53 | ) |
| 54 | |
| 55 | /** |
| 56 | * Provides the socket server used by cluster runners, listening on |
| 57 | * `ShardingConfig.runnerListenAddress` or `runnerAddress`. |
| 58 | * |
| 59 | * @category layers |
| 60 | * @since 4.0.0 |
| 61 | */ |
| 62 | export const layerSocketServer: Layer.Layer< |
| 63 | SocketServer.SocketServer, |
| 64 | SocketServer.SocketServerError, |
| 65 | ShardingConfig.ShardingConfig |
| 66 | > = Effect.gen(function*() { |
| 67 | const config = yield* ShardingConfig.ShardingConfig |
| 68 | const listenAddress = Option.orElse(config.runnerListenAddress, () => config.runnerAddress) |
| 69 | if (Option.isNone(listenAddress)) { |
| 70 | return yield* Effect.die("layerSocketServer: ShardingConfig.runnerListenAddress is None") |
| 71 | } |
| 72 | return NodeSocketServer.layer(listenAddress.value) |
| 73 | }).pipe(Layer.unwrap) |