Skip to content

Commit 257f211

Browse files
danjadanja
committed
multi-pipe
1 parent bc485d5 commit 257f211

3 files changed

Lines changed: 67 additions & 48 deletions

File tree

‎src/engine/TransmissionBuilder.js‎

Lines changed: 25 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -93,54 +93,41 @@ class TransmissionBuilder {
9393
// logger.log(`pipenodes = ${JSON.stringify(pipenodes)}`)
9494
for (const node of pipenodes) {
9595

96-
const np = rdf.grapoi({ dataset: transmissionsDataset, term: node })
96+
const np = rdf.grapoi({ dataset: transmissionsDataset, term: node })
9797
const processorType = np.out(ns.rdf.type).term
98-
logger.log(`processorType = ${JSON.stringify(processorType)}`)
99-
100-
logger.log('is a Trans')
101-
const isTransmissionReference = processorType.equals(ns.trn.Transmission) // doesn't cover multi, see isTransmissionReference() below
102-
103-
104-
if (!isTransmissionReference && !transmission.get(node.value)) {
105-
// const np = rdf.grapoi({ dataset: transmissionsDataset, term: node })
106-
// const processorType = np.out(ns.rdf.type).term
98+
const isTransmissionReference = processorType.equals(ns.trn.Transmission) // doesn't cover multi, see isTransmissionReference() below
10799

100+
if (isTransmissionReference) {
101+
const nestedTransmission = await this.constructTransmission(
102+
transmissionsDataset,
103+
node,
104+
configDataset
105+
)
106+
transmission.register(node.value, nestedTransmission)
107+
} else if (!transmission.get(node.value)) {
108+
// Regular processor handling
108109
const settingsNode = np.out(ns.trn.settings).term
109-
// const settingsNodeName = settingsNode ? settingsNode.value : undefined
110110

111111
logger.debug(`Creating processor:
112112
Node: :${ns.shortName(node?.value)}
113113
Type: :${ns.shortName(processorType?.value)}
114114
SettingsNode: :${ns.shortName(settingsNode?.value)}
115115
`)
116-
// Config: \n${processorsConfig}
117-
// Check if node is a nested transmission transmissionsDataset
118-
// if (processorType && this.isTransmissionReference(processorType)) {
119-
// if (processorType && this.isTransmissionReference(transmissionsDataset, node)) { //// processorType
120-
if (processorType && isTransmissionReference) {
121-
const nestedTransmission = await this.constructTransmission(
122-
transmissionsDataset,
123-
processorType, // is used?
124-
configDataset
125-
)
126-
transmission.register(node.value, nestedTransmission)
127-
} else {
128-
// Regular processor handling
129-
const processorBase = await this.createProcessor(processorType, configDataset)
130-
processorBase.id = node.value
131-
processorBase.type = processorType
132-
if (settingsNode) {
133-
processorBase.settingsNode = settingsNode
134-
}
135-
136-
// Set the app reference on the processor
137-
processorBase.app = transmission.app
138-
139-
// Connect to the transmission's whiteboard
140-
processorBase.whiteboard = transmission.whiteboard // feels redundant...
141-
processorBase.x = `X`
142-
const processorInstance = transmission.register(node.value, processorBase)
116+
117+
const processorBase = await this.createProcessor(processorType, configDataset)
118+
processorBase.id = node.value
119+
processorBase.type = processorType
120+
if (settingsNode) {
121+
processorBase.settingsNode = settingsNode
143122
}
123+
124+
// Set the app reference on the processor
125+
processorBase.app = transmission.app
126+
127+
// Connect to the transmission's whiteboard
128+
processorBase.whiteboard = transmission.whiteboard // feels redundant...
129+
processorBase.x = `X`
130+
const processorInstance = transmission.register(node.value, processorBase)
144131
}
145132
}
146133
}

‎src/model/Connector.js‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,14 @@ class Connector extends EventEmitter {
1515
logger.trace(`Connector.connect this.fromName = ${this.fromName} this.toName = ${this.toName}`)
1616
const fromProcessor = processors[this.fromName]
1717
const toProcessor = processors[this.toName]
18-
if (fromProcessor instanceof Transmission) {
18+
if (fromProcessor instanceof Transmission && toProcessor instanceof Transmission) {
19+
// Connect last node of from-transmission to first node of to-transmission
20+
const lastNode = fromProcessor.getLastNode()
21+
const firstNode = toProcessor.getFirstNode()
22+
lastNode.on('message', async (message) => {
23+
await firstNode.receive(message)
24+
})
25+
} else if (fromProcessor instanceof Transmission) {
1926
// Connect last node of nested transmission
2027
const lastNode = fromProcessor.getLastNode()
2128
lastNode.on('message', async (message) => {

‎src/model/Transmission.js‎

Lines changed: 34 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -44,17 +44,29 @@ class Transmission {
4444
lastProcessor = currentProcessor
4545
}
4646
}
47-
47+
48+
// If lastProcessor is a Transmission, get its last node
49+
let actualLastProcessor = lastProcessor
50+
if (lastProcessor instanceof Transmission) {
51+
actualLastProcessor = lastProcessor.getLastNode()
52+
}
53+
4854
// Listen for the final result from the last processor
4955
const handleFinalMessage = (finalMessage) => {
50-
lastProcessor.off('message', handleFinalMessage)
56+
actualLastProcessor.off('message', handleFinalMessage)
5157
resolve(finalMessage)
5258
}
53-
54-
lastProcessor.on('message', handleFinalMessage)
55-
59+
60+
actualLastProcessor.on('message', handleFinalMessage)
61+
62+
// If the first processor is a Transmission, send message to its first node
63+
let actualFirstProcessor = processor
64+
if (processor instanceof Transmission) {
65+
actualFirstProcessor = processor.getFirstNode()
66+
}
67+
5668
// Start the chain by sending message to first processor
57-
processor.receive(message).catch(reject)
69+
actualFirstProcessor.receive(message).catch(reject)
5870
})
5971
} else {
6072
throw new Error(`No valid processor found to execute, looked for ${processorName}`)
@@ -89,9 +101,22 @@ class Transmission {
89101
}
90102

91103
getFirstNode() { // used for nested transmissions
92-
// logger.log(this)
93-
//logger.log(this.processors[0])
94-
return this.processors[0]
104+
const processorName = this.connectors[0]?.fromName || Object.keys(this.processors)[0]
105+
return this.get(processorName)
106+
}
107+
108+
getLastNode() { // used for nested transmissions
109+
// Find the last processor in the chain by looking for one that's not a source in any connector
110+
const processorNames = Object.keys(this.processors)
111+
for (const name of processorNames) {
112+
const isSource = this.connectors.some(c => c.fromName === name)
113+
if (!isSource) {
114+
return this.get(name)
115+
}
116+
}
117+
// Fallback to the last processor in the object
118+
const lastProcessorName = processorNames[processorNames.length - 1]
119+
return this.get(lastProcessorName)
95120
}
96121

97122
// is used?

0 commit comments

Comments
 (0)