15974960261

Committed 30 Jun 2025 01:57PM UTC coverage: 96.796% (+0.007%) from 96.789%

Build # 15974960261

Build Type

push

github

Committed by

web-flow

Commit Message

fix: make setup() in callBackPersistence async (#106)

* fix: make callback setup() async

* add test to verify that .setup() is async

Run Details

377 of 406 branches covered (92.86%)

Branch coverage included in aggregate %.

10 of 10 new or added lines in 2 files covered. (100.0%)

1 existing line in 1 file now uncovered.

2463 of 2528 relevant lines covered (97.43%)

346.38 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.33

/callBackPersistence.js

'use strict'

/* This module provides a callback layer for async persistence implementations */
const { Readable } = require('node:stream')
const { EventEmitter } = require('node:events')

class CallBackPersistence extends EventEmitter {
  constructor (asyncInstanceFactory, opts = {}) {
    super()

    this.ready = false
    this.asyncPersistence = asyncInstanceFactory(opts)
  }

  get broker () {
    return this.asyncPersistence.broker
  }

  set broker (broker) {
    if (this.ready) {
      return
    }

    this.asyncPersistence.setup(broker)
      .then(() => {
        this.ready = true
        this.emit('ready')
      })
      .catch(err => {
        this.emit('error', err)
      })
  }

  async setup (broker) {
    return this.asyncPersistence.setup(broker)
  }

  subscriptionsByTopic (topic, cb) {
    if (cb) {
      if (!this.ready) {
        this.once('ready', this.subscriptionsByTopic.bind(this, topic, cb))
        return this
      }
      this.asyncPersistence.subscriptionsByTopic(topic)
        .then(resubs => {
          process.nextTick(cb, null, resubs)
        })
        .catch(cb)
    } else {
      return this.asyncPersistence.subscriptionsByTopic(topic)
    }
  }

  cleanSubscriptions (client, cb) {
    if (cb) {
      this.asyncPersistence.cleanSubscriptions(client)
        .then(client => {
          process.nextTick(cb, null, client)
        })
        .catch(cb)
    } else {
      return this.asyncPersistence.cleanSubscriptions(client)
    }
  }

  storeRetained (packet, cb) {
    if (cb) {
      if (!this.ready) {
        this.once('ready', this.storeRetained.bind(this, packet, cb))
        return
      }
      this.asyncPersistence.storeRetained(packet).then(() => {
        cb(null)
      }).catch(cb)
    } else {
      return this.asyncPersistence.storeRetained(packet)
    }
  }

  createRetainedStream (pattern) {
    return Readable.from(this.asyncPersistence.createRetainedStream(pattern))
  }

  createRetainedStreamCombi (patterns) {
    return Readable.from(this.asyncPersistence.createRetainedStreamCombi(patterns))
  }

  addSubscriptions (client, subs, cb) {
    if (cb) {
      if (!this.ready) {
        this.once('ready', this.addSubscriptions.bind(this, client, subs, cb))
        return
      }
      this.asyncPersistence.addSubscriptions(client, subs)
        .then(() => cb(null, client))
        .catch(cb)
    } else {
      return this.asyncPersistence.addSubscriptions(client, subs)
    }
  }

  removeSubscriptions (client, subs, cb) {
    if (cb) {
      if (!this.ready) {
        this.once('ready', this.removeSubscriptions.bind(this, client, subs, cb))
        return
      }

      this.asyncPersistence.removeSubscriptions(client, subs)
        .then(() => cb(null, client))
        .catch(cb)
    } else {
      return this.asyncPersistence.removeSubscriptions(client, subs)
    }
  }

  subscriptionsByClient (client, cb) {
    if (cb) {
      if (!this.ready) {
        this.once('ready', this.subscriptionsByClient.bind(this, client, cb))
        return
      }

      this.asyncPersistence.subscriptionsByClient(client)
        .then(resubs => {
          process.nextTick(cb, null, resubs.length > 0 ? resubs : null, client)
        })
        .catch(cb)
    } else {
      return this.asyncPersistence.subscriptionsByClient(client)
    }
  }

  countOffline (cb) {
    if (cb) {
      this.asyncPersistence.countOffline()
        .then(res => process.nextTick(cb, null, res.subsCount, res.clientsCount))
        .catch(cb)
    } else {
      return this.asyncPersistence.countOffline()
    }
  }

  destroy (cb) {
    if (cb) {
      if (!this.ready) {
        this.once('ready', this.destroy.bind(this, cb))
        return
      }
      this.asyncPersistence.destroy()
        .finally(cb) // swallow err in case of failure
    } else {
      return this.asyncPersistence.destroy()
    }
  }

  outgoingEnqueue (sub, packet, cb) {
    if (cb) {
      if (!this.ready) {
        this.once('ready', this.outgoingEnqueue.bind(this, sub, packet, cb))
        return
      }
      this.asyncPersistence.outgoingEnqueue(sub, packet)
        .then(() => process.nextTick(cb, null, packet))
        .catch(cb)
    } else {
      return this.asyncPersistence.outgoingEnqueue(sub, packet)
    }
  }

  outgoingEnqueueCombi (subs, packet, cb) {
    if (cb) {
      if (!this.ready) {
        this.once('ready', this.outgoingEnqueueCombi.bind(this, subs, packet, cb))
        return
      }
      this.asyncPersistence.outgoingEnqueueCombi(subs, packet)
        .then(() => process.nextTick(cb, null, packet))
        .catch(cb)
    } else {
      return this.asyncPersistence.outgoingEnqueueCombi(subs, packet)
    }
  }

  outgoingStream (client) {
    return Readable.from(this.asyncPersistence.outgoingStream(client))
  }

  outgoingUpdate (client, packet, cb) {
    if (cb) {
      if (!this.ready) {
        this.once('ready', this.outgoingUpdate.bind(this, client, packet, cb))
        return
      }
      this.asyncPersistence.outgoingUpdate(client, packet)
        .then(() => cb(null, client, packet))
        .catch(cb)
    } else {
      return this.asyncPersistence.outgoingUpdate(client, packet)
    }
  }

  outgoingClearMessageId (client, packet, cb) {
    if (cb) {
      if (!this.ready) {
        this.once('ready', this.outgoingClearMessageId.bind(this, client, packet, cb))
        return
      }
      this.asyncPersistence.outgoingClearMessageId(client, packet)
        .then((packet) => cb(null, packet))
        .catch(cb)
    } else {
      return this.asyncPersistence.outgoingClearMessageId(client, packet)
    }
  }

  incomingStorePacket (client, packet, cb) {
    if (cb) {
      if (!this.ready) {
        this.once('ready', this.incomingStorePacket.bind(this, client, packet, cb))
        return
      }
      this.asyncPersistence.incomingStorePacket(client, packet)
        .then(() => cb(null))
        .catch(cb)
    } else {
      return this.asyncPersistence.incomingStorePacket(client, packet)
    }
  }

  incomingGetPacket (client, packet, cb) {
    if (cb) {
      if (!this.ready) {
        this.once('ready', this.incomingGetPacket.bind(this, client, packet, cb))
        return
      }
      this.asyncPersistence.incomingGetPacket(client, packet)
        .then((packet) => cb(null, packet, client))
        .catch(cb)
    } else {
      return this.asyncPersistence.incomingGetPacket(client, packet)
    }
  }

  incomingDelPacket (client, packet, cb) {
    if (cb) {
      if (!this.ready) {
        this.once('ready', this.incomingDelPacket.bind(this, client, packet, cb))
        return
      }
      this.asyncPersistence.incomingDelPacket(client, packet)
        .then(() => cb(null))
        .catch(cb)
    } else {
      return this.asyncPersistence.incomingDelPacket(client, packet)
    }
  }

  putWill (client, packet, cb) {
    if (cb) {
      if (!this.ready) {
        this.once('ready', this.putWill.bind(this, client, packet, cb))
        return
      }
      this.asyncPersistence.putWill(client, packet)
        .then(() => cb(null, client))
        .catch(cb)
    } else {
      return this.asyncPersistence.putWill(client, packet)
    }
  }

  getWill (client, cb) {
    if (cb) {
      this.asyncPersistence.getWill(client)
        .then(packet => {
          cb(null, packet, client)
        })
        .catch(cb)
    } else {
      return this.asyncPersistence.getWill(client)
    }
  }

  delWill (client, cb) {
    if (cb) {
      this.asyncPersistence.delWill(client)
        .then(packet => {
          cb(null, packet, client)
        })
        .catch(cb)
    } else {
      return this.asyncPersistence.delWill(client)
    }
  }

  streamWill (brokers) {
    return Readable.from(this.asyncPersistence.streamWill(brokers))
  }

  getClientList (topic) {
    return Readable.from(this.asyncPersistence.getClientList(topic))
  }
}

module.exports = { CallBackPersistence }

1	'use strict'	12✔
2		12✔
3	/* This module provides a callback layer for async persistence implementations */	12✔
4	const { Readable } = require('node:stream')	12✔
5	const { EventEmitter } = require('node:events')	12✔
6		12✔
7	class CallBackPersistence extends EventEmitter {	12✔
8	constructor (asyncInstanceFactory, opts = {}) {	12✔
9	super()	705✔
10		705✔
11	this.ready = false	705✔
12	this.asyncPersistence = asyncInstanceFactory(opts)	705✔
13	}	705✔
14		12✔
15	get broker () {	12✔
16	return this.asyncPersistence.broker	30,600✔
17	}	30,600✔
18		12✔
19	set broker (broker) {	12✔
20	if (this.ready) {	423!
UNCOV 21	return	×
22	}	×
23		423✔
24	this.asyncPersistence.setup(broker)	423✔
25	.then(() => {	423✔
26	this.ready = true	423✔
27	this.emit('ready')	423✔
28	})	423✔
29	.catch(err => {	423✔
30	this.emit('error', err)	×
31	})	423✔
32	}	423✔
33		12✔
34	async setup (broker) {	12✔
35	return this.asyncPersistence.setup(broker)	282✔
36	}	282✔
37		12✔
38	subscriptionsByTopic (topic, cb) {	12✔
39	if (cb) {	195✔
40	if (!this.ready) {	117!
41	this.once('ready', this.subscriptionsByTopic.bind(this, topic, cb))	×
42	return this	×
43	}	×
44	this.asyncPersistence.subscriptionsByTopic(topic)	117✔
45	.then(resubs => {	117✔
46	process.nextTick(cb, null, resubs)	117✔
47	})	117✔
48	.catch(cb)	117✔
49	} else {	195✔
50	return this.asyncPersistence.subscriptionsByTopic(topic)	78✔
51	}	78✔
52	}	195✔
53		12✔
54	cleanSubscriptions (client, cb) {	12✔
55	if (cb) {	30✔
56	this.asyncPersistence.cleanSubscriptions(client)	18✔
57	.then(client => {	18✔
58	process.nextTick(cb, null, client)	18✔
59	})	18✔
60	.catch(cb)	18✔
61	} else {	24✔
62	return this.asyncPersistence.cleanSubscriptions(client)	12✔
63	}	12✔
64	}	30✔
65		12✔
66	storeRetained (packet, cb) {	12✔
67	if (cb) {	15,210✔
68	if (!this.ready) {	9,126!
69	this.once('ready', this.storeRetained.bind(this, packet, cb))	×
70	return	×
71	}	×
72	this.asyncPersistence.storeRetained(packet).then(() => {	9,126✔
73	cb(null)	9,126✔
74	}).catch(cb)	9,126✔
75	} else {	15,210✔
76	return this.asyncPersistence.storeRetained(packet)	6,084✔
77	}	6,084✔
78	}	15,210✔
79		12✔
80	createRetainedStream (pattern) {	12✔
81	return Readable.from(this.asyncPersistence.createRetainedStream(pattern))	150✔
82	}	150✔
83		12✔
84	createRetainedStreamCombi (patterns) {	12✔
85	return Readable.from(this.asyncPersistence.createRetainedStreamCombi(patterns))	30✔
86	}	30✔
87		12✔
88	addSubscriptions (client, subs, cb) {	12✔
89	if (cb) {	405✔
90	if (!this.ready) {	243!
91	this.once('ready', this.addSubscriptions.bind(this, client, subs, cb))	×
92	return	×
93	}	×
94	this.asyncPersistence.addSubscriptions(client, subs)	243✔
95	.then(() => cb(null, client))	243✔
96	.catch(cb)	243✔
97	} else {	405✔
98	return this.asyncPersistence.addSubscriptions(client, subs)	162✔
99	}	162✔
100	}	405✔
101		12✔
102	removeSubscriptions (client, subs, cb) {	12✔
103	if (cb) {	225✔
104	if (!this.ready) {	135!
105	this.once('ready', this.removeSubscriptions.bind(this, client, subs, cb))	×
106	return	×
107	}	×
108		135✔
109	this.asyncPersistence.removeSubscriptions(client, subs)	135✔
110	.then(() => cb(null, client))	135✔
111	.catch(cb)	135✔
112	} else {	225✔
113	return this.asyncPersistence.removeSubscriptions(client, subs)	90✔
114	}	90✔
115	}	225✔
116		12✔
117	subscriptionsByClient (client, cb) {	12✔
118	if (cb) {	210✔
119	if (!this.ready) {	126!
120	this.once('ready', this.subscriptionsByClient.bind(this, client, cb))	×
121	return	×
122	}	×
123		126✔
124	this.asyncPersistence.subscriptionsByClient(client)	126✔
125	.then(resubs => {	126✔
126	process.nextTick(cb, null, resubs.length > 0 ? resubs : null, client)	126✔
127	})	126✔
128	.catch(cb)	126✔
129	} else {	210✔
130	return this.asyncPersistence.subscriptionsByClient(client)	84✔
131	}	84✔
132	}	210✔
133		12✔
134	countOffline (cb) {	12✔
135	if (cb) {	330✔
136	this.asyncPersistence.countOffline()	198✔
137	.then(res => process.nextTick(cb, null, res.subsCount, res.clientsCount))	198✔
138	.catch(cb)	198✔
139	} else {	264✔
140	return this.asyncPersistence.countOffline()	132✔
141	}	132✔
142	}	330✔
143		12✔
144	destroy (cb) {	12✔
145	if (cb) {	690✔
146	if (!this.ready) {	414!
147	this.once('ready', this.destroy.bind(this, cb))	×
148	return	×
149	}	×
150	this.asyncPersistence.destroy()	414✔
151	.finally(cb) // swallow err in case of failure	414✔
152	} else {	690✔
153	return this.asyncPersistence.destroy()	276✔
154	}	276✔
155	}	690✔
156		12✔
157	outgoingEnqueue (sub, packet, cb) {	12✔
158	if (cb) {	75✔
159	if (!this.ready) {	45!
160	this.once('ready', this.outgoingEnqueue.bind(this, sub, packet, cb))	×
161	return	×
162	}	×
163	this.asyncPersistence.outgoingEnqueue(sub, packet)	45✔
164	.then(() => process.nextTick(cb, null, packet))	45✔
165	.catch(cb)	45✔
166	} else {	75✔
167	return this.asyncPersistence.outgoingEnqueue(sub, packet)	30✔
168	}	30✔
169	}	75✔
170		12✔
171	outgoingEnqueueCombi (subs, packet, cb) {	12✔
172	if (cb) {	135✔
173	if (!this.ready) {	81!
174	this.once('ready', this.outgoingEnqueueCombi.bind(this, subs, packet, cb))	×
175	return	×
176	}	×
177	this.asyncPersistence.outgoingEnqueueCombi(subs, packet)	81✔
178	.then(() => process.nextTick(cb, null, packet))	81✔
179	.catch(cb)	81✔
180	} else {	135✔
181	return this.asyncPersistence.outgoingEnqueueCombi(subs, packet)	54✔
182	}	54✔
183	}	135✔
184		12✔
185	outgoingStream (client) {	12✔
186	return Readable.from(this.asyncPersistence.outgoingStream(client))	225✔
187	}	225✔
188		12✔
189	outgoingUpdate (client, packet, cb) {	12✔
190	if (cb) {	195✔
191	if (!this.ready) {	117!
192	this.once('ready', this.outgoingUpdate.bind(this, client, packet, cb))	×
193	return	×
194	}	×
195	this.asyncPersistence.outgoingUpdate(client, packet)	117✔
196	.then(() => cb(null, client, packet))	117✔
197	.catch(cb)	117✔
198	} else {	195✔
199	return this.asyncPersistence.outgoingUpdate(client, packet)	78✔
200	}	78✔
201	}	195✔
202		12✔
203	outgoingClearMessageId (client, packet, cb) {	12✔
204	if (cb) {	60✔
205	if (!this.ready) {	36!
206	this.once('ready', this.outgoingClearMessageId.bind(this, client, packet, cb))	×
207	return	×
208	}	×
209	this.asyncPersistence.outgoingClearMessageId(client, packet)	36✔
210	.then((packet) => cb(null, packet))	36✔
211	.catch(cb)	36✔
212	} else {	60✔
213	return this.asyncPersistence.outgoingClearMessageId(client, packet)	24✔
214	}	24✔
215	}	60✔
216		12✔
217	incomingStorePacket (client, packet, cb) {	12✔
218	if (cb) {	15✔
219	if (!this.ready) {	9!
220	this.once('ready', this.incomingStorePacket.bind(this, client, packet, cb))	×
221	return	×
222	}	×
223	this.asyncPersistence.incomingStorePacket(client, packet)	9✔
224	.then(() => cb(null))	9✔
225	.catch(cb)	9✔
226	} else {	15✔
227	return this.asyncPersistence.incomingStorePacket(client, packet)	6✔
228	}	6✔
229	}	15✔
230		12✔
231	incomingGetPacket (client, packet, cb) {	12✔
232	if (cb) {	30✔
233	if (!this.ready) {	18!
234	this.once('ready', this.incomingGetPacket.bind(this, client, packet, cb))	×
235	return	×
236	}	×
237	this.asyncPersistence.incomingGetPacket(client, packet)	18✔
238	.then((packet) => cb(null, packet, client))	18✔
239	.catch(cb)	18✔
240	} else {	30✔
241	return this.asyncPersistence.incomingGetPacket(client, packet)	12✔
242	}	12✔
243	}	30✔
244		12✔
245	incomingDelPacket (client, packet, cb) {	12✔
246	if (cb) {	15✔
247	if (!this.ready) {	9!
248	this.once('ready', this.incomingDelPacket.bind(this, client, packet, cb))	×
249	return	×
250	}	×
251	this.asyncPersistence.incomingDelPacket(client, packet)	9✔
252	.then(() => cb(null))	9✔
253	.catch(cb)	9✔
254	} else {	15✔
255	return this.asyncPersistence.incomingDelPacket(client, packet)	6✔
256	}	6✔
257	}	15✔
258		12✔
259	putWill (client, packet, cb) {	12✔
260	if (cb) {	75✔
261	if (!this.ready) {	45!
262	this.once('ready', this.putWill.bind(this, client, packet, cb))	×
263	return	×
264	}	×
265	this.asyncPersistence.putWill(client, packet)	45✔
266	.then(() => cb(null, client))	45✔
267	.catch(cb)	45✔
268	} else {	75✔
269	return this.asyncPersistence.putWill(client, packet)	30✔
270	}	30✔
271	}	75✔
272		12✔
273	getWill (client, cb) {	12✔
274	if (cb) {	30✔
275	this.asyncPersistence.getWill(client)	18✔
276	.then(packet => {	18✔
277	cb(null, packet, client)	18✔
278	})	18✔
279	.catch(cb)	18✔
280	} else {	24✔
281	return this.asyncPersistence.getWill(client)	12✔
282	}	12✔
283	}	30✔
284		12✔
285	delWill (client, cb) {	12✔
286	if (cb) {	60✔
287	this.asyncPersistence.delWill(client)	36✔
288	.then(packet => {	36✔
289	cb(null, packet, client)	36✔
290	})	36✔
291	.catch(cb)	36✔
292	} else {	48✔
293	return this.asyncPersistence.delWill(client)	24✔
294	}	24✔
295	}	60✔
296		12✔
297	streamWill (brokers) {	12✔
298	return Readable.from(this.asyncPersistence.streamWill(brokers))	30✔
299	}	30✔
300		12✔
301	getClientList (topic) {	12✔
302	return Readable.from(this.asyncPersistence.getClientList(topic))	30✔
303	}	30✔
304	}	12✔
305		12✔
306	module.exports = { CallBackPersistence }	12✔

moscajs / aedes-persistence / 15974960261

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous