(options)
| 1335 | generateRequestId: () => snowflakeGen.nextUnsafe() as any, |
| 1336 | flatten: true, |
| 1337 | onFromClient(options): Effect.Effect< |
| 1338 | void, |
| 1339 | MailboxFull | AlreadyProcessingMessage | PersistenceError | EntityNotAssignedToRunner |
| 1340 | > { |
| 1341 | const address = Context.getUnsafe(options.context, ClientAddressTag) |
| 1342 | switch (options.message._tag) { |
| 1343 | case "Request": { |
| 1344 | const fiber = Fiber.getCurrent()! |
| 1345 | const id = Snowflake.Snowflake(options.message.id) |
| 1346 | const rpc = entity.protocol.requests.get(options.message.tag)! |
| 1347 | let respond: (reply: Reply.Reply<any>) => Effect.Effect<void> |
| 1348 | const envelope = Envelope.makeRequest<any>({ |
| 1349 | requestId: id, |
| 1350 | address, |
| 1351 | tag: options.message.tag, |
| 1352 | payload: options.message.payload, |
| 1353 | headers: options.message.headers, |
| 1354 | traceId: options.message.traceId, |
| 1355 | spanId: options.message.spanId, |
| 1356 | sampled: options.message.sampled |
| 1357 | }) |
| 1358 | const message = new Message.OutgoingRequest({ |
| 1359 | envelope, |
| 1360 | lastReceivedReply: Option.none(), |
| 1361 | rpc, |
| 1362 | context: fiber.context as Context.Context<any>, |
| 1363 | respond: (reply) => respond(reply), |
| 1364 | annotations: Context.get(rpc.annotations, ClusterSchema.Dynamic)( |
| 1365 | rpc.annotations, |
| 1366 | envelope as any |
| 1367 | ) |
| 1368 | }) |
| 1369 | if (!options.discard) { |
| 1370 | const entry: ClientRequestEntry = { |
| 1371 | rpc: rpc as any, |
| 1372 | context: fiber.context, |
| 1373 | message |
| 1374 | } |
| 1375 | clientRequests.set(id, entry) |
| 1376 | respond = makeClientRespond(entry, client.write) |
| 1377 | } else { |
| 1378 | respond = clientRespondDiscard |
| 1379 | } |
| 1380 | return sendOutgoing(message, options.discard) |
| 1381 | } |
| 1382 | case "Ack": { |
| 1383 | const requestId = Snowflake.Snowflake(options.message.requestId) |
| 1384 | const entry = clientRequests.get(requestId) |
| 1385 | if (!entry) return Effect.void |
| 1386 | return sendOutgoing( |
| 1387 | new Message.OutgoingEnvelope({ |
| 1388 | envelope: new Envelope.AckChunk({ |
| 1389 | id: snowflakeGen.nextUnsafe(), |
| 1390 | address, |
| 1391 | requestId, |
| 1392 | replyId: entry.lastChunkId! |
| 1393 | }), |
| 1394 | rpc: entry.rpc |
nothing calls this directly
no test coverage detected