Skip to content

Commit aa54875

Browse files
tysonrmtysonrm
authored
Merge pull request #272 from module-federation/release-0-1-0
Release 0 1 0
2 parents bc6049d + 5e07443 commit aa54875

File tree

7 files changed

+125
-124
lines changed

7 files changed

+125
-124
lines changed

‎.gitignore

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,3 +8,5 @@ tst-rtry.js
88
.VSCodeCounter/
99
esm
1010
lib
11+
nohup.out
12+
nohup.out

‎src/adapters/datasources/datasource-mongodb.js

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,14 @@ const dsOptions = configRoot.adapters.datasources.DataSourceMongoDb.options || {
2828
const cacheSize = configRoot.adapters.cacheSize || 3000
2929
let connPool
3030

31+
class DsMongoError extends Error {
32+
constructor (error, code) {
33+
super(error)
34+
this.code = code
35+
console.error(this)
36+
}
37+
}
38+
3139
/**
3240
* @type {Map<string,MongoClient>}
3341
*/
@@ -84,7 +92,9 @@ export class DataSourceMongoDb extends DataSource {
8492
}, 500)
8593
await client.connect()
8694
clearTimeout(timerId)
87-
if (timeout) throw new Error('mongo conn timeout')
95+
if (timeout) {
96+
throw new DsMongoError('mongo conn timeout', 500)
97+
}
8898
}
8999
}
90100

@@ -381,7 +391,7 @@ export class DataSourceMongoDb extends DataSource {
381391
}
382392
console.log({ options })
383393

384-
if (options.streamRequested) return this.streamList(options)
394+
if (options.streamResult) return this.streamList(options)
385395

386396
const data = (await this.mongoFind(options)).toArray()
387397
const count = data?.length

‎src/domain/datasource-factory.js

Lines changed: 21 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ import configRoot from '../config'
2020
import DataSource from './datasource'
2121
import compose from './util/compose'
2222
import { withSharedMemory } from './shared-memory'
23+
import { assert } from 'node:console'
2324

2425
const debug = /\*|datasource/i.test(process.env.DEBUG)
2526
const broker = EventBrokerFactory.getInstance()
@@ -28,13 +29,17 @@ const defaultAdapter = configRoot.hostConfig.adapters.defaultDatasource
2829
const DefaultDataSource =
2930
adapters[defaultAdapter] || dsClasses['DataSourceMemory']
3031

32+
class DsError extends Error {
33+
constructor (error, code) {
34+
super(error)
35+
this.code = code
36+
}
37+
}
38+
3139
/**
3240
* Core extensions include object caching, marshalling and serialization.
3341
* Using this compositional mixin, these extensions are applied transparently
3442
* to any {@link DataSource} class in the hierarchy.
35-
*
36-
* @param {*} superclass
37-
* @returns
3843
*/
3944
const DsCoreExtensions = superclass =>
4045
class extends superclass {
@@ -50,11 +55,6 @@ const DsCoreExtensions = superclass =>
5055
return this[FACTORY]
5156
}
5257

53-
// serialize (data, options) {
54-
// if (options?.serializers) return options.serializers['serialize'](data)
55-
// return JSON.stringify(data)
56-
// }
57-
5858
/**
5959
* Override the super class, adding cache and serialization functions.
6060
* @override
@@ -67,7 +67,7 @@ const DsCoreExtensions = superclass =>
6767
await super.save(id, JSON.parse(JSON.stringify(data)))
6868
} catch (error) {
6969
console.error({ fn: this.save.name, error })
70-
throw error
70+
throw new DsError(error, 500)
7171
}
7272
}
7373

@@ -79,22 +79,17 @@ const DsCoreExtensions = superclass =>
7979
* @returns {Promise<Model>|undefined}
8080
*/
8181
async find (id) {
82-
try {
83-
const cached = this.findSync(id)
84-
if (cached) return cached
82+
const cached = this.findSync(id)
83+
if (cached) return cached
8584

86-
const model = await super.find(id)
85+
const model = await super.find(id)
86+
if (model) {
87+
// save to cache
88+
this.saveSync(id, model)
8789

88-
if (model) {
89-
// save to cache
90-
this.saveSync(id, model)
91-
return isMainThread
92-
? model
93-
: ModelFactory.loadModel(broker, this, model, this.name)
94-
}
95-
} catch (error) {
96-
console.error({ fn: this.find.name, error })
97-
throw error
90+
return isMainThread
91+
? model // dont unmarshall - main gets readonly copy
92+
: ModelFactory.loadModel(broker, this, model, this.name)
9893
}
9994
}
10095

@@ -149,6 +144,7 @@ const DsCoreExtensions = superclass =>
149144
*/
150145
stream (list, options) {
151146
return new Promise((resolve, reject) => {
147+
assert.ok(list && options, 'missing m kkkakkk')
152148
options.writable.on('error', reject)
153149
options.writable.on('end', resolve)
154150

@@ -179,13 +175,11 @@ const DsCoreExtensions = superclass =>
179175

180176
const opts = {
181177
...options,
182-
streamRequested:
178+
streamResult:
183179
options?.writable && !options?.query?.__aggregate ? true : false
184180
}
185181
const list = [await super.list(opts)].flat()
186-
187-
if (list.length < 1) return []
188-
182+
if (list.length < 1) throw new Error()
189183
if (list[0] instanceof Readable || list[0] instanceof Transform)
190184
return this.stream(list, options)
191185

‎src/domain/event-router.js

Lines changed: 79 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -1,50 +1,95 @@
11
'use strict'
22

3-
import { workerData, BroadcastChannel, isMainThread } from 'worker_threads'
4-
import { modelsInDomain } from './use-cases'
3+
import { workerData, BroadcastChannel } from 'worker_threads'
54

65
export class PortEventRouter {
76
constructor (models, broker) {
87
this.models = models
98
this.broker = broker
109
}
1110

12-
getThreadLocalPorts () {
13-
const localSpec = this.models.getModelSpec(
14-
workerData.poolName.toUpperCase()
15-
)
16-
return this.models
11+
get localSpec () {
12+
if (this.__localSpec) return this.__localSpec
13+
this.__localSpec = this.models.getModelSpec(workerData.poolName)
14+
return this.__localSpec
15+
}
16+
17+
get threadLocalPorts () {
18+
if (this.__threadLocalPorts) return this.__threadLocalPorts
19+
this.__threadLocalPorts = this.models
1720
.getModelSpecs()
1821
.filter(
1922
spec =>
2023
spec.ports &&
21-
(spec.domain.toUpperCase() === localSpec.domain.toUpperCase() ||
22-
spec.modelName.toUpperCase() === localSpec.modelName.toUpperCase())
24+
(spec.domain === this.localSpec.domain ||
25+
spec.modelName === this.localSpec.modelName)
2326
)
2427
.flatMap(spec =>
2528
Object.values(spec.ports)
2629
.filter(port => port.consumesEvent || port.producesEvent)
27-
.map(port => ({ ...port, modelName: spec.modelName }))
30+
.map(port => ({
31+
...port,
32+
modelName: spec.modelName,
33+
domain: spec.domain
34+
}))
2835
)
36+
return this.__threadLocalPorts
2937
}
3038

31-
getThreadRemotePorts () {
32-
return this.models
39+
get threadRemotePorts () {
40+
if (this.__threadRemotePorts) return this.__threadRemotePorts
41+
this.__threadRemotePorts = this.models
3342
.getModelSpecs()
3443
.filter(
3544
spec =>
3645
spec.ports &&
37-
!this.getThreadLocalPorts().find(l => l.modelName === spec.modelName)
46+
!this.threadLocalPorts.find(l => l.modelName === spec.modelName)
3847
)
3948
.flatMap(spec =>
4049
Object.values(spec.ports)
4150
.filter(port => port.consumesEvent || port.producesEvent)
42-
.map(port => ({ ...port, modelName: spec.modelName }))
51+
.map(port => ({
52+
...port,
53+
modelName: spec.modelName,
54+
domain: spec.domain
55+
}))
4356
)
57+
return this.__threadRemotePorts
58+
}
59+
60+
get publisherPorts () {
61+
if (this.__publisherPorts) this.__publisherPorts
62+
this.__publisherPorts = this.threadRemotePorts.filter(remote =>
63+
this.threadLocalPorts.find(
64+
local => local.producesEvent === remote.consumesEvent
65+
)
66+
)
67+
return this.__publisherPorts
4468
}
4569

46-
handleChannelEvent (msg) {
47-
if (msg.data.eventName) this.broker.notify(msg.data.eventName, msg.data)
70+
get subscriberPorts () {
71+
if (this.__subscriberPorts) return this.__subscriberPorts
72+
this.__subscriberPorts = this.threadRemotePorts.filter(remote =>
73+
this.threadLocalPorts.find(
74+
local => local.consumesEvent === remote.producesEvent
75+
)
76+
)
77+
return this.__subscriberPorts
78+
}
79+
80+
get unhandledPorts () {
81+
if (this.__unhandledPorts) return this.__unhandledPorts
82+
this.__unhandledPorts = this.threadLocalPorts.filter(
83+
local =>
84+
!this.threadRemotePorts.find(
85+
remote => local.producesEvent === remote.consumesEvent
86+
) && !this.localPorts.find(l => local.producesEvent === l.consumesEvent)
87+
)
88+
return this.__unhandledPorts
89+
}
90+
91+
handleBroadcastEvent (msg) {
92+
if (msg?.data?.eventName) this.broker.notify(msg.data.eventName, msg.data)
4893
else {
4994
console.log('missing eventName', msg.data)
5095
this.broker.notify('missingEventName', msg.data)
@@ -55,78 +100,56 @@ export class PortEventRouter {
55100
* Listen for producer events from other thread pools and invoke
56101
* local ports that consume them. Listen for local producer events
57102
* and forward to pools that consume them. If a producer event is
58-
* not consumed by any local thread, foward to service mesh.
103+
* not consumed by any local thread, foward to the service mesh.
59104
*/
60105
listen () {
61-
const localPorts = this.getThreadLocalPorts()
62-
const remotePorts = this.getThreadRemotePorts()
63-
64-
console.debug({ localPorts })
65-
console.debug({ remotePorts })
66-
67-
const publishPorts = remotePorts.filter(remote =>
68-
localPorts.find(local => local.producesEvent === remote.consumesEvent)
69-
)
70-
const subscribePorts = remotePorts.filter(remote =>
71-
localPorts.find(local => local.consumesEvent === remote.producesEvent)
72-
)
73-
const unhandledPorts = localPorts.filter(
74-
local =>
75-
!remotePorts.find(
76-
remote => local.producesEvent === remote.consumesEvent
77-
) && !localPorts.find(l => local.producesEvent === l.consumesEvent)
78-
)
79-
80106
const services = new Set()
81107
const channels = new Map()
82108

83-
publishPorts.forEach(port => services.add(port.modelName))
84-
subscribePorts.forEach(port => services.add(port.modelName))
109+
this.publisherPorts.forEach(port => services.add(port.modelName))
110+
this.subscriberPorts.forEach(port => services.add(port.modelName))
85111

86112
services.forEach(service =>
87113
channels.set(service, new BroadcastChannel(service))
88114
)
89115

90-
console.log('publishPorts', publishPorts)
91-
console.log('subscribePorts', subscribePorts)
92-
console.log('unhandledPorts', unhandledPorts)
116+
console.log('publisherPorts', this.publisherPorts)
117+
console.log('subscriberPorts', this.subscriberPorts)
118+
console.log('unhandledPorts', this.unhandledPorts)
93119
console.log('channels', channels)
94120

95-
// dispatch outgoing events
96-
publishPorts.forEach(port =>
121+
// dispatch outgoing events to local pools
122+
this.publisherPorts.forEach(port =>
97123
this.broker.on(port.consumesEvent, event => {
98124
console.log('broadcasting...', { port, event })
99125
channels
100126
.get(port.modelName)
101-
.postMessage(
102-
JSON.parse(
103-
JSON.stringify({ ...event, route: 'balanceEventConsumer' })
104-
)
105-
)
127+
.postMessage(JSON.parse(JSON.stringify(event)))
106128
})
107129
)
108130

109-
// listen for incoming events
110-
subscribePorts.forEach(port => {
131+
// listen for incoming events from local pools
132+
this.subscriberPorts.forEach(port => {
111133
channels.get(port.modelName).onmessage = msg => {
112134
console.log('subscribePorts.onmessage', msg.data)
113-
this.handleChannelEvent(msg)
135+
this.handleBroadcastEvent(msg)
114136
}
115137
})
116-
117-
unhandledPorts.forEach(port => {
138+
139+
// send ports not handled by local pool to mesh
140+
this.unhandledPorts.forEach(port => {
118141
this.broker.on(port.producesEvent, event => {
119142
this.broker.notify('to_main', {
120143
...event,
121-
route: 'balanceEventConsumer'
144+
route: 'balanceEventConsumer' // mesh routing algo
122145
})
123146
})
124147
})
125148

126149
// listen to this model's channel
127-
new BroadcastChannel(workerData.poolName.toUpperCase()).onmessage = msg => {
150+
new BroadcastChannel(workerData.poolName).onmessage = msg => {
128151
console.log('onmessage', msg.data)
129-
this.handleChannelEvent(msg)
152+
this.handleBroadcastEvent(msg)
130153
}
131154
}
132155
}

‎src/domain/thread-pool.js

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -502,13 +502,13 @@ export class ThreadPool extends EventEmitter {
502502
return {
503503
name: this.name,
504504
open: !this.closed,
505-
max: this.maxPoolSize(),
506-
min: this.minPoolSize(),
507-
total: this.poolSize(),
508-
waiting: this.jobQueueDepth(),
509-
available: this.availThreadCount(),
505+
maxThreads: this.maxPoolSize(),
506+
minThreads: this.minPoolSize(),
507+
totalThreads: this.poolSize(),
508+
waitingJobs: this.jobQueueDepth(),
509+
availableThreads: this.availThreadCount(),
510510
jobsRequested: this.totalJobsRequested(),
511-
avgDuration: this.avgJobDuration(),
511+
avgJobDuration: this.avgJobDuration(),
512512
avgDurTolerance: this.jobDurationThreshold(),
513513
queueRate: this.jobQueueRate(),
514514
queueTolerance: this.jobQueueThreshold(),

0 commit comments

Comments
 (0)