feat(RxJS 6): updated RxJS version 6 (#84)

- Library now uses RxJS directly rather than installing it somehow
- Removes a lot of unnecessary null checks
- Updates to use pipeable operators
- Users using RxJS 5 will need to install rxjs-compat
- Updates tests to use RxJS v6
- Adds notes about possible future changes regarding RxJS
- Fixes a test relying on synchronous error throwing, that behavior was
  removed in RxJS 6 due to bugs it could cause.

BREAKING CHANGE: RxJS 4 no longer supported
BREAKING CHANGE: Rx global no longer required
BREAKING CHANGE: RxJS 5 users will need to install rxjs-compat

Resolves #83

Author: benlesh

package-lock.json

@@ -43,7 +43,7 @@
     "rollup": "^0.50.0",
     "rollup-plugin-buble": "^0.16.0",
     "rollup-watch": "^4.3.1",
-    "rxjs": "^5.2.0",
+    "rxjs": "^6.2.0",
     "typescript": "^2.5.2",
     "vue": "^2.5.0"
   },

package.json

@@ -1,24 +1,21 @@
-import { Rx, hasRx, isSubject, warn, getKey, unsub } from '../util'
+import { isObserver, warn, getKey } from '../util'
+import { fromEvent } from 'rxjs'
 
 export default {
   // Example ./example/counter_dir.html
   bind (el, binding, vnode) {
-    if (!hasRx()) {
-      return
-    }
-
     let handle = binding.value
     const event = binding.arg
     const streamName = binding.expression
     const modifiers = binding.modifiers
 
-    if (isSubject(handle)) {
+    if (isObserver(handle)) {
       handle = { subject: handle }
-    } else if (!handle || !isSubject(handle.subject)) {
+    } else if (!handle || !isObserver(handle.subject)) {
       warn(
         'Invalid Subject found in directive with key "' + streamName + '".' +
-        streamName + ' should be an instance of Rx.Subject or have the ' +
-        'type { subject: Rx.Subject, data: any }.',
+        streamName + ' should be an instance of Subject or have the ' +
+        'type { subject: Subject, data: any }.',
         vnode.context
       )
       return
@@ -45,17 +42,8 @@ export default {
         })
       })
     } else {
-      if (!Rx.Observable.fromEvent) {
-        warn(
-          `No 'fromEvent' method on Observable class. ` +
-          `v-stream directive requires Rx.Observable.fromEvent method. ` +
-          `Try import 'rxjs/add/observable/fromEvent' for ${streamName}`,
-          vnode.context
-        )
-        return
-      }
       const fromEventArgs = handle.options ? [el, event, handle.options] : [el, event]
-      handle.subscription = Rx.Observable.fromEvent(...fromEventArgs).subscribe(e => {
+      handle.subscription = fromEvent(...fromEventArgs).subscribe(e => {
         modifiersExists.forEach(mod => modifiersFuncs[mod](e))
         next({
           event: e,
@@ -72,7 +60,7 @@ export default {
   update (el, binding) {
     const handle = binding.value
     const _handle = el._rxHandles && el._rxHandles[getKey(binding)]
-    if (_handle && handle && isSubject(handle.subject)) {
+    if (_handle && handle && isObserver(handle.subject)) {
       _handle.data = handle.data
     }
   },
@@ -81,7 +69,9 @@ export default {
     const key = getKey(binding)
     const handle = el._rxHandles && el._rxHandles[key]
     if (handle) {
-      unsub(handle.subscription)
+      if (handle.subscription) {
+        handle.subscription.unsubscribe()
+      }
       el._rxHandles[key] = null
     }
   }

src/directives/stream.js

@@ -1,4 +1,6 @@
-import { Rx, hasRx, warn } from '../util'
+import { share } from 'rxjs/operators'
+import { Observable } from 'rxjs'
+import { warn } from '../util'
 
 /**
  * @name Vue.prototype.$createObservableMethod
@@ -8,21 +10,8 @@ import { Rx, hasRx, warn } from '../util'
  * @return {Observable} Hot stream
  */
 export default function createObservableMethod (methodName, passContext) {
-  if (!hasRx()) {
-    return
-  }
   const vm = this
 
-  if (!Rx.Observable.prototype.share) {
-    warn(
-      `No 'share' operator. ` +
-      `$createObservableMethod returns a shared hot observable. ` +
-      `Try import 'rxjs/add/operator/share' for creating ${methodName}`,
-      vm
-    )
-    return
-  }
-
   if (vm[methodName] !== undefined) {
     warn(
       'Potential bug: ' +
@@ -52,5 +41,5 @@ export default function createObservableMethod (methodName, passContext) {
   }
 
   // Must be a hot stream otherwise function context may overwrite over and over again
-  return Rx.Observable.create(creator).share()
+  return new Observable(creator).pipe(share())
 }

src/methods/createObservableMethod.js

@@ -1,17 +1,14 @@
-import { Rx, hasRx } from '../util'
+import { Observable } from 'rxjs'
 
 /**
  * @see {@link https://vuejs.org/v2/api/#vm-on}
  * @param {String||Array} evtName Event name
  * @return {Observable} Event stream
  */
 export default function eventToObservable (evtName) {
-  if (!hasRx()) {
-    return
-  }
   const vm = this
   const evtNames = Array.isArray(evtName) ? evtName : [evtName]
-  const obs$ = Rx.Observable.create(observer => {
+  const obs$ = new Observable(observer => {
     const eventPairs = evtNames.map(name => {
       const callback = msg => observer.next({ name, msg })
       vm.$on(name, callback)

src/methods/eventToObservable.js

@@ -1,16 +1,15 @@
-import { Rx, hasRx, getDisposable } from '../util'
+import { Observable, Subscription, NEVER } from 'rxjs'
 
 export default function fromDOMEvent (selector, event) {
-  if (!hasRx()) {
-    return
-  }
   if (typeof window === 'undefined') {
-    return Rx.Observable.create(() => {})
+    // TODO(benlesh): I'm not sure if this is really what you want here,
+    // but it's equivalent to what you were doing. You might want EMPTY
+    return NEVER
   }
 
   const vm = this
   const doc = document.documentElement
-  const obs$ = Rx.Observable.create(observer => {
+  const obs$ = new Observable(observer => {
     function listener (e) {
       if (!vm.$el) return
       if (selector === null && vm.$el === e.target) return observer.next(e)
@@ -22,7 +21,7 @@ export default function fromDOMEvent (selector, event) {
     }
     doc.addEventListener(event, listener)
     // Returns function which disconnects the $watch expression
-    return getDisposable(() => {
+    return new Subscription(() => {
       doc.removeEventListener(event, listener)
     })
   })

src/methods/fromDOMEvent.js

@@ -1,5 +1,7 @@
+import { Subscription } from 'rxjs'
+
 export default function subscribeTo (observable, next, error, complete) {
-  var obs$ = observable.subscribe(next, error, complete)
-  ;(this._obSubscriptions || (this._obSubscriptions = [])).push(obs$)
-  return obs$
+  const subscription = observable.subscribe(next, error, complete)
+  ;(this._subscription || (this._subscription = new Subscription())).add(subscription)
+  return subscription
 }

src/methods/subscribeTo.js

@@ -1,12 +1,8 @@
-import { Rx, hasRx, getDisposable } from '../util'
+import { Observable, Subscription } from 'rxjs'
 
 export default function watchAsObservable (expOrFn, options) {
-  if (!hasRx()) {
-    return
-  }
-
   const vm = this
-  const obs$ = Rx.Observable.create(observer => {
+  const obs$ = new Observable(observer => {
     let _unwatch
     const watch = () => {
       _unwatch = vm.$watch(expOrFn, (newValue, oldValue) => {
@@ -24,7 +20,7 @@ export default function watchAsObservable (expOrFn, options) {
     }
 
     // Returns function which disconnects the $watch expression
-    return getDisposable(() => {
+    return new Subscription(() => {
       _unwatch && _unwatch()
     })
   })

src/methods/watchAsObservable.js

@@ -1,17 +1,14 @@
-import { Rx, defineReactive, isObservable, warn, unsub } from './util'
+import { defineReactive, isObservable, warn } from './util'
+import { Subject, Subscription } from 'rxjs'
 
 export default {
   created () {
     const vm = this
     const domStreams = vm.$options.domStreams
     if (domStreams) {
-      if (!Rx.Subject) {
-        warn('Rx.Subject is required to use the "domStreams" option.')
-      } else {
-        domStreams.forEach(key => {
-          vm[key] = new Rx.Subject()
-        })
-      }
+      domStreams.forEach(key => {
+        vm[key] = new Subject()
+      })
     }
 
     const observableMethods = vm.$options.observableMethods
@@ -33,7 +30,7 @@ export default {
     }
     if (obs) {
       vm.$observables = {}
-      vm._obSubscriptions = []
+      vm._subscription = new Subscription()
       Object.keys(obs).forEach(key => {
         defineReactive(vm, key, undefined)
         const ob = vm.$observables[key] = obs[key]
@@ -44,16 +41,16 @@ export default {
           )
           return
         }
-        vm._obSubscriptions.push(obs[key].subscribe(value => {
+        vm._subscription.add(obs[key].subscribe(value => {
           vm[key] = value
         }, (error) => { throw error }))
       })
     }
   },
 
   beforeDestroy () {
-    if (this._obSubscriptions) {
-      this._obSubscriptions.forEach(unsub)
+    if (this._subscription) {
+      this._subscription.unsubscribe()
     }
   }
 }

src/mixin.js

@@ -1,53 +1,26 @@
-export let Rx
 export let Vue
 export let warn = function () {}
 
-export function install (_Vue, _Rx) {
-  Rx = _Rx
+// NOTE(benlesh): the value of this method seems dubious now, but I'm not sure
+// if this is a Vue convention I'm just not familiar with. Perhaps it would
+// be better to just import and use Vue directly?
+export function install (_Vue) {
   Vue = _Vue
   warn = Vue.util.warn || warn
 }
 
-export function hasRx (vm) {
-  if (!Rx) {
-    warn(
-      '$watchAsObservable requires Rx to be present globally or ' +
-      'be passed to Vue.use() as the second argument.',
-      vm
-    )
-    return false
-  }
-  return true
-}
-
+// TODO(benlesh): as time passes, this should be updated to use RxJS 6.1's
+// `isObservable` method. But wait until you're ready to drop support for Rx 5
 export function isObservable (ob) {
   return ob && typeof ob.subscribe === 'function'
 }
 
-export function isSubject (subject) {
+export function isObserver (subject) {
   return subject && (
-    typeof subject.next === 'function' ||
-    typeof subject.onNext === 'function'
+    typeof subject.next === 'function'
   )
 }
 
-export function unsub (handle) {
-  if (!handle) return
-  if (handle.dispose) {
-    handle.dispose()
-  } else if (handle.unsubscribe) {
-    handle.unsubscribe()
-  }
-}
-
-export function getDisposable (target) {
-  if (Rx.Subscription) { // Rx5
-    return new Rx.Subscription(target)
-  } else { // Rx4
-    return Rx.Disposable.create(target)
-  }
-}
-
 export function defineReactive (vm, key, val) {
   if (key in vm) {
     vm[key] = val