upgrade rxjs to v5

Author: jcouyang

lib/engine/most.js

@@ -9,7 +9,7 @@ export default function Engine() {
   travelStream.send = travelStream.next.bind(travelStream);
 
   function flatObserve(actionsSinks, f){
-    return mergeArray(actionsSinks)
+    let subscriptions = mergeArray(actionsSinks)
       .recoverWith(e => {
         console.error('There is Error in your reducer:',e, e.stack)
         return of(x=>x)
@@ -18,6 +18,8 @@ export default function Engine() {
         next:f,
         error: (e)=>console.error('Something is Wrong:',e, e.stack),
       });
+    subscriptions.isDisposed = subscriptions.disposable.disposed
+    return subscriptions;
   }
   historyStream.travel = travelStream;
   return {intentStream, flatObserve, historyStream}

lib/engine/rx.js

@@ -1,11 +1,22 @@
-import Rx from 'rx'
+import {Subject} from 'rxjs/Subject'
+import {Observable} from 'rxjs/Observable'
+import {from} from 'most'
+import 'rxjs/add/operator/map'
+import 'rxjs/add/operator/take'
+import 'rxjs/add/observable/from'
+import 'rxjs/add/operator/mergeAll'
 export default function rxEngine() {
-  Rx.Subject.prototype.send = Rx.Subject.prototype.onNext
-  let historyStream = new Rx.Subject();
-  historyStream.travel = new Rx.Subject();
-  let intentStream = new Rx.Subject();
+  Subject.prototype.send = Subject.prototype.next
+  let historyStream = new Subject();
+  historyStream.travel = new Subject();
+  let intentStream = new Subject();
   function flatObserve(actionsSinks, f){
-    return Rx.Observable.from(actionsSinks).mergeAll().subscribe(f);
+    let subscriptions =  Observable.from(actionsSinks).mergeAll().subscribe(
+      f,
+      (e)=>console.error('Something is Wrong:',e, e.stack)
+    )
+    subscriptions.unsubscribe = subscriptions.dispose
+    return subscriptions
   }
   return {intentStream,flatObserve,historyStream}
 }

lib/history.js

@@ -1,12 +1,13 @@
+import {from} from 'most'
 export default function initHistory(contextHistory){
-  let history = contextHistory
+  let history =from(contextHistory)
         .timestamp()
         .scan((acc,state)=>{
           acc.push(state)
           return acc;
         }, [])
         .multicast()
-  let travel = contextHistory.travel
+  let travel = from(contextHistory.travel)
   history.cursor = -1
   history.travel = travel
     .sample((offset,states)=>{

lib/react-most.js

@@ -1,12 +1,13 @@
 import React from 'react'
 import initHistory from './history'
+import {from} from 'most'
 import mostEngine from './engine/most'
 import mergeAll from 'ramda/src/mergeAll'
 import pick from 'ramda/src/pick'
 import keys from 'ramda/src/keys'
 // unfortunately React doesn't support symbol as context key yet, so let me just preteding using Symbol until react implement the Symbol version of Object.assign
-const INTENT_STREAM = "__reactive.react.intentStream__";
-const HISTORY_STREAM = "__reactive.react.historyStream__";
+export const INTENT_STREAM = "__reactive.react.intentStream__";
+export const HISTORY_STREAM = "__reactive.react.historyStream__";
 const EACH_FLATMAP = "__reactive.react.flatObserve__";
 
 const CONTEXT_TYPE = {
@@ -39,7 +40,7 @@ export function connect(main, initprops={}) {
           super(props, context);
           if(initprops.history || props.history){
             initprops.history = initHistory(context[HISTORY_STREAM])
-            initprops.history.travel.observe(state=>{
+            initprops.history.travel.forEach(state=>{
               return this.setState(state)
             })
           }
@@ -55,7 +56,7 @@ export function connect(main, initprops={}) {
         }
         componentDidMount(){
           this.subscriptions = this.context[EACH_FLATMAP](this.sink$, (action)=>{
-            if(this.subscriptions && this.subscriptions.disposable.disposed)
+            if(this.subscriptions && this.subscriptions.isDisposed)
               return
             else if(action instanceof Function) {
               this.setState((prevState, props)=>{
@@ -95,9 +96,9 @@ let Most = React.createClass({
     /* istanbul ignore next */
     if(process.env.NODE_ENV==='debug') {
       if(engineClass == mostEngine) {
-        engine.intentStream.timestamp()
+        from(engine.intentStream).timestamp()
           .observe(stamp=>console.log(`[${new Date(stamp.time).toJSON()}][INTENT]:}`, stamp.value));
-        engine.historyStream.timestamp()
+        from(engine.historyStream).timestamp()
           .observe(stamp=>console.log(`[${new Date(stamp.time).toJSON()}][STATE]:}`, stamp.value));
       }
     }
@@ -126,7 +127,7 @@ function actionsAndSinks(sinks, self){
       self.context[INTENT_STREAM].send(f(e));
     },
     fromPromise(p){
-      p.then(self.context[INTENT_STREAM].send);
+      p.then(x=>self.context[INTENT_STREAM].send(x));
     }
   };
   for(let name in sinks){

lib/test-utils.js

@@ -1,25 +1,25 @@
 import {from } from 'most'
-
+import {INTENT_STREAM, HISTORY_STREAM} from './react-most'
 function getStreamOf(component, name) {
-  let s = component.context[name];
+  let s = from(component.context[name]);
   s.take$ = n => s.take(n).forEach(_=>_)
   return s
 }
 
 function historyStreamOf(component) {
-  return getStreamOf(component, '__reactive.react.historyStream__')
+  return getStreamOf(component, HISTORY_STREAM)
 }
 
 function intentStreamOf(component) {
-  return getStreamOf(component, '__reactive.react.intentStream__')
+  return getStreamOf(component, INTENT_STREAM)
 }
 
 function do$(listOfActions) {
   return from(listOfActions).forEach(x=>x())
 }
 
 function dispatch(listOfIntent, component) {
-  let s = intentStreamOf(component)
+  let s = component.context[INTENT_STREAM]
   return from(listOfIntent).forEach(i=>s.send(i))
 }
 

package.json

@@ -28,7 +28,7 @@
     "most-subject": "^5.2.0",
     "ramda": "^0.22.1",
     "react": "^15.3.2",
-    "rx": "^4.1.0"
+    "rxjs": "^5.0.0-rc.4"
   },
   "jest": {
     "moduleFileExtensions": [

yarn.lock

@@ -2454,9 +2454,11 @@ rimraf@^2.4.3, rimraf@^2.4.4, rimraf@~2.5.1, rimraf@~2.5.4, rimraf@2:
   dependencies:
     glob "^7.0.5"
 
-rx@^4.1.0:
-  version "4.1.0"
-  resolved "https://registry.yarnpkg.com/rx/-/rx-4.1.0.tgz#a5f13ff79ef3b740fe30aa803fb09f98805d4782"
+rxjs:
+  version "5.0.0-rc.4"
+  resolved "https://registry.yarnpkg.com/rxjs/-/rxjs-5.0.0-rc.4.tgz#a4d08bc5d7f30d48ed7130e2995490c326a325c4"
+  dependencies:
+    symbol-observable "^1.0.1"
 
 sane@~1.4.1:
   version "1.4.1"
@@ -2604,7 +2606,7 @@ supports-color@^3.1.0, supports-color@^3.1.2:
   dependencies:
     has-flag "^1.0.0"
 
-symbol-observable@^1.0.2:
+symbol-observable@^1.0.1, symbol-observable@^1.0.2:
   version "1.0.4"
   resolved "https://registry.yarnpkg.com/symbol-observable/-/symbol-observable-1.0.4.tgz#29bf615d4aa7121bdd898b22d4b3f9bc4e2aa03d"