Treat errors on handling income bolt messages (#957)

Invalid income messages which could not be parsed or handled correctly where throwing exceptions in non-safe context. This leads the driver to crash hard since the error is bubble up to the socket thread.

Author: bigmontz

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

@@ -70,7 +70,11 @@ export default function create ({
 
     // setup dechunker to dechunk messages and forward them to the message handler
     dechunker.onmessage = buf => {
-      responseHandler.handleResponse(protocol.unpacker().unpack(buf))
+      try {
+        responseHandler.handleResponse(protocol.unpacker().unpack(buf))
+      } catch (e) {
+        return observer.onError(e)
+      }
     }
 
     return responseHandler

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

@@ -275,6 +275,65 @@ describe('#unit Bolt', () => {
         expect(receivedMessage.signature).toEqual(expectedMessage.signature)
         expect(receivedMessage.fields).toEqual(expectedMessage.fields)
       })
+
+      it(`it should configure the channel.onmessage to dechunk and notify unpacking issues ${version}`, () => {
+        const params = createBoltCreateParams({ version })
+        let receivedMessage = null
+        const message = {
+          signature: 0x10,
+          fields: [123]
+        }
+        const protocol = Bolt.create(params)
+        protocol._responseHandler.handleResponse = msg => {
+          receivedMessage = msg
+        }
+
+        protocol.packer().packStruct(
+          message.signature,
+          message.fields.map(field => protocol.packer().packable(field))
+        )
+
+        params.chunker.messageBoundary()
+        params.chunker.flush()
+
+        const expectedError = newError('Something went wrong')
+        protocol.unpacker = jest.fn(() => {
+          return {
+            unpack: () => { 
+              throw expectedError
+            }
+          }
+        })
+
+        params.channel.onmessage(params.channel.toBuffer())
+
+        expect(receivedMessage).toBeNull()
+        expect(params.observer.errors).toEqual([expectedError])
+      })
+
+      it(`it should configure the channel.onmessage to dechunk and notify response handler issues ${version}`, () => {
+        const params = createBoltCreateParams({ version })
+        let receivedMessage = null
+        const expectedMessage = {
+          signature: 0x10,
+          fields: [123]
+        }
+        const protocol = Bolt.create(params)
+        const expectedError = newError('Something went wrong')
+        protocol._responseHandler.handleResponse = msg => {
+          throw expectedError
+        }
+
+        protocol.packer().packStruct(
+          expectedMessage.signature,
+          expectedMessage.fields.map(field => protocol.packer().packable(field))
+        )
+        params.chunker.messageBoundary()
+        params.chunker.flush()
+        params.channel.onmessage(params.channel.toBuffer())
+
+        expect(params.observer.errors).toEqual([expectedError])
+      })
     })
 
     forEachUnknownProtocolVersion(version => {