@@ -117,21 +117,18 @@ export default class DbClient {
117117 pushdev,
118118 channelRouter,
119119 } : {
120- sub : SocketWrapper | string [ ] ;
121- subdev : SocketWrapper | string [ ] ;
120+ sub ? : SocketWrapper | string [ ] ;
121+ subdev ? : SocketWrapper | string [ ] ;
122122 push : SocketWrapper | string [ ] ;
123123 pushdev : SocketWrapper | string [ ] ;
124124 channelRouter ?: EventEmitter ;
125125 } ,
126126 _log : ReturnType < typeof logger . createLogger > | undefined = log
127- ) : Promise < {
128- sub : SocketWrapper ;
129- subdev : SocketWrapper ;
130- push : SocketWrapper ;
131- pushdev : SocketWrapper ;
132- channelRouter : EventEmitter ;
133- } > {
134- if ( Array . isArray ( sub ) ) {
127+ ) {
128+ let finalSub : SocketWrapper | undefined
129+ let finalSubdev : SocketWrapper | undefined
130+
131+ if ( sub && Array . isArray ( sub ) ) {
135132 const _sub = zmqutil . socket ( 'sub' )
136133 await Promise . all (
137134 sub . map ( async ( endpoint ) => {
@@ -153,10 +150,13 @@ export default class DbClient {
153150 }
154151 } )
155152 )
156- sub = _sub
153+ finalSub = _sub
154+ }
155+ else if ( sub ) {
156+ finalSub = sub as SocketWrapper
157157 }
158158
159- if ( Array . isArray ( subdev ) ) {
159+ if ( subdev && Array . isArray ( subdev ) ) {
160160 const _subdev = zmqutil . socket ( 'sub' )
161161 await Promise . all (
162162 subdev . map ( async ( endpoint ) => {
@@ -178,12 +178,15 @@ export default class DbClient {
178178 }
179179 } )
180180 )
181- subdev = _subdev
181+ finalSubdev = _subdev
182+ }
183+ else if ( subdev ) {
184+ finalSubdev = subdev as SocketWrapper
182185 }
183186
184187 if ( Array . isArray ( push ) ) {
185188 const _push = zmqutil . socket ( 'push' )
186- Promise . all (
189+ await Promise . all (
187190 push . map ( async ( endpoint ) => {
188191 try {
189192 srv . attempt (
@@ -205,7 +208,7 @@ export default class DbClient {
205208
206209 if ( Array . isArray ( pushdev ) ) {
207210 const _pushdev = zmqutil . socket ( 'push' )
208- Promise . all (
211+ await Promise . all (
209212 pushdev . map ( async ( endpoint ) => {
210213 try {
211214 srv . attempt (
@@ -229,23 +232,48 @@ export default class DbClient {
229232 }
230233
231234 if ( ! channelRouter ) {
232- channelRouter = new EventEmitter ( ) ;
233- [ wireutil . global ] . forEach ( ( channel ) => {
234- _log . info ( 'Subscribing to permanent channel "%s"' , channel )
235- sub . subscribe ( channel )
236- subdev . subscribe ( channel )
237- } )
235+ channelRouter = new EventEmitter ( )
238236
239- sub . on ( 'message' , ( channel , data ) => {
240- channelRouter ?. emit ( channel . toString ( ) , channel , data )
241- } )
237+ if ( finalSub || finalSubdev ) {
238+ ; [ wireutil . global ] . forEach ( ( channel ) => {
239+ _log . info ( 'Subscribing to permanent channel "%s"' , channel )
240+ if ( finalSub ) {
241+ finalSub . subscribe ( channel )
242+ }
243+ if ( finalSubdev ) {
244+ finalSubdev . subscribe ( channel )
245+ }
246+ } )
247+
248+ if ( finalSub ) {
249+ finalSub . on ( 'message' , ( channel , data ) => {
250+ channelRouter ?. emit ( channel . toString ( ) , channel , data )
251+ } )
252+ }
242253
243- subdev . on ( 'message' , ( channel , data ) => {
244- channelRouter ?. emit ( channel . toString ( ) , channel , data )
245- } )
254+ if ( finalSubdev ) {
255+ finalSubdev . on ( 'message' , ( channel , data ) => {
256+ channelRouter ?. emit ( channel . toString ( ) , channel , data )
257+ } )
258+ }
259+ }
246260 }
247261
248- return { sub, subdev, push, pushdev, channelRouter}
262+ const result : {
263+ sub ?: SocketWrapper ;
264+ subdev ?: SocketWrapper ;
265+ push : SocketWrapper ;
266+ pushdev : SocketWrapper ;
267+ channelRouter : EventEmitter ;
268+ } = {
269+ push,
270+ pushdev,
271+ channelRouter,
272+ ... ! ! finalSub && { sub : finalSub } ,
273+ ... ! ! finalSubdev && { subdev : finalSubdev }
274+ }
275+
276+ return result
249277 }
250278
251279 // Verifies that we can form a connection. Useful if it's necessary to make
@@ -254,11 +282,12 @@ export default class DbClient {
254282 // an issue with the processor unit, as it started processing messages before
255283 // it was actually truly able to save anything to the database. This lead to
256284 // lost messages in certain situations.
257- static ensureConnectivity = async < T extends Function > ( fn : T ) => {
258- await DbClient . connect ( )
259- log . info ( "Db is up" )
260- return fn
261- }
285+ static ensureConnectivity = < T extends ( ...args : any [ ] ) => any > ( fn : T ) =>
286+ async ( ...args : Parameters < T > ) : Promise < ReturnType < T > > => {
287+ await DbClient . connect ( ) ;
288+ log . info ( "Db is up" ) ;
289+ return fn ( ...args ) ;
290+ }
262291
263292 // Sets up the database
264293 static setup = ( ) => DbClient . connect ( ) . then ( ( conn ) => _setup ( conn ) )
0 commit comments