Wait for the reset command finishes before notify onComplete

Author: bigmontz

packages/bolt-connection/src/connection/connection-channel.js

@@ -123,6 +123,7 @@ export default class ChannelConnection extends Connection {
     super(errorHandler)
 
     this._reseting = false
+    this._resetObservers = []
     this._id = idGenerator++
     this._address = address
     this._server = { address: address.asHostPort() }
@@ -339,18 +340,24 @@ export default class ChannelConnection extends Connection {
   }
 
   _reset(observer) {
+    this._resetObservers.push(observer)
     if (this._reseting) {
-      observer.onComplete()
       return
     }
     this._reseting = true
+
+    const notifyFinish = (notify) => {
+      this._reseting = false
+      const observers = this._resetObservers
+      this._resetObservers = []
+      observers.forEach(notify)
+    }
+
     this._protocol.reset({
       onError: error => {
-        this._reseting = false
-        observer.onError(error)
+        notifyFinish(obs => obs.onError(error))
       }, onComplete: () => {
-        this._reseting = false
-        observer.onComplete()
+        notifyFinish(obs => obs.onComplete())
       }
     })
   }

packages/bolt-connection/test/connection/connection-channel.test.js

@@ -282,7 +282,7 @@ describe('ChannelConnection', () => {
         connection._resetOnFailure()
 
         expect(protocol.reset).toHaveBeenCalledTimes(1)
-        expect(protocol.resetFailure).toHaveBeenCalledTimes(1)
+        expect(protocol.resetFailure).not.toHaveBeenCalled()
       })
 
       it('should call protocol.reset() when after a previous reset completed', () => {
@@ -397,20 +397,23 @@ describe('ChannelConnection', () => {
       }
 
       const protocol = {
-        reset: jest.fn(),
+        reset: jest.fn(observer => {
+          setTimeout(() => observer.onComplete(), 100)
+        }),
         resetFailure: jest.fn()
       }
       const protocolSupplier = () => protocol
       const connection = spyOnConnectionChannel({ channel, protocolSupplier })
 
-      // to not block since the reset will never complete
-      connection.resetAndFlush()
+      const completeFirstResetAndFlush = connection.resetAndFlush()
 
       expect(protocol.reset).toHaveBeenCalledTimes(1)
 
       await connection.resetAndFlush()
 
       expect(protocol.reset).toHaveBeenCalledTimes(1)
+
+      await completeFirstResetAndFlush
     })
 
     it('should call protocol.reset() when after a previous reset completed', async () => {