Support Cycle Unified

src/index.ts

@@ -1,6 +1,7 @@
 import {DevToolEnabledSource} from '@cycle/run';
 import xs, {Stream, MemoryStream} from 'xstream';
 import dropRepeats from 'xstream/extra/dropRepeats';
+import { adapt } from '@cycle/run/lib/adapt';
 
 export type MainFn<So, Si> = (sources: So) => Si;
 export type Reducer<T> = (state: T | undefined) => T | undefined;
@@ -16,21 +17,23 @@ export type Scope<T, R> = string | number | Lens<T, R>;
 
 export function pick(selector: Selector | string) {
   if (typeof selector === 'string') {
-    return function pickWithString(sinksArray$: Stream<Array<any>>): Stream<Array<any>> {
-      return sinksArray$.map(sinksArray => sinksArray.map(sinks => sinks[selector]));
+    return function pickWithString(sinksArray$: any): Stream<Array<any>> {
+      return adapt((xs.fromObservable(sinksArray$) as Stream<Array<any>>)
+        .map(sinksArray => sinksArray.map(sinks => sinks[selector])));
     };
   } else {
-    return function pickWithFunction(sinksArray$: Stream<Array<any>>): Stream<Array<any>> {
-      return sinksArray$.map(sinksArray => sinksArray.map(selector));
+    return function pickWithFunction(sinksArray$: any): Stream<Array<any>> {
+      return adapt((xs.fromObservable(sinksArray$) as Stream<Array<any>>)
+        .map(sinksArray => sinksArray.map(selector)));
     };
   }
 }
 
 export function mix(aggregator: Aggregator) {
-  return function mixOperator(streamArray$: Stream<Array<Stream<any>>>): Stream<any> {
-    return streamArray$
+  return function mixOperator(streamArray$: any): Stream<Array<Stream<any>>> {
+    return adapt((xs.fromObservable(streamArray$) as Stream<Array<Stream<any>>>)
       .map(streamArray => aggregator(...streamArray))
-      .flatten();
+      .flatten());
   }
 }
 
@@ -105,7 +108,7 @@ export class StateSource<T> {
 
   constructor(stream: Stream<any>, name: string | null) {
     this._name = name;
-    this.state$ = stream.compose(dropRepeats()).remember();
+    this.state$ = adapt(stream.compose(dropRepeats()).remember());
     if (!name) {
       return;
     }
@@ -134,7 +137,10 @@ export default function onionify<So, Si>(
       .drop(1);
     sources[name] = new StateSource<any>(state$, name) as any;
     const sinks = main(sources as So);
-    reducerMimic$.imitate(sinks[name]);
+    if (sinks[name]) {
+      const stream$ = xs.fromObservable<Reducer<any>>(sinks[name]);
+      reducerMimic$.imitate(stream$);
+    }
     return sinks;
   }
 }