Merge pull request #38 from oskarhane/1.0-bubble-socket-errors

Ability to observe socket errors on driver object

Author: pontusmelke

src/v1/driver.js

@@ -20,6 +20,7 @@
 import Session from './session';
 import {Pool} from './internal/pool';
 import {connect} from "./internal/connector";
+import StreamObserver from './internal/stream-observer';
 
 /**
   * A Driver instance is used for mananging {@link Session}s.
@@ -52,8 +53,9 @@ class Driver {
    */
   _createConnection( release ) {
     let sessionId = this._sessionIdGenerator++;
+    let streamObserver = new _ConnectionStreamObserver(this);
     let conn = connect(this._url);
-    conn.initialize(this._userAgent, this._token);
+    conn.initialize(this._userAgent, this._token, streamObserver);
     conn._id = sessionId;
     conn._release = () => release(conn);
 
@@ -121,4 +123,22 @@ class Driver {
   }
 }
 
+/** Internal stream observer used for connection state */
+class _ConnectionStreamObserver extends StreamObserver {
+  constructor(driver) {
+    super();
+    this._driver = driver;
+    this._hasFailed = false;
+  }
+  onError(error) {
+    if (!this._hasFailed) {
+      super.onError(error);
+      if(this._driver.onError) {
+        this._driver.onError(error);
+      }
+      this._hasFailed = true;
+    }
+  }
+}
+
 export default Driver

src/v1/internal/ch-websocket.js

@@ -55,9 +55,9 @@ class WebSocketChannel {
       } 
     };
 
-    this._ws.onerror = () => {
+    this._ws.onerror = (err) => {
       if( self.onerror ) {
-        self.onerror();
+        self.onerror(err);
       }
     }
   }

src/v1/internal/connector.js

@@ -207,7 +207,12 @@ class Connection {
       }
     };
 
-    this._ch.onerror = () => self._isBroken = true;
+    this._ch.onerror = (error) => {
+      self._isBroken = true;
+      if(this._currentObserver.onError) {
+        this._currentObserver.onError(error);
+      }
+    }
 
     this._dechunker.onmessage = (buf) => {
       self._handleMessage( self._unpacker.unpack( buf ) );
@@ -266,9 +271,9 @@ class Connection {
         break;
       case IGNORED:
         try {
-          if (this._errorMsg)
+          if (this._errorMsg && this._currentObserver.onError)
             this._currentObserver.onError(this._errorMsg);
-          else
+          else if(this._currentObserver.onError)
             this._currentObserver.onError(msg);
         } finally {
           this._currentObserver = this._pendingObservers.shift();