Migrate to from RxJS to xstream

package.json

@@ -20,13 +20,12 @@
     "@cycle/run": "*"
   },
   "peerDependencies": {
-    "@cycle/run": "^3.1.0",
     "electron-prebuilt": "^0.36.0",
-    "rxjs": "^5.4.0"
+    "xstream": "^10.8.0"
   },
   "devDependencies": {
     "@cycle/run": "^3.1.0",
-    "@cycle/rxjs-run": "^7.0.0",
+
     "babel-cli": "^6.5.1",
     "babel-preset-es2015": "^6.5.0",
     "babel-register": "^6.5.0",
@@ -35,9 +34,9 @@
     "electron-prebuilt": "^0.36.7",
     "istanbul": "^1.0.0-alpha.2",
     "mocha": "^2.4.5",
-    "rxjs": "^5.4.0",
     "sinon": "^1.17.3",
-    "sinon-chai": "^2.8.0"
+    "sinon-chai": "^2.8.0",
+    "xstream": "^10.8.0"
   },
   "scripts": {
     "test": "mocha --compilers js:babel-register --recursive ./test/specs",

src/AppConfigDriver.js

@@ -1,5 +1,5 @@
-import { Observable } from 'rxjs';
-import { adapt } from '@cycle/run/lib/adapt';
+import xs from 'xstream';
+import dropRepeats from 'xstream/extra/dropRepeats';
 
 import pathNames from './pathNames';
 
@@ -8,12 +8,16 @@ export default function AppPathsDriver(app) {
     const config$ = Observable.from(configXs$);
 
     const task$ = config$.map(config => config.tasks).filter(Boolean);
-    task$.forEach(tasks => app.setUserTasks(tasks));
+    task$.addListener({
+      next: tasks => app.setUserTasks(tasks)
+    });
 
     const allowNTMLForNonIntranet$ = config$
       .map(config => config.allowNTMLForNonIntranet)
       .filter(value => typeof(value) === 'boolean');
-    allowNTMLForNonIntranet$.forEach(value => app.allowNTLMCredentialsForAllDomains(value));
+    allowNTMLForNonIntranet$.addListener({
+      next: value => app.allowNTLMCredentialsForAllDomains(value)
+    });
 
     return {
       allowNTMLForNonIntranet$: allowNTMLForNonIntranet$.startWith(false),
@@ -23,19 +27,19 @@ export default function AppPathsDriver(app) {
           .map(config => config.paths && config.paths[name])
           .filter(Boolean);
 
-        pathChange$.forEach(path => app.setPath(name, path));
+        pathChange$.addListener({
+          next: path => app.setPath(name, path)
+        });
 
         return Object.defineProperty(sources, name + '$', {
           get() {
-            return adapt(Observable
-              .of(app.getPath(name))
-              .concat(pathChange$)
-              .distinct()
-            );
+            return pathChange$
+              .startWith(app.getPath(name))
+              .compose(dropRepeats());
           }
         })
       }, {
-        app$: adapt(Observable.of(app.getAppPath()))
+        app$: xs.of(app.getAppPath())
       })
     };
   }

src/AppEventsDriver.js

@@ -1,6 +1,5 @@
-import { Observable } from 'rxjs';
-import { adapt } from '@cycle/run/lib/adapt';
-
+import xs from 'xstream';
+import fromEvent from 'xstream/extra/fromEvent';
 const eventParams = {
   'open-file':                  ['path'],
   'open-url':                   ['url'],
@@ -15,11 +14,17 @@ const eventParams = {
 };
 
 export default function AppEventsDriver(app) {
-  const events = Object.keys(eventParams).map(type => Observable
-    .fromEvent(app, type, (event, ...args) => eventParams[type].reduce(
-      (event, param, i) => Object.assign(event, { [param]: args[i] }),
-      Object.assign(event, { type })))
+  const events = Object.keys(eventParams).map(type =>
+    fromEvent(app, type)
+      .map(arg => {
+        arg = Array.isArray(arg) ? arg : [arg];
+        const [event, ...paramValues] = arg;
+        return eventParams[type].reduce(
+          (ev, param, i) => Object.assign(ev, { [param]: paramValues[i] }),
+          Object.assign(event, { type })
+        )
+      })
   );
 
-  return () => adapt(Observable.merge(...events));
+  return () => xs.merge(...events);
 }

src/AppLifecycleDriver.js

@@ -1,5 +1,5 @@
-import { Observable } from 'rxjs';
-import { adapt } from '@cycle/run/lib/adapt';
+import xs from 'xstream';
+import fromEvent from 'xstream/extra/fromEvent';
 
 const eventMapping = {
   willFinishLaunching$: 'will-finish-launching',
@@ -10,35 +10,41 @@ const eventMapping = {
 };
 
 export default function AppLifecycleDriver(app) {
-  return cfgXs$ => {
-    const cfg$ = Observable.from(cfgXs$);
+  return cfg$ => {
 
     const source = Object.keys(eventMapping).reduce((obj, prop) => Object.defineProperty(obj, prop, {
       get() {
-        return adapt(Observable.fromEvent(app, eventMapping[prop]));
+        return fromEvent(app, eventMapping[prop])
       }
     }), {});
 
     Object.defineProperty(source, 'quit$', {
       get() {
-        return Observable.fromEvent(app, 'quit', (e, exitCode) => Object.assign({ exitCode }, e));
+        return fromEvent(app, 'quit')
+          .map(([ev, exitCode]) => Object.assign(ev, { exitCode }));
       }
     });
 
     let subscriptions = [];
     cfg$
       .startWith({})
-      .forEach(({ state = 'started', exitCode, isQuittingEnabled = true, isAutoExitEnabled = true } = {}) => {
-        subscriptions.filter(Boolean).forEach(s => s.dispose());
-        subscriptions = [
-          !isQuittingEnabled && source.beforeQuit$.forEach(e => e.preventDefault()),
-          !isAutoExitEnabled && source.willQuit$.forEach(e => e.preventDefault())
-        ];
+      .addListener({
+        next: ({ state = 'started', exitCode, isQuittingEnabled = true, isAutoExitEnabled = true } = {}) => {
+          subscriptions.filter(Boolean).forEach(s => s.dispose());
+          subscriptions = [
+            !isQuittingEnabled && source.beforeQuit$.addListener({
+              next: e => e.preventDefault()
+            }),
+            !isAutoExitEnabled && source.willQuit$.addListener({
+              next: e => e.preventDefault()
+            })
+          ];
 
-        if (state === 'quitting') {
-          app.quit();
-        } else if (state === 'exiting') {
-          app.exit(exitCode);
+          if (state === 'quitting') {
+            app.quit();
+          } else if (state === 'exiting') {
+            app.exit(exitCode);
+          }
         }
       });
 

src/AppMetadataDriver.js

@@ -1,8 +1,7 @@
-import { Observable } from 'rxjs';
-import { adapt } from '@cycle/run/lib/adapt';
+import xs from 'xstream';
 
 export default function AppMetadataDriver(app) {
-  return () => adapt(Observable.of({
+  return () => xs.of({
     get name() { return app.getName(); },
     get version() { return app.getVersion(); },
     get locale() { return app.getLocale(); }

src/AppVisibilityDriver.js

@@ -1,8 +1,9 @@
-import { Observable } from 'rxjs';
-import { adapt } from '@cycle/run/lib/adapt';
+import xs from 'xstream';
 
 export default function AppVisibilityDriver(app) {
   return visibility$ => {
-    adapt(Observable.from(visibility$).forEach(isVisible => isVisible ? app.show() : app.hide()));
+    visibility$.addListener({
+      next: isVisible => isVisible ? app.show() : app.hide()
+    });
   };
 }

src/BasicAuthDriver.js

@@ -1,17 +1,19 @@
-import { Observable } from 'rxjs';
-import { adapt } from '@cycle/run/lib/adapt';
+import xs from 'xstream';
+import fromEvent from 'xstream/extra/fromEvent';
 
 export default function BasicAuthDriver(app) {
   return login$ => {
-    Observable.from(login$)
-      .filter(({ event }) => event.callback)
-      .forEach(({ event, username, password }) => event.callback(username, password));
+    login$
+      .filter(({event}) => event.callback)
+      .addListener({
+        next: ({event, username, password}) => event.callback(username, password)
+      });
 
-    return adapt(Observable.fromEvent(
-      app,
-      'login',
-      (e, webContents, request, authInfo, callback) => {
-        e.preventDefault && e.preventDefault();
+    return fromEvent(app, 'login')
+      .map(arg => {
+        arg = Array.isArray(arg) ? arg : [arg];
+        const [ev, webContents, request, authInfo, callback] = arg;
+        ev.preventDefault && ev.preventDefault();
         return { webContents, request, authInfo, callback };
       }));
   };

src/CertErrorOverrideDriver.js

@@ -1,22 +1,22 @@
-import { Observable } from 'rxjs';
-import { adapt } from '@cycle/run/lib/adapt';
+import fromEvent from 'xstream/extra/fromEvent';
 
 export default function CertErrorOverrideDriver(app) {
   return (override$) => {
     Observable.from(override$)
       .filter(({ event }) => event && event.callback)
-      .forEach(({ event: { callback }, allow = false }) => callback(allow));
+      .addListener({
+        next: ({ event: { callback }, allow = false }) => callback(allow)
+      });
 
-    return adapt(Observable.fromEvent(
-      app,
-      'certificate-error',
-      (event, webContents, url, error, certificate, callback) => {
+    return fromEvent(app, 'certificate-error')
+      .map(arg => {
+        arg = Array.isArray(arg) ? arg : [arg];
+        const [event, webContents, url, error, certificate, callback] = arg;
         event.preventDefault && event.preventDefault();
         return Object.assign(
           event,
           { webContents, url, error, certificate, callback }
-        )
-      }
-    ));
-  }
+        );
+      });
+  };
 }

src/ClientCertDriver.js

@@ -1,18 +1,19 @@
-import { Observable } from 'rxjs';
-import { adapt } from '@cycle/run/lib/adapt';
+import fromEvent from 'xstream/extra/fromEvent';
 
 export default function ClientCertDriver(app) {
   return prompt$ => {
     Observable.from(prompt$)
       .filter(({ event: { callback } }) => callback)
-      .forEach(({ event: { callback }, cert }) => callback(cert));
+      .addListener({
+        next: ({ event: { callback }, cert }) => callback(cert)
+      });
 
-    return adapt(Observable.fromEvent(
-      app,
-      'select-client-certificate',
-      ({ preventDefault }, webContents, url, certificateList, callback) => {
+    return fromEvent(app, 'select-client-certificate')
+      .map(arg => {
+        arg = Array.isArray(arg) ? arg : [arg];
+        const [{ preventDefault }, webContents, url, certificateList, callback] = arg;
         preventDefault && preventDefault();
         return { webContents, url, certificateList, callback };
-      }));
-  }
+      });
+  };
 }