Fix(publish): call callback when messageId available (mqttjs#1393)

redboltz · web-flow · commit 58fb8dfdc3ca · 2022-01-05T09:16:11.000-08:00
Process is enqueued if messageId allocation is failure.
diff --git a/lib/client.js b/lib/client.js
@@ -660,16 +660,14 @@ MqttClient.prototype.publish = function (topic, message, opts, callback) {
     return true
   }
 
-  if (this._storeProcessing || this._storeProcessingQueue.length > 0) {
+  if (this._storeProcessing || this._storeProcessingQueue.length > 0 || !publishProc()) {
     this._storeProcessingQueue.push(
       {
         invoke: publishProc,
         cbStorePut: opts.cbStorePut,
         callback: callback
       }
     )
-  } else {
-    publishProc()
   }
   return this
 }
@@ -842,15 +840,13 @@ MqttClient.prototype.subscribe = function () {
     return true
   }
 
-  if (this._storeProcessing || this._storeProcessingQueue.length > 0) {
+  if (this._storeProcessing || this._storeProcessingQueue.length > 0 || !subscribeProc()) {
     this._storeProcessingQueue.push(
       {
         invoke: subscribeProc,
         callback: callback
       }
     )
-  } else {
-    subscribeProc()
   }
 
   return this
@@ -935,15 +931,13 @@ MqttClient.prototype.unsubscribe = function () {
     return true
   }
 
-  if (this._storeProcessing || this._storeProcessingQueue.length > 0) {
+  if (this._storeProcessing || this._storeProcessingQueue.length > 0 || !unsubscribeProc()) {
     this._storeProcessingQueue.push(
       {
         invoke: unsubscribeProc,
         callback: callback
       }
     )
-  } else {
-    unsubscribeProc()
   }
 
   return this

Original file line number	Diff line number	Diff line change
`@@ -660,16 +660,14 @@ MqttClient.prototype.publish = function (topic, message, opts, callback) {`
`660`	`660`	`return true`
`661`	`661`	`}`
`662`	`662`
`663`		`- if (this._storeProcessing \|\| this._storeProcessingQueue.length > 0) {`
	`663`	`+ if (this._storeProcessing \|\| this._storeProcessingQueue.length > 0 \|\| !publishProc()) {`
`664`	`664`	`this._storeProcessingQueue.push(`
`665`	`665`	`{`
`666`	`666`	`invoke: publishProc,`
`667`	`667`	`cbStorePut: opts.cbStorePut,`
`668`	`668`	`callback: callback`
`669`	`669`	`}`
`670`	`670`	`)`
`671`		`- } else {`
`672`		`- publishProc()`
`673`	`671`	`}`
`674`	`672`	`return this`
`675`	`673`	`}`
`@@ -842,15 +840,13 @@ MqttClient.prototype.subscribe = function () {`
`842`	`840`	`return true`
`843`	`841`	`}`
`844`	`842`
`845`		`- if (this._storeProcessing \|\| this._storeProcessingQueue.length > 0) {`
	`843`	`+ if (this._storeProcessing \|\| this._storeProcessingQueue.length > 0 \|\| !subscribeProc()) {`
`846`	`844`	`this._storeProcessingQueue.push(`
`847`	`845`	`{`
`848`	`846`	`invoke: subscribeProc,`
`849`	`847`	`callback: callback`
`850`	`848`	`}`
`851`	`849`	`)`
`852`		`- } else {`
`853`		`- subscribeProc()`
`854`	`850`	`}`
`855`	`851`
`856`	`852`	`return this`
`@@ -935,15 +931,13 @@ MqttClient.prototype.unsubscribe = function () {`
`935`	`931`	`return true`
`936`	`932`	`}`
`937`	`933`
`938`		`- if (this._storeProcessing \|\| this._storeProcessingQueue.length > 0) {`
	`934`	`+ if (this._storeProcessing \|\| this._storeProcessingQueue.length > 0 \|\| !unsubscribeProc()) {`
`939`	`935`	`this._storeProcessingQueue.push(`
`940`	`936`	`{`
`941`	`937`	`invoke: unsubscribeProc,`
`942`	`938`	`callback: callback`
`943`	`939`	`}`
`944`	`940`	`)`
`945`		`- } else {`
`946`		`- unsubscribeProc()`
`947`	`941`	`}`
`948`	`942`
`949`	`943`	`return this`