(
message: M,
discard: boolean,
options?: {
readonly waitUntilRead?: boolean | undefined
}
)
| 761 | } |
| 762 | const pendingNotifications = new Map<Snowflake.Snowflake, PendingNotification>() |
| 763 | const notifyLocal = <M extends Message.Outgoing<any> | Message.Incoming<any>>( |
| 764 | message: M, |
| 765 | discard: boolean, |
| 766 | options?: { |
| 767 | readonly waitUntilRead?: boolean | undefined |
| 768 | } |
| 769 | ) => |
| 770 | Effect.suspend(function loop(): Effect.Effect< |
| 771 | void, |
| 772 | | EntityNotAssignedToRunner |
| 773 | | AlreadyProcessingMessage |
| 774 | | (M extends Message.Incoming<any> ? never : PersistenceError) |
| 775 | > { |
| 776 | const address = message.envelope.address |
| 777 | const state = entityManagers.get(address.entityType) |
| 778 | if (!state) { |
| 779 | return Effect.flatMap(waitForEntityManager(address.entityType), loop) |
| 780 | } else if (state.status === "closed" || !isEntityOnLocalShards(address)) { |
| 781 | return Effect.fail(new EntityNotAssignedToRunner({ address })) |
| 782 | } |
| 783 | |
| 784 | const isLocal = isEntityOnLocalShards(address) |
| 785 | const notify = storageEnabled |
| 786 | ? openStorageReadLatch |
| 787 | : () => Effect.die("Sharding.notifyLocal: storage is disabled") |
| 788 | |
| 789 | if (message._tag === "IncomingRequest" || message._tag === "IncomingEnvelope") { |
| 790 | if (!isLocal) { |
| 791 | return Effect.fail(new EntityNotAssignedToRunner({ address })) |
| 792 | } else if ( |
| 793 | message._tag === "IncomingRequest" && state.manager.isProcessingFor(message, { excludeReplies: true }) |
| 794 | ) { |
| 795 | return Effect.fail(new AlreadyProcessingMessage({ address, envelopeId: message.envelope.requestId })) |
| 796 | } else if (message._tag === "IncomingRequest" && options?.waitUntilRead) { |
| 797 | if (!storageEnabled) return notify() |
| 798 | return Effect.async<void, EntityNotAssignedToRunner>((resume) => { |
| 799 | let entry = pendingNotifications.get(message.envelope.requestId) |
| 800 | if (entry) { |
| 801 | const prevResume = entry.resume |
| 802 | entry.resume = (effect) => { |
| 803 | prevResume(effect) |
| 804 | resume(effect) |
| 805 | } |
| 806 | return |
| 807 | } |
| 808 | entry = { resume, message } |
| 809 | pendingNotifications.set(message.envelope.requestId, entry) |
| 810 | storageReadLatch.unsafeOpen() |
| 811 | }) |
| 812 | } |
| 813 | return notify() |
| 814 | } |
| 815 | |
| 816 | return runnersService.notifyLocal({ message, notify, discard, storageOnly: !isLocal }) as any |
| 817 | }) |
| 818 | |
| 819 | function sendOutgoing( |
| 820 | message: Message.Outgoing<any>, |
no test coverage detected
searching dependent graphs…