Fix lazy currentUser (#586)

* currentUser now will force a handshake

Fixes #567

* Add handshake to utensils

client/src/ast.js

@@ -6,6 +6,7 @@ import 'rxjs/add/operator/scan'
 import 'rxjs/add/operator/filter'
 import 'rxjs/add/operator/map'
 import 'rxjs/add/operator/toArray'
+import 'rxjs/add/operator/take'
 
 import snakeCase from 'snake-case'
 
@@ -355,3 +356,39 @@ export class Limit extends TermBase {
     super(sendRequest, query, [])
   }
 }
+
+
+export class UserDataTerm {
+  constructor(hz, handshake, socket) {
+    this._hz = hz
+    this._before = Observable.merge(
+      socket.take(0), // just need to force connection
+      handshake // guarantee we get handshake even if we're already
+                // connected
+    )
+  }
+
+  _query(userId) {
+    return this._hz('users').find(userId)
+  }
+
+  fetch() {
+    return this._before.concatMap(handshake => {
+      if (handshake.id === null) {
+        return Observable.of({})
+      } else {
+        return this._query(handshake.id).fetch()
+      }
+    })
+  }
+
+  watch(...args) {
+    return this._before.concatMap(handshake => {
+      if (handshake.id === null) {
+        return Observable.of({})
+      } else {
+        return this._query(handshake.id).watch(...args)
+      }
+    })
+  }
+}

client/src/index.js

@@ -9,7 +9,7 @@ import 'rxjs/add/operator/filter'
 // Extra operators not used, but useful to Horizon end-users
 import 'rxjs/add/operator/defaultIfEmpty'
 
-const { Collection } = require('./ast.js')
+const { Collection, UserDataTerm } = require('./ast.js')
 const HorizonSocket = require('./socket.js')
 const { log, logError, enableLogging } = require('./logging.js')
 const { authEndpoint, TokenStorage, clearAuthTokens } = require('./auth')
@@ -58,7 +58,8 @@ function Horizon({
     return new Collection(sendRequest, name, lazyWrites)
   }
 
-  horizon.currentUser = () => new UserDataTerm(horizon, socket.handshake)
+  horizon.currentUser = () =>
+    new UserDataTerm(horizon, socket.handshake, socket)
 
   horizon.disconnect = () => {
     socket.complete()
@@ -96,6 +97,7 @@ function Horizon({
   horizon.utensils = {
     sendRequest,
     tokenStorage,
+    handshake: socket.handshake,
   }
   Object.freeze(horizon.utensils)
 
@@ -138,38 +140,6 @@ function subscribeOrObservable(observable) {
   }
 }
 
-class UserDataTerm {
-  constructor(hz, baseObservable) {
-    this._hz = hz
-    this._baseObservable = baseObservable.map(handshake => handshake.id)
-  }
-
-  _query(userId) {
-    return this._hz('users').find(userId)
-  }
-
-  fetch() {
-    return this._baseObservable.concatMap(userId => {
-      if (userId === null) {
-        return Observable.of({})
-      } else {
-        return this._query(userId).fetch()
-      }
-    })
-  }
-
-  watch(...args) {
-    return this._baseObservable.concatMap(userId => {
-      if (userId === null) {
-        return Observable.of({})
-      } else {
-        return this._query(userId).watch(...args)
-      }
-    })
-  }
-}
-
-
 Horizon.log = log
 Horizon.logError = logError
 Horizon.enableLogging = enableLogging

client/test/auth.js

@@ -8,17 +8,21 @@ const authSuite = global.authSuite = (getHorizon) => () => {
     horizon = getHorizon()
   })
   it('gets an empty object when unauthenticated', done => {
-    horizon.currentUser().fetch().subscribe(
-      user => {
+    horizon.currentUser().fetch().subscribe({
+      next(user) {
         assert.isObject(user)
         assert.deepEqual([], Object.keys(user))
       },
-      err => done(err),
-      complete => done()
-    )
+      error(err) { done(err) },
+      complete() { done() },
+    })
   })
   it('gets a normal user object when anonymous', done => {
-    const myHorizon = Horizon({ secure: false, lazyWrites: true, authType: 'anonymous' })
+    const myHorizon = Horizon({
+      secure: false,
+      lazyWrites: true,
+      authType: 'anonymous'
+    })
     Horizon.clearAuthTokens()
     myHorizon.connect()
     myHorizon.currentUser().fetch().subscribe({