@@ -139,5 +139,47 @@ module.exports = {
139139
140140 return asyncClient ;
141141 } ,
142+ connectAsync ( brokerURL , opts , allowRetries = true ) {
143+ const client = mqtt . connect ( brokerURL , opts ) ;
144+ const asyncClient = new AsyncClient ( client ) ;
145+
146+ return new Promise ( ( resolve , reject ) => {
147+ // Listeners added to client to trigger promise resolution
148+ const promiseResolutionListeners = {
149+ connect : ( connack ) => {
150+ removePromiseResolutionListeners ( ) ;
151+ resolve ( asyncClient ) ; // Resolve on connect
152+ } ,
153+ end : ( ) => {
154+ removePromiseResolutionListeners ( ) ;
155+ resolve ( asyncClient ) ; // Resolve on end
156+ } ,
157+ error : ( err ) => {
158+ removePromiseResolutionListeners ( ) ;
159+ client . end ( ) ;
160+ reject ( err ) ; // Reject on error
161+ }
162+ } ;
163+
164+ // If retries are not allowed, reject on close
165+ if ( allowRetries === false ) {
166+ promiseResolutionListeners . close = ( ) => {
167+ promiseResolutionListeners . error ( "Couldn't connect to server" ) ;
168+ }
169+ }
170+
171+ // Remove listeners added to client by this promise
172+ const removePromiseResolutionListeners = ( ) => {
173+ Object . keys ( promiseResolutionListeners ) . forEach ( ( eventName ) => {
174+ client . off ( eventName , promiseResolutionListeners [ eventName ] ) ;
175+ } ) ;
176+ } ;
177+
178+ // Add listeners to client
179+ Object . keys ( promiseResolutionListeners ) . forEach ( ( eventName ) => {
180+ client . on ( eventName , promiseResolutionListeners [ eventName ] ) ;
181+ } ) ;
182+ } ) ;
183+ } ,
142184 AsyncClient
143185} ;
0 commit comments