Skip to content

Commit 82a7ee1

Browse files
riclolsenriclolsen
authored
Merge pull request #198 from json-scada/master
Improvements to cs_data_processor and other nodejs processes.
2 parents 3798c8e + c6377f5 commit 82a7ee1

File tree

11 files changed

+141
-94
lines changed

11 files changed

+141
-94
lines changed

‎src/OPC-UA-Server/index.js

+3-1
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,8 @@ const AppDefs = require('./app-defs')
3232
const { LoadConfig, getMongoConnectionOptions } = require('./load-config')
3333
let HintMongoIsConnected = true
3434

35+
process.on('uncaughtException', err => Log.log('Uncaught Exception:' + JSON.stringify(err)))
36+
3537
;(async () => {
3638
const jsConfig = LoadConfig() // load and parse config file
3739
Log.levelCurrent = jsConfig.LogLevel
@@ -779,7 +781,7 @@ async function checkConnectedMongo(client) {
779781
if (!client) {
780782
return false
781783
}
782-
const CheckMongoConnectionTimeout = 1000
784+
const CheckMongoConnectionTimeout = 10000
783785
let tr = setTimeout(() => {
784786
Log.log('Mongo ping timeout error!')
785787
HintMongoIsConnected = false

‎src/alarm_beep/alarm_beep.js

+3-1
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,8 @@ const { MongoClient } = require('mongodb')
2727
const { setInterval } = require('timers')
2828
const sys = require('child_process')
2929

30+
process.on('uncaughtException', err => Log.log('Uncaught Exception:' + JSON.stringify(err)))
31+
3032
const WavFilesWin = [
3133
'c:\\windows\\media\\Windows Default.wav',
3234
'c:\\windows\\media\\Windows Background.wav',
@@ -167,7 +169,7 @@ async function checkConnectedMongo(client) {
167169
if (!client) {
168170
return false
169171
}
170-
const CheckMongoConnectionTimeout = 1000
172+
const CheckMongoConnectionTimeout = 10000
171173
let tr = setTimeout(() => {
172174
Log.log('Mongo ping timeout error!')
173175
HintMongoIsConnected = false

‎src/cs_custom_processor/cs_custom_processor.js

+3-1
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@ const Redundancy = require('./redundancy')
2525
const { MongoClient } = require('mongodb')
2626
const CustomProcessor = require('./customized_module').CustomProcessor
2727

28+
process.on('uncaughtException', err => Log.log('Uncaught Exception:' + JSON.stringify(err)))
29+
2830
const args = process.argv.slice(2)
2931
let inst = null
3032
if (args.length > 0) inst = parseInt(args[0])
@@ -81,7 +83,7 @@ async function checkConnectedMongo(client) {
8183
if (!client) {
8284
return false
8385
}
84-
const CheckMongoConnectionTimeout = 1000
86+
const CheckMongoConnectionTimeout = 10000
8587
let tr = setTimeout(() => {
8688
Log.log('Mongo ping timeout error!')
8789
MongoStatus.HintMongoIsConnected = false

‎src/cs_data_processor/cs_data_processor.js

+111-84
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,10 @@ const { setInterval } = require('timers')
3232
const MongoStatus = { HintMongoIsConnected: false }
3333
const LowestPriorityThatBeeps = 1 // will beep for priorities zero and one
3434

35+
process.on('uncaughtException', (err) =>
36+
Log.log('Uncaught Exception:' + JSON.stringify(err))
37+
)
38+
3539
const args = process.argv.slice(2)
3640
var inst = null
3741
if (args.length > 0) inst = parseInt(args[0])
@@ -96,10 +100,11 @@ const pipeline = [
96100
let sqlRtDataQueue = new Queue() // queue of realtime values to insert on postgreSQL
97101
let mongoRtDataQueue = new Queue() // queue of realtime values to insert on MongoDB
98102
let digitalUpdatesCount = 0
103+
let clientMongo = null
99104

100105
// mark as frozen unchanged analog values greater than 1 after timeout
101106
setInterval(async function () {
102-
if (collection && MongoStatus.HintMongoIsConnected) {
107+
if (collection && MongoStatus.HintMongoIsConnected && clientMongo) {
103108
collection
104109
.updateMany(
105110
{
@@ -133,13 +138,17 @@ const pipeline = [
133138
}
134139
}, 17317)
135140

136-
setInterval(async function () {
141+
// process updates to mongo/realtimeData
142+
async function processRtDataMongoUpdates() {
137143
if (
138144
!collection ||
145+
!clientMongo ||
139146
!MongoStatus.HintMongoIsConnected ||
140147
mongoRtDataQueue.isEmpty()
141-
)
148+
) {
149+
setTimeout(processRtDataMongoUpdates, 150)
142150
return
151+
}
143152
let cnt = 0
144153
let updArr = []
145154
while (!mongoRtDataQueue.isEmpty()) {
@@ -179,75 +188,81 @@ const pipeline = [
179188
.catch(function (err) {
180189
Log.log('Error on Mongodb query!', err)
181190
})
182-
// Log.log(JSON.stringify(res))
183191
if (cnt) Log.log('Mongo Updates ' + cnt)
184-
}, 150)
192+
setTimeout(processRtDataMongoUpdates, 150)
193+
}
194+
processRtDataMongoUpdates()
185195

186-
// write values to sql files for later insertion on postgreSQL
187-
setInterval(async function () {
188-
if (!histCollection || !MongoStatus.HintMongoIsConnected) return
189-
let doInsertData = false
190-
let sqlTransaction =
191-
'START TRANSACTION;\n' +
192-
'INSERT INTO hist (tag, time_tag, value, value_json, time_tag_at_source, flags) VALUES '
193-
194-
let cntH = 0
195-
let insertArr = []
196-
while (!sqlHistQueue.isEmpty()) {
197-
doInsertData = true
198-
let entry = sqlHistQueue.peek()
199-
sqlHistQueue.dequeue()
200-
sqlTransaction = sqlTransaction + '\n(' + entry.sql + '),'
201-
insertArr.push(entry.obj)
202-
cntH++
196+
// write values to sql files for later insertion on postgreSQL, and mongo hist
197+
async function processSqlAndMongoHistUpdates() {
198+
if (!histCollection || !clientMongo || !MongoStatus.HintMongoIsConnected) {
199+
setTimeout(processSqlAndMongoHistUpdates, 333)
200+
return
203201
}
204-
if (cntH) Log.log('PGSQL/Mongo Hist updates ' + cntH)
205202

206-
if (doInsertData) {
207-
histCollection
208-
.insertMany(insertArr, { ordered: false, writeConcern: { w: 0 } })
209-
.catch(function (err) {
210-
Log.log('Error on Mongodb query!', err)
211-
})
203+
try {
204+
let doInsertData = false
205+
let sqlTransaction =
206+
'START TRANSACTION;\n' +
207+
'INSERT INTO hist (tag, time_tag, value, value_json, time_tag_at_source, flags) VALUES '
208+
209+
let cntH = 0
210+
let insertArr = []
211+
while (!sqlHistQueue.isEmpty()) {
212+
doInsertData = true
213+
let entry = sqlHistQueue.peek()
214+
sqlHistQueue.dequeue()
215+
sqlTransaction = sqlTransaction + '\n(' + entry.sql + '),'
216+
insertArr.push(entry.obj)
217+
cntH++
218+
}
219+
if (cntH) Log.log('PGSQL/Mongo Hist updates ' + cntH)
220+
221+
if (doInsertData) {
222+
histCollection
223+
.insertMany(insertArr, { ordered: false, writeConcern: { w: 0 } })
224+
.catch(function (err) {
225+
Log.log('Error on Mongodb query!', err)
226+
})
227+
sqlTransaction = sqlTransaction.substring(0, sqlTransaction.length - 1) // remove last comma
228+
sqlTransaction = sqlTransaction + ' \n'
229+
// this cause problems when tag/time repeated on same transaction
230+
// sqlTransaction = sqlTransaction + "ON CONFLICT (tag, time_tag) DO UPDATE SET value=EXCLUDED.value, value_json=EXCLUDED.value_json, time_tag_at_source=EXCLUDED.time_tag_at_source, flags=EXCLUDED.flags;\n";
231+
sqlTransaction =
232+
sqlTransaction + 'ON CONFLICT (tag, time_tag) DO NOTHING;\n'
233+
sqlTransaction = sqlTransaction + 'COMMIT;\n'
234+
fs.writeFile(
235+
sqlFilesPath +
236+
'pg_hist_' +
237+
new Date().getTime() +
238+
'_' +
239+
jsConfig.Instance +
240+
'.sql',
241+
sqlTransaction,
242+
(err) => {
243+
if (err) Log.log('Error writing SQL file!')
244+
}
245+
)
246+
}
247+
248+
doInsertData = false
249+
sqlTransaction = ''
250+
let cntR = 0
251+
sqlTransaction =
252+
sqlTransaction +
253+
'WITH ordered_values AS ( SELECT DISTINCT ON (tag) tag, time_tag, json_data FROM (VALUES '
254+
while (!sqlRtDataQueue.isEmpty()) {
255+
doInsertData = true
256+
let sql = sqlRtDataQueue.peek()
257+
sqlRtDataQueue.dequeue()
258+
sqlTransaction = sqlTransaction + '\n (' + sql + '),'
259+
cntR++
260+
}
212261
sqlTransaction = sqlTransaction.substring(0, sqlTransaction.length - 1) // remove last comma
213262
sqlTransaction = sqlTransaction + ' \n'
214-
// this cause problems when tag/time repeated on same transaction
215-
// sqlTransaction = sqlTransaction + "ON CONFLICT (tag, time_tag) DO UPDATE SET value=EXCLUDED.value, value_json=EXCLUDED.value_json, time_tag_at_source=EXCLUDED.time_tag_at_source, flags=EXCLUDED.flags;\n";
216263
sqlTransaction =
217-
sqlTransaction + 'ON CONFLICT (tag, time_tag) DO NOTHING;\n'
218-
sqlTransaction = sqlTransaction + 'COMMIT;\n'
219-
fs.writeFile(
220-
sqlFilesPath +
221-
'pg_hist_' +
222-
new Date().getTime() +
223-
'_' +
224-
jsConfig.Instance +
225-
'.sql',
226-
sqlTransaction,
227-
(err) => {
228-
if (err) Log.log('Error writing SQL file!')
229-
}
230-
)
231-
}
232-
233-
doInsertData = false
234-
sqlTransaction = ''
235-
let cntR = 0
236-
sqlTransaction =
237-
sqlTransaction +
238-
'WITH ordered_values AS ( SELECT DISTINCT ON (tag) tag, time_tag, json_data FROM (VALUES '
239-
while (!sqlRtDataQueue.isEmpty()) {
240-
doInsertData = true
241-
let sql = sqlRtDataQueue.peek()
242-
sqlRtDataQueue.dequeue()
243-
sqlTransaction = sqlTransaction + '\n (' + sql + '),'
244-
cntR++
245-
}
246-
sqlTransaction = sqlTransaction.substring(0, sqlTransaction.length - 1) // remove last comma
247-
sqlTransaction = sqlTransaction + ' \n'
248-
sqlTransaction =
249-
sqlTransaction +
250-
`) AS t(tag, time_tag, json_data)
264+
sqlTransaction +
265+
`) AS t(tag, time_tag, json_data)
251266
ORDER BY tag, time_tag DESC
252267
)
253268
INSERT INTO realtime_data (tag, time_tag, json_data)
@@ -257,27 +272,32 @@ const pipeline = [
257272
SET time_tag = EXCLUDED.time_tag,
258273
json_data = EXCLUDED.json_data;
259274
`
260-
if (cntR) Log.log('PGSQL RT updates ' + cntR)
261-
262-
if (doInsertData) {
263-
fs.writeFile(
264-
sqlFilesPath +
265-
'pg_rtdata_' +
266-
new Date().getTime() +
267-
'_' +
268-
jsConfig.Instance +
269-
'.sql',
270-
sqlTransaction,
271-
(err) => {
272-
if (err) Log.log('Error writing SQL file!')
273-
}
274-
)
275+
if (cntR) Log.log('PGSQL RT updates ' + cntR)
276+
277+
if (doInsertData) {
278+
fs.writeFile(
279+
sqlFilesPath +
280+
'pg_rtdata_' +
281+
new Date().getTime() +
282+
'_' +
283+
jsConfig.Instance +
284+
'.sql',
285+
sqlTransaction,
286+
(err) => {
287+
if (err) Log.log('Error writing SQL file!')
288+
}
289+
)
290+
}
291+
} catch (e) {
292+
Log.log('Error in processSqlAndMongoHistUpdates: ' + e)
275293
}
276-
}, 1000)
294+
setTimeout(processSqlAndMongoHistUpdates, 333)
295+
}
296+
processSqlAndMongoHistUpdates()
277297

278-
let clientMongo = null
279298
let invalidDetectIntervalHandle = null
280299
let latencyIntervalHandle = null
300+
let resumeToken = null
281301
while (true) {
282302
if (clientMongo === null)
283303
await MongoClient.connect(
@@ -286,12 +306,15 @@ const pipeline = [
286306
)
287307
.then(async (client) => {
288308
clientMongo = client
289-
clientMongo.on('topologyClosed', (_) => {
309+
clientMongo.on('topologyClosed', () => {
290310
MongoStatus.HintMongoIsConnected = false
311+
clientMongo = null
291312
Log.log('MongoDB server topologyClosed')
292313
})
293314
MongoStatus.HintMongoIsConnected = true
294315
Log.log('Connected correctly to MongoDB server')
316+
if (resumeToken)
317+
Log.log('resumeToken: ' + JSON.stringify(resumeToken))
295318

296319
let latencyAccTotal = 0
297320
let latencyTotalCnt = 0
@@ -310,6 +333,7 @@ const pipeline = [
310333
histCollection = db.collection(jsConfig.HistCollectionName)
311334
const changeStream = collection.watch(pipeline, {
312335
fullDocument: 'updateLookup',
336+
resumeAfter: resumeToken,
313337
})
314338

315339
await createSpecialTags(collection)
@@ -458,6 +482,7 @@ const pipeline = [
458482
// start listen to changes
459483
changeStream.on('change', (change) => {
460484
try {
485+
resumeToken = changeStream.resumeToken
461486
if (change.operationType === 'delete') return
462487

463488
// // for older versions of mongodb
@@ -1256,7 +1281,7 @@ async function checkConnectedMongo(client) {
12561281
if (!client) {
12571282
return false
12581283
}
1259-
const CheckMongoConnectionTimeout = 1000
1284+
const CheckMongoConnectionTimeout = 10000
12601285
let tr = setTimeout(() => {
12611286
Log.log('Mongo ping timeout error!')
12621287
MongoStatus.HintMongoIsConnected = false
@@ -1274,6 +1299,8 @@ async function checkConnectedMongo(client) {
12741299
MongoStatus.HintMongoIsConnected = true
12751300
return true
12761301
} else {
1302+
if (!!client && !!client.topology && client.topology.isConnected())
1303+
return true
12771304
MongoStatus.HintMongoIsConnected = false
12781305
return false
12791306
}

‎src/demo_simul/index.js

+3-1
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,8 @@ const AppDefs = require('./app-defs')
2424
const LoadConfig = require('./load-config')
2525
const { MongoClient, Double } = require('mongodb')
2626

27+
process.on('uncaughtException', err => Log.log('Uncaught Exception:' + JSON.stringify(err)))
28+
2729
const pipeline = [
2830
{
2931
$project: { documentKey: false },
@@ -537,7 +539,7 @@ async function checkConnectedMongo(client) {
537539
if (!client) {
538540
return false
539541
}
540-
const CheckMongoConnectionTimeout = 1000
542+
const CheckMongoConnectionTimeout = 10000
541543
const tr = setTimeout(() => {
542544
Log.log('Mongo ping timeout error!')
543545
HintMongoIsConnected = false

‎src/grafana_alert2event/grafana_alert2event.js

+3-1
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,8 @@ const app = express()
4646
app.use(express.json())
4747
let soeQueue = new Queue() // queue of SOE events
4848

49+
process.on('uncaughtException', err => console.log('Uncaught Exception:' + JSON.stringify(err)))
50+
4951
app.listen(HTTP_PORT, IP_BIND, () => {
5052
console.log('listening on ' + IP_BIND + ':' + HTTP_PORT)
5153
})
@@ -307,7 +309,7 @@ if (
307309
})()
308310

309311
// test mongoDB connectivity
310-
let CheckMongoConnectionTimeout = 1000
312+
let CheckMongoConnectionTimeout = 10000
311313
let HintMongoIsConnected = true
312314
async function checkConnectedMongo(client) {
313315
if (!client) {

‎src/mongofw/index.js

+3-1
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,8 @@ const Redundancy = require('./redundancy')
2424
const { MongoClient } = require('mongodb')
2525
const CustomProcessor = require('./customized_module').CustomProcessor
2626

27+
process.on('uncaughtException', err => Log.log('Uncaught Exception:' + JSON.stringify(err)))
28+
2729
const args = process.argv.slice(2)
2830
let inst = null
2931
if (args.length > 0) inst = parseInt(args[0])
@@ -80,7 +82,7 @@ async function checkConnectedMongo(client) {
8082
if (!client) {
8183
return false
8284
}
83-
const CheckMongoConnectionTimeout = 2000
85+
const CheckMongoConnectionTimeout = 10000
8486
let tr = setTimeout(() => {
8587
Log.log('Mongo ping timeout error!')
8688
MongoStatus.HintMongoIsConnected = false

0 commit comments

Comments
 (0)