replace current websocket implementation with engine.io, made tests work in node without mocking window

client/package.json

@@ -16,9 +16,12 @@
   "dependencies": {
     "babel-runtime": "^6.6.1",
     "core-js": "^2.1.0",
+    "engine.io-client": "^1.6.9",
+    "engine.io-parser": "socketio/engine.io-parser#748144b50a1d10e8c7c9b8100f2e21f8ac424c7a",
     "isomorphic-fetch": "^2.2.1",
     "rxjs": "^5.0.0-beta.6",
-    "snake-case": "^1.1.2"
+    "snake-case": "^1.1.2",
+    "ws": "1.0.1"
   },
   "engines": {
     "node": ">=4.0.0"
@@ -43,8 +46,7 @@
     "mocha": "2.3.4",
     "nodemon": "^1.9.1",
     "source-map-support": "^0.4.0",
-    "webpack": "^1.12.14",
-    "ws": "^1.0.1"
+    "webpack": "^1.12.14"
   },
   "main": "lib/index.js",
   "repository": {

client/src/auth.js

@@ -91,7 +91,8 @@ export class TokenStorage {
   }
 
   setAuthFromQueryParams() {
-    const parsed = queryParse(window.location.search)
+    const parsed = typeof window !== 'undefined' ? queryParse(window.location.search) : {}
+
     if (parsed.horizon_auth != null) {
       this.set(parsed.horizon_auth)
     }

client/src/index.js

@@ -81,6 +81,12 @@ function Horizon({
   horizon.onSocketError = subscribeOrObservable(
     socket.status::filter(x => x.type === 'error'))
 
+  horizon.utensils = {
+    sendRequest,
+    tokenStorage,
+  }
+  Object.freeze(horizon.utensils)
+
   horizon._authMethods = null
   horizon._horizonPath = path
   horizon.authEndpoint = authEndpoint

client/src/socket.js

@@ -6,16 +6,20 @@ import { merge } from 'rxjs/observable/merge'
 import { filter } from 'rxjs/operator/filter'
 import { share } from 'rxjs/operator/share'
 
-import { WebSocket } from './shim.js'
 import { serialize, deserialize } from './serialization.js'
 import { log } from './logging.js'
+import socket from 'engine.io-client'
+
+let transports
+
+if (typeof window === 'undefined') {
+  transports = [ 'websocket' ]
+}
 
 const PROTOCOL_VERSION = 'rethinkdb-horizon-v0'
 
 // Before connecting the first time
 const STATUS_UNCONNECTED = { type: 'unconnected' }
-// After the websocket is opened, but before handshake
-const STATUS_CONNECTED = { type: 'connected' }
 // After the websocket is opened and handshake is completed
 const STATUS_READY = { type: 'ready' }
 // After unconnected, maybe before or after connected. Any socket level error
@@ -42,43 +46,68 @@ class ProtocolError extends Error {
 // observable is closed.
 class HorizonSocket extends Subject {
   constructor(host, secure, path, handshaker) {
-    const hostString = `ws${secure ? 's' : ''}://${host}/${path}`
+    const hostString = `ws${secure ? 's' : ''}://${host}`
     const msgBuffer = []
     let ws, handshakeDisp
     // Handshake is an asyncsubject because we want it to always cache
     // the last value it received, like a promise
     const handshake = new AsyncSubject()
     const statusSubject = new BehaviorSubject(STATUS_UNCONNECTED)
 
-    const isOpen = () => Boolean(ws) && ws.readyState === WebSocket.OPEN
+    const isOpen = () => Boolean(ws) && ws.readyState === 'open'
 
     // Serializes to a string before sending
     function wsSend(msg) {
       const stringMsg = JSON.stringify(serialize(msg))
+
       ws.send(stringMsg)
     }
 
     // This is the observable part of the Subject. It forwards events
     // from the underlying websocket
     const socketObservable = Observable.create(subscriber => {
-      ws = new WebSocket(hostString, PROTOCOL_VERSION)
-      ws.onerror = () => {
+      ws = socket(hostString, {
+        protocol: PROTOCOL_VERSION,
+        path: `/${path}`,
+        transports
+      })
+
+      ws.on('error', () => {
         // If the websocket experiences the error, we forward it through
         // to the observable. Unfortunately, the event we receive in
         // this callback doesn't tell us much of anything, so there's no
         // reason to forward it on and we just send a generic error.
         statusSubject.next(STATUS_ERROR)
         const errMsg = `Websocket ${hostString} experienced an error`
         subscriber.error(new Error(errMsg))
-      }
-      ws.onopen = () => {
+      })
+
+      ws.on('open', () => {
+        ws.on('message', data => {
+          const deserialized = deserialize(JSON.parse(data))
+          log('Received', deserialized)
+          subscriber.next(deserialized)
+        })
+
+        ws.on('close', e => {
+          // This will happen if the socket is closed by the server If
+          // .close is called from the client (see closeSocket), this
+          // listener will be removed
+          statusSubject.next(STATUS_DISCONNECTED)
+          if (e !== 'forced close') {
+            subscriber.error(
+              new Error(`Socket closed unexpectedly with code: ${e}`)
+            )
+          } else {
+            subscriber.complete()
+          }
+        })
+
         // Send the handshake
-        statusSubject.next(STATUS_CONNECTED)
         handshakeDisp = this.makeRequest(handshaker()).subscribe(
           x => {
             handshake.next(x)
             handshake.complete()
-
             statusSubject.next(STATUS_READY)
           },
           err => handshake.error(err),
@@ -90,24 +119,7 @@ class HorizonSocket extends Subject {
           log('Sending buffered:', msg)
           wsSend(msg)
         }
-      }
-      ws.onmessage = event => {
-        const deserialized = deserialize(JSON.parse(event.data))
-        log('Received', deserialized)
-        subscriber.next(deserialized)
-      }
-      ws.onclose = e => {
-        // This will happen if the socket is closed by the server If
-        // .close is called from the client (see closeSocket), this
-        // listener will be removed
-        statusSubject.next(STATUS_DISCONNECTED)
-        if (e.code !== 1000 || !e.wasClean) {
-          subscriber.error(
-            new Error(`Socket closed unexpectedly with code: ${e.code}`))
-        } else {
-          subscriber.complete()
-        }
-      }
+      })
       return () => {
         if (handshakeDisp) {
           handshakeDisp.unsubscribe()
@@ -157,9 +169,6 @@ class HorizonSocket extends Subject {
       } else {
         ws.close(code, reason)
       }
-      ws.onopen = undefined
-      ws.onclose = undefined
-      ws.onmessage = undefined
     }
 
     super(socketSubscriber, socketObservable)

client/src/util/glob.js

@@ -0,0 +1,9 @@
+module.exports = function glob() {
+  return typeof self !== 'undefined' ?
+  self :
+  typeof window !== 'undefined' ?
+  window :
+  typeof global !== 'undefined' ?
+  global :
+  {}
+}

client/test/above.js

@@ -3,7 +3,7 @@ import { toArray } from 'rxjs/operator/toArray'
 
 import { assertCompletes, assertThrows, assertErrors } from './utils'
 
-const aboveSuite = window.aboveSuite = (getData) => () => {
+const aboveSuite = global.aboveSuite = (getData) => () => {
   let data
 
   before(() => {

client/test/aboveSub.js

@@ -2,7 +2,7 @@ import { concat } from 'rxjs/operator/concat'
 
 import { assertCompletes, observableInterleave } from './utils'
 
-const aboveSubscriptionSuite = window.aboveSubscriptionSuite = getData => () => {
+const aboveSubscriptionSuite = global.aboveSubscriptionSuite = getData => () => {
   let data
 
   before(() => {

client/test/auth.js

@@ -1,6 +1,6 @@
 import { TokenStorage, FakeStorage } from '../src/auth'
 
-const authSuite = window.authSuite = () => {
+const authSuite = global.authSuite = () => {
   describe('TokenStorage', () => {
     let fakeStorage
     let tokenStore

client/test/below.js

@@ -3,7 +3,7 @@ import { toArray } from 'rxjs/operator/toArray'
 
 import { assertCompletes, assertThrows, assertErrors } from './utils'
 
-const belowSuite = window.belowSuite = (getData) => () => {
+const belowSuite = global.belowSuite = (getData) => () => {
   let data
 
   before(() => {

client/test/belowSub.js

@@ -2,7 +2,7 @@ import { concat } from 'rxjs/operator/concat'
 
 import { assertCompletes, observableInterleave } from './utils'
 
-const belowSubscriptionSuite = window.belowSubscriptionSuite = getData => () => {
+const belowSubscriptionSuite = global.belowSubscriptionSuite = getData => () => {
   let data
 
   before(() => {

client/test/chaining.js

@@ -3,7 +3,7 @@ import { toArray } from 'rxjs/operator/toArray'
 
 import { assertCompletes, assertThrows } from './utils'
 
-const chainingSuite = window.chainingSuite = getData => () => {
+const chainingSuite = global.chainingSuite = getData => () => {
   let data
 
   before(() => {

client/test/collection.js

@@ -3,7 +3,7 @@ import { toArray } from 'rxjs/operator/toArray'
 
 import { assertCompletes, removeAllData } from './utils'
 
-const collectionSuite = window.collectionSuite = (getHorizon, getData, getTestData) => () => {
+const collectionSuite = global.collectionSuite = (getHorizon, getData, getTestData) => () => {
   let horizon, data, testData, empty_collection
 
   before(() => {

client/test/find.js

@@ -3,7 +3,7 @@ import { toArray } from 'rxjs/operator/toArray'
 
 import { assertCompletes, assertThrows, assertErrors } from './utils'
 
-const findSuite = window.findSuite = getData => () => {
+const findSuite = global.findSuite = getData => () => {
   let data
 
   before(() => {

client/test/findAll.js

@@ -3,7 +3,7 @@ import { toArray } from 'rxjs/operator/toArray'
 
 import { assertCompletes, assertThrows, assertErrors } from './utils'
 
-const findAllSuite = window.findAllSuite = getData => () => {
+const findAllSuite = global.findAllSuite = getData => () => {
   let data
 
   before(() => {

client/test/findAllSub.js

@@ -2,7 +2,7 @@ import { concat } from 'rxjs/operator/concat'
 
 import { assertCompletes, observableInterleave } from './utils'
 
-const findAllSubscriptionSuite = window.findAllSubscriptionSuite = getData => () => {
+const findAllSubscriptionSuite = global.findAllSubscriptionSuite = getData => () => {
   let data
 
   before(() => {

client/test/findSub.js

@@ -2,7 +2,7 @@ import { concat } from 'rxjs/operator/concat'
 
 import { assertCompletes, observableInterleave } from './utils'
 
-const findSubscriptionSuite = window.findSubscriptionSuite = getData => () => {
+const findSubscriptionSuite = global.findSubscriptionSuite = getData => () => {
   let data
 
   before(() => {

client/test/horizonObject.js

@@ -3,7 +3,7 @@
 // Test object creation, the `disconnect` method, and `connected/disconnected`
 // events.
 
-var horizonObjectSuite = window.horizonObjectSuite = () => {
+var horizonObjectSuite = global.horizonObjectSuite = () => {
   describe('Horizon', () => {
     it('connects and can track its status', done => {
       Horizon.clearAuthTokens()
@@ -14,7 +14,7 @@ var horizonObjectSuite = window.horizonObjectSuite = () => {
           switch (stat.type) {
           case 'unconnected':
             break
-          case 'connected':
+          case 'ready':
             horizon.disconnect()
             break
           case 'error':

client/test/insert.js

@@ -4,7 +4,7 @@ import { mergeMap } from 'rxjs/operator/mergeMap'
 import { toArray } from 'rxjs/operator/toArray'
 
 import { assertCompletes, assertThrows, assertErrors } from './utils'
-const insertSuite = window.insertSuite = getData => () => {
+const insertSuite = global.insertSuite = getData => () => {
   let data
 
   before(() => {

client/test/limit.js

@@ -3,7 +3,7 @@ import { toArray } from 'rxjs/operator/toArray'
 
 import { assertCompletes, assertThrows, assertErrors } from './utils'
 
-const limitSuite = window.limitSuite = getData => () => {
+const limitSuite = global.limitSuite = getData => () => {
   let data
 
   before(() => {

client/test/order.js

@@ -3,7 +3,7 @@ import { toArray } from 'rxjs/operator/toArray'
 
 import { assertCompletes, assertThrows, assertErrors } from './utils'
 
-const orderSuite = window.orderSuite = (getData, getTestData) => () => {
+const orderSuite = global.orderSuite = (getData, getTestData) => () => {
   let data, testData
 
   before(() => {

client/test/orderLimitSub.js

@@ -2,7 +2,7 @@ import { concat } from 'rxjs/operator/concat'
 
 import { assertCompletes, observableInterleave } from './utils'
 
-const orderLimitSubSuite = window.orderLimitSubSuite = getData => () => {
+const orderLimitSubSuite = global.orderLimitSubSuite = getData => () => {
   let data
 
   before(() => {

client/test/remove.js

@@ -7,7 +7,7 @@ import { ignoreElements } from 'rxjs/operator/ignoreElements'
 
 import { assertCompletes, assertThrows, removeAllData } from './utils'
 
-const removeSuite = window.removeSuite = getData => () => {
+const removeSuite = global.removeSuite = getData => () => {
   let data
   const testData = [
     { id: 1, a: 1 },

client/test/removeAll.js

@@ -7,7 +7,7 @@ import { ignoreElements } from 'rxjs/operator/ignoreElements'
 
 import { assertCompletes, assertThrows, assertErrors, removeAllData } from './utils'
 
-const removeAllSuite = window.removeAllSuite = getData => () => {
+const removeAllSuite = global.removeAllSuite = getData => () => {
   let data
   const testData = [
     { id: 1, a: 1 },

client/test/replace.js

@@ -4,7 +4,7 @@ import { toArray } from 'rxjs/operator/toArray'
 
 import { assertCompletes, assertThrows, assertErrors } from './utils'
 
-const replaceSuite = window.replaceSuite = getData => () => {
+const replaceSuite = global.replaceSuite = getData => () => {
   let data
 
   before(() => {

client/test/store.js

@@ -4,7 +4,7 @@ import { mergeMap } from 'rxjs/operator/mergeMap'
 import { toArray } from 'rxjs/operator/toArray'
 import { assertCompletes, assertThrows, assertErrors } from './utils'
 
-const storeSuite = window.storeSuite = getData => () => {
+const storeSuite = global.storeSuite = getData => () => {
   let data
 
   before(() => {