adjust v-stream syntax

Author: yyx990803

example/counter_dir.html

@@ -0,0 +1,43 @@
+<script src="https://unpkg.com/@reactivex/rxjs/dist/global/Rx.js"></script>
+<script src="https://unpkg.com/vue/dist/vue.js"></script>
+<script src="../vue-rx.js"></script>
+
+<div id="app">
+  <div>{{ count }}</div>
+  <button v-stream:click="plus$">Add on Click</button>
+  <button
+    v-stream:click="minus$"
+    v-stream:mousemove="minus$">
+    Minus on Click &amp; Mousemove
+  </button>
+</div>
+
+<script>
+new Vue({
+  el: '#app',
+  data:function () {
+    return {
+      delta1: 1,
+      delta2: -1
+    }
+  },
+  created:function () {
+    //Speed up after 10s
+    setTimeout(() => {
+      this.delta2 = -5
+    }, 10 * 1000)
+  },
+  domStreams: ['plus$', 'minus$'],
+  subscriptions:function () {
+    var plus = this.plus$.map(o => this.delta1)
+    var minus = this.minus$.map(o => this.delta2)
+    var count$ = Rx.Observable.merge(plus, minus)
+      .startWith(0)
+      .scan((total, change) => total + change);
+
+    return {
+      count: count$
+    }
+  }
+})
+</script>

example/couter-with-directive.html

@@ -53,27 +53,35 @@
     Vue.mixin({
       created: function init () {
         var vm = this
+        var domStreams = vm.$option.domStreams
+        if (domStreams) {
+          domStreams.forEach(function (key) {
+            vm[key] = new Rx.Subject()
+          })
+        }
+
         var obs = vm.$options.subscriptions
         if (typeof obs === 'function') {
           obs = obs.call(vm)
         }
-        if (!obs) return
-        vm.$observables = {}
-        vm._obSubscriptions = []
-        Object.keys(obs).forEach(function (key) {
-          defineReactive(vm, key, undefined)
-          var ob = vm.$observables[key] = obs[key]
-          if (!isObservable(ob)) {
-            warn(
-              'Invalid Observable found in subscriptions option with key "' + key + '".',
-              vm
-            )
-            return
-          }
-          vm._obSubscriptions.push(obs[key].subscribe(function (value) {
-            vm[key] = value
-          }))
-        })
+        if (obs) {
+          vm.$observables = {}
+          vm._obSubscriptions = []
+          Object.keys(obs).forEach(function (key) {
+            defineReactive(vm, key, undefined)
+            var ob = vm.$observables[key] = obs[key]
+            if (!isObservable(ob)) {
+              warn(
+                'Invalid Observable found in subscriptions option with key "' + key + '".',
+                vm
+              )
+              return
+            }
+            vm._obSubscriptions.push(obs[key].subscribe(function (value) {
+              vm[key] = value
+            }))
+          })
+        }
       },
       beforeDestroy: function () {
         if (this._obSubscriptions) {
@@ -157,37 +165,26 @@
         if (!hasRx()) {
           return
         }
-        var streamName = binding.arg
-        var vmStream = vnode.context[streamName] || vnode.context.$observables[streamName]
-        var eventNames = Object.keys(binding.modifiers)
-
-        if (isSubject(vmStream)) {
-          var onNext = (vmStream.next || vmStream.onNext).bind(vmStream) // Rx4 Rx5
-          el.vStreamData = binding.value
-          el._obs$ = eventNames.map(function (evtName) {
-            return Rx.Observable.fromEvent(el, evtName)
-              .subscribe(function (evt) {
-                onNext({
-                  event: evt,
-                  data: el.vStreamData // Not using binding.value for data updating reason
-                })
-              })
+        var event = binding.arg
+        var stream = binding.value
+        var streamName = binding.expression
+
+        if (isSubject(stream)) {
+          var onNext = (stream.next || stream.onNext).bind(stream) // Rx4 Rx5
+          el._ob$ = Rx.Observable.fromEvent(el, event).subscribe(function (evt) {
+            onNext({ event: evt })
           })
         } else {
           warn(
             'Invalid Subject found in directive with key "' + streamName + '".' +
-            'Please declare ' + streamName + ' as an new Rx.Subject'
+            streamName + ' should be an instance of Rx.Subject',
+            vnode.context
           )
         }
       },
-      update: function (el, binding, vnode) {
-        el.vStreamData = binding.value
-      },
       unbind: function (el, binding, vnode) {
-        if (Array.isArray(el._obs$)) {
-          el._obs$.forEach(function (ob) {
-            unsub(ob)
-          })
+        if (el._ob$) {
+          unsub(el._ob$)
         }
       }
     })