Add feature eventToObservable

README.md

@@ -183,6 +183,30 @@ vm.$watchAsObservable('a')
 
 The optional `options` object accepts the same options as `vm.$watch`.
 
+#### `$eventToObservable(event)`
+
+> This feature requires RxJS.
+
+Convert vue.$on (including lifecycle events) to Observables. The emitted value is in the format of `{ name, msg }`:
+
+``` js
+var vm = new Vue({
+  created () {
+    this.$eventToObservable('customEvent')
+	  .subscribe((event) => console.log(event.name,event.msg))	
+  }
+})
+
+// vm.$once vue-rx version
+this.$eventToObservable('customEvent')
+  .take(1)
+
+// Another way to auto unsub:
+let beforeDestroy$ = this.$eventToObservable('hook:beforeDestroy').take(1)
+Rx.Observable.interval(500)
+  .takeUntil(beforeDestroy$)
+```
+
 #### `$subscribeTo(observable, next, error, complete)`
 
 This is a prototype method added to instances. You can use it to subscribe to an observable, but let VueRx manage the dispose/unsubscribe.

example/counter-simple.html

@@ -6,20 +6,31 @@
   <div>{{ count }}</div>
   <button v-stream:click="plus$">+</button>
   <button v-stream:click="minus$">-</button>
+  <my-component v-stream:more="more$"></my-component>
 </div>
 
 <script>
+Vue.component('my-component', {
+  template: '<button v-on:click="more">Iwant more!</button>',
+  methods:{
+    more:function(){
+      this.$emit('more',100)
+    }
+  }
+})
+
 new Vue({
   el: '#app',
 
   // declare dom stream Subjects
-  domStreams: ['plus$', 'minus$'],
+  domStreams: ['plus$', 'minus$','more$'],
 
   subscriptions () {
     var plus$ = this.plus$.map(() => 1)
     var minus$ = this.minus$.map(() => -1)
+    var more$ = this.more$.map(() => 100)
     var count$ = Rx.Observable
-      .merge(plus$, minus$)
+      .merge(plus$, minus$, more$)
       .startWith(0)
       .scan((total, change) => total + change)
 
@@ -28,4 +39,4 @@
     }
   }
 })
-</script>
+</script>

src/directives/stream.js

@@ -11,15 +11,6 @@ export default {
     const event = binding.arg
     const streamName = binding.expression
 
-    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
-    }
     if (isSubject(handle)) {
       handle = { subject: handle }
     } else if (!handle || !isSubject(handle.subject)) {
@@ -34,17 +25,36 @@ export default {
 
     const subject = handle.subject
     const next = (subject.next || subject.onNext).bind(subject)
-    let fromEventArgs = handle.options ? [el, event, handle.options] : [el, event]
-    handle.subscription = Rx.Observable.fromEvent(...fromEventArgs).subscribe(e => {
-      next({
-        event: e,
-        data: handle.data
+
+    if (vnode.componentInstance) {
+      handle.subscription = vnode.componentInstance.$eventToObservable(event).subscribe(e => {
+        next({
+          event: e,
+          data: handle.data
+        })
+      })
+    } 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
+      }
+      let fromEventArgs = handle.options ? [el, event, handle.options] : [el, event]
+      handle.subscription = Rx.Observable.fromEvent(...fromEventArgs).subscribe(e => {
+        next({
+          event: e,
+          data: handle.data
+        })
       })
-    })
 
-    // store handle on element with a unique key for identifying
-    // multiple v-stream directives on the same node
-    ;(el._rxHandles || (el._rxHandles = {}))[getKey(binding)] = handle
+      // store handle on element with a unique key for identifying
+      // multiple v-stream directives on the same node
+      ;(el._rxHandles || (el._rxHandles = {}))[getKey(binding)] = handle
+    }
   },
 
   update (el, binding) {

src/index.js

@@ -4,6 +4,7 @@ import streamDirective from './directives/stream'
 import watchAsObservable from './methods/watchAsObservable'
 import fromDOMEvent from './methods/fromDOMEvent'
 import subscribeTo from './methods/subscribeTo'
+import eventToObservable from './methods/eventToObservable'
 
 export default function VueRx (Vue, Rx) {
   install(Vue, Rx)
@@ -12,6 +13,7 @@ export default function VueRx (Vue, Rx) {
   Vue.prototype.$watchAsObservable = watchAsObservable
   Vue.prototype.$fromDOMEvent = fromDOMEvent
   Vue.prototype.$subscribeTo = subscribeTo
+  Vue.prototype.$eventToObservable = eventToObservable
 }
 
 // auto install

src/methods/eventToObservable.js

@@ -0,0 +1,27 @@
+import { Rx, hasRx } from '../util'
+
+/**
+ * @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
+  let evtNames = Array.isArray(evtName) ? evtName : [evtName]
+  const obs$ = Rx.Observable.create(observer => {
+    let eventPairs = evtNames.map(name => {
+      let callback = msg => observer.next({name, msg})
+      vm.$on(name, callback)
+      return {name, callback}
+    })
+    return () => {
+      // Only remove the specific callback
+      eventPairs.forEach(pair => vm.$off(pair.name, pair.callback))
+    }
+  })
+
+  return obs$
+}

src/methods/fromDOMEvent.js

@@ -27,6 +27,5 @@ export default function fromDOMEvent (selector, event) {
     })
   })
 
-  ;(vm._obSubscriptions || (vm._obSubscriptions = [])).push(obs$)
   return obs$
 }

src/methods/watchAsObservable.js

@@ -29,6 +29,5 @@ export default function watchAsObservable (expOrFn, options) {
     })
   })
 
-  ;(vm._obSubscriptions || (vm._obSubscriptions = [])).push(obs$)
   return obs$
 }

test/test.js

@@ -284,3 +284,42 @@ test('$subscribeTo()', () => {
   next(2)
   expect(results).toEqual([1]) // should not trigger anymore
 })
+
+
+test('$eventToObservable()', done => {
+  let calls = 0;
+  const vm = new Vue({
+    created(){
+      let ob = this.$eventToObservable('ping')
+        .subscribe(function (event) {
+          expect(event.name).toEqual('ping');
+          expect(event.msg).toEqual('ping message');
+          calls++
+        });
+    }
+  });
+  vm.$emit('ping','ping message');
+
+  nextTick(()=>{
+    vm.$destroy();
+    //Should not emit
+    vm.$emit('pong','pong message');
+    expect(calls).toEqual(1);
+    done()
+  });
+});
+
+
+test('$eventToObservable() with lifecycle hooks', done => {
+  const vm = new Vue({
+    created(){
+      this.$eventToObservable('hook:beforeDestroy')
+        .subscribe(function (event) {
+          done(event)
+        });
+    }
+  });
+  nextTick(()=>{
+    vm.$destroy()
+  })
+});