(
message: Message.Outgoing<any>,
discard: boolean,
retries?: number
)
| 1088 | }) |
| 1089 | |
| 1090 | function sendOutgoing( |
| 1091 | message: Message.Outgoing<any>, |
| 1092 | discard: boolean, |
| 1093 | retries?: number |
| 1094 | ): Effect.Effect< |
| 1095 | void, |
| 1096 | MailboxFull | AlreadyProcessingMessage | PersistenceError | EntityNotAssignedToRunner |
| 1097 | > { |
| 1098 | const isPersisted = Context.get( |
| 1099 | message._tag === "OutgoingRequest" ? message.annotations : message.rpc.annotations, |
| 1100 | Persisted |
| 1101 | ) |
| 1102 | const shouldFail = !discard && |
| 1103 | (message._tag === "OutgoingRequest" || message.envelope._tag === "AckChunk") |
| 1104 | const abandon = (error: EntityNotAssignedToRunner) => { |
| 1105 | if (!isPersisted) { |
| 1106 | return shouldFail |
| 1107 | ? Effect.fail(error) |
| 1108 | : Effect.logDebug("Abandoning outgoing message during shutdown", message.envelope.address) |
| 1109 | } |
| 1110 | const persist = message._tag === "OutgoingRequest" |
| 1111 | ? storage.saveRequest(message) |
| 1112 | : storage.saveEnvelope(message) |
| 1113 | return Effect.catchTag(persist, "MalformedMessage", Effect.die).pipe( |
| 1114 | Effect.andThen( |
| 1115 | shouldFail |
| 1116 | ? Effect.fail(error) |
| 1117 | : Effect.logWarning("Persisting outgoing message abandoned during shutdown", message.envelope.address) |
| 1118 | ) |
| 1119 | ) |
| 1120 | } |
| 1121 | return Effect.catchFilter( |
| 1122 | Effect.suspend(() => { |
| 1123 | const address = message.envelope.address |
| 1124 | if (isPersisted && !storageEnabled) { |
| 1125 | return Effect.die("Sharding.sendOutgoing: Persisted messages require MessageStorage") |
| 1126 | } |
| 1127 | if (shouldFail && MutableRef.get(isShutdown)) { |
| 1128 | return Effect.fail(new EntityNotAssignedToRunner({ address })) |
| 1129 | } |
| 1130 | const maybeRunner = MutableHashMap.get(shardAssignments, address.shardId) |
| 1131 | const runnerIsLocal = Option.isSome(maybeRunner) && isLocalRunner(maybeRunner.value) |
| 1132 | if (isPersisted) { |
| 1133 | return runnerIsLocal |
| 1134 | ? notifyLocal(message, discard) |
| 1135 | : runnersService.notify({ address: maybeRunner, message, discard }) |
| 1136 | } else if (Option.isNone(maybeRunner)) { |
| 1137 | return Effect.fail(new EntityNotAssignedToRunner({ address })) |
| 1138 | } |
| 1139 | return runnerIsLocal |
| 1140 | ? sendLocal(message) |
| 1141 | : discard |
| 1142 | ? runnersService.notify({ address: maybeRunner, message, discard }) |
| 1143 | : runnersService.send({ address: maybeRunner.value, message }) |
| 1144 | }), |
| 1145 | (error) => |
| 1146 | error._tag === "EntityNotAssignedToRunner" || error._tag === "RunnerUnavailable" |
| 1147 | ? Result.succeed(error) |
no test coverage detected