Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
refactor(node): convert handlePeerMessage to an actor
- Loading branch information
Showing
7 changed files
with
233 additions
and
192 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
97 changes: 49 additions & 48 deletions
97
packages/app-node/src/backend/peer-protocol/handlers/interledger-packet.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,70 +1,71 @@ | ||
| import { createLogger } from "@dassie/lib-logger" | ||
| import type { Reactor } from "@dassie/lib-reactive" | ||
| import { createActor } from "@dassie/lib-reactive" | ||
|
|
||
| import { EMPTY_UINT8ARRAY } from "../../../common/constants/general" | ||
| import { subnetBalanceMapStore } from "../../balances/stores/subnet-balance-map" | ||
| import { configSignal } from "../../config" | ||
| import { incomingIlpPacketTopic } from "../../ilp-connector/topics/incoming-ilp-packet" | ||
| import subnetModules from "../../subnets/modules" | ||
| import type { | ||
| IncomingPeerMessageEvent, | ||
| PeerMessageContent, | ||
| } from "../actions/handle-peer-message" | ||
| import type { IncomingPeerMessageEvent } from "../actions/handle-peer-message" | ||
|
|
||
| const logger = createLogger("das:node:handle-interledger-packet") | ||
|
|
||
| export const handleInterledgerPacket = (reactor: Reactor) => { | ||
| const incomingIlpPacketTopicValue = reactor.use(incomingIlpPacketTopic) | ||
| const { ilpAllocationScheme } = reactor.use(configSignal).read() | ||
| const balanceMap = reactor.use(subnetBalanceMapStore) | ||
| export const handleInterledgerPacket = () => | ||
| createActor((sig) => { | ||
| const incomingIlpPacketTopicValue = sig.use(incomingIlpPacketTopic) | ||
| const { ilpAllocationScheme } = sig.getKeys(configSignal, [ | ||
| "ilpAllocationScheme", | ||
| ]) | ||
| const balanceMap = sig.use(subnetBalanceMapStore) | ||
|
|
||
| const handleInterledgerPacketAsync = async ( | ||
| content: PeerMessageContent<"interledgerPacket">, | ||
| { message: { sender, subnetId }, authenticated }: IncomingPeerMessageEvent | ||
| ) => { | ||
| if (!authenticated) { | ||
| logger.warn("received unauthenticated interledger packet, discarding") | ||
| return | ||
| } | ||
| const handleInterledgerPacketAsync = async ({ | ||
| message: { | ||
| sender, | ||
| subnetId, | ||
| content: { | ||
| value: { value: content }, | ||
| }, | ||
| }, | ||
| authenticated, | ||
| }: IncomingPeerMessageEvent<"interledgerPacket">) => { | ||
| if (!authenticated) { | ||
| logger.warn("received unauthenticated interledger packet, discarding") | ||
| return | ||
| } | ||
|
|
||
| logger.debug("handle interledger packet", { | ||
| subnet: subnetId, | ||
| from: sender, | ||
| }) | ||
| logger.debug("handle interledger packet", { | ||
| subnet: subnetId, | ||
| from: sender, | ||
| }) | ||
|
|
||
| const incomingPacketEvent = incomingIlpPacketTopicValue.prepareEvent({ | ||
| source: `${ilpAllocationScheme}.das.${subnetId}.${sender}`, | ||
| packet: content.signed.packet, | ||
| requestId: content.signed.requestId, | ||
| }) | ||
| const incomingPacketEvent = incomingIlpPacketTopicValue.prepareEvent({ | ||
| source: `${ilpAllocationScheme}.das.${subnetId}.${sender}`, | ||
| packet: content.signed.packet, | ||
| requestId: content.signed.requestId, | ||
| }) | ||
|
|
||
| const subnetModule = subnetModules[subnetId] | ||
| const subnetModule = subnetModules[subnetId] | ||
|
|
||
| if (!subnetModule) { | ||
| throw new Error(`unknown subnet: ${subnetId}`) | ||
| } | ||
| if (!subnetModule) { | ||
| throw new Error(`unknown subnet: ${subnetId}`) | ||
| } | ||
|
|
||
| await subnetModule.processIncomingPacket({ | ||
| subnetId, | ||
| balanceMap, | ||
| packet: incomingPacketEvent.packet, | ||
| }) | ||
| await subnetModule.processIncomingPacket({ | ||
| subnetId, | ||
| balanceMap, | ||
| packet: incomingPacketEvent.packet, | ||
| }) | ||
|
|
||
| incomingIlpPacketTopicValue.emit(incomingPacketEvent) | ||
| } | ||
| incomingIlpPacketTopicValue.emit(incomingPacketEvent) | ||
| } | ||
|
|
||
| return ( | ||
| content: PeerMessageContent<"interledgerPacket">, | ||
| parameters: IncomingPeerMessageEvent | ||
| ) => { | ||
| handleInterledgerPacketAsync(content, parameters).catch( | ||
| (error: unknown) => { | ||
| return (parameters: IncomingPeerMessageEvent<"interledgerPacket">) => { | ||
| handleInterledgerPacketAsync(parameters).catch((error: unknown) => { | ||
| logger.error("error while handling interledger packet", { | ||
| error, | ||
| }) | ||
| } | ||
| ) | ||
| }) | ||
|
|
||
| return EMPTY_UINT8ARRAY | ||
| } | ||
| } | ||
| return EMPTY_UINT8ARRAY | ||
| } | ||
| }) |
31 changes: 20 additions & 11 deletions
31
packages/app-node/src/backend/peer-protocol/handlers/link-state-request.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,19 +1,28 @@ | ||
| import type { Reactor } from "@dassie/lib-reactive" | ||
| import { createActor } from "@dassie/lib-reactive" | ||
|
|
||
| import { EMPTY_UINT8ARRAY } from "../../../common/constants/general" | ||
| import { configSignal } from "../../config" | ||
| import type { PeerMessageContent } from "../actions/handle-peer-message" | ||
| import type { IncomingPeerMessageEvent } from "../actions/handle-peer-message" | ||
| import { nodeTableStore } from "../stores/node-table" | ||
|
|
||
| export const handleLinkStateRequest = (reactor: Reactor) => { | ||
| const nodeTable = reactor.use(nodeTableStore) | ||
| const { nodeId } = reactor.use(configSignal).read() | ||
| export const handleLinkStateRequest = () => | ||
| createActor((sig) => { | ||
| const nodeTable = sig.use(nodeTableStore) | ||
| const { nodeId } = sig.getKeys(configSignal, ["nodeId"]) | ||
|
|
||
| return ({ subnetId }: PeerMessageContent<"linkStateRequest">) => { | ||
| const ownNodeTableEntry = nodeTable.read().get(`${subnetId}.${nodeId}`) | ||
| return ({ | ||
| message: { | ||
| content: { | ||
| value: { | ||
| value: { subnetId }, | ||
| }, | ||
| }, | ||
| }, | ||
| }: IncomingPeerMessageEvent<"linkStateRequest">) => { | ||
| const ownNodeTableEntry = nodeTable.read().get(`${subnetId}.${nodeId}`) | ||
|
|
||
| if (!ownNodeTableEntry?.lastLinkStateUpdate) return EMPTY_UINT8ARRAY | ||
| if (!ownNodeTableEntry?.lastLinkStateUpdate) return EMPTY_UINT8ARRAY | ||
|
|
||
| return ownNodeTableEntry.lastLinkStateUpdate | ||
| } | ||
| } | ||
| return ownNodeTableEntry.lastLinkStateUpdate | ||
| } | ||
| }) |
Oops, something went wrong.