add more generic method from replacing fromPromise

Author: jcouyang

src/__tests__/fantasyx-test.ts

@@ -35,7 +35,7 @@ describe('FantasyX', () => {
     intent.next(2)
 
     let res = FlatMap.Xstream.flatMap(
-      (x: number) => Xstream.fromPromise<"ArrayStream", number, number>(new Id({ count: x + 1 }))
+      (x: number) => Xstream.from(new Id({ count: x + 1 }))
       , Xstream.fromIntent<"ArrayStream", number>())
       .toFantasyX()
       .toStream(intent)
@@ -59,8 +59,8 @@ describe('FantasyX', () => {
   it('concat promise', () => {
 
     let res = Semigroup.Xstream.concat(
-      Xstream.fromPromise(new Id({ count1: 1 }))
-      , Xstream.fromPromise(new Id({ count2: 2 })))
+      Xstream.from(new Id({ count1: 1 }))
+      , Xstream.from(new Id({ count2: 2 })))
       .toFantasyX()
       .toStream(intent)
       .reduce((acc, f: any) => f(acc), { count: 0 })

src/fantasy/xstream.ts

@@ -1,4 +1,4 @@
-import { Functor, map } from './typeclasses/functor'
+import { Functor, FunctorInstances, map } from './typeclasses/functor'
 import { Cartesian, product } from './typeclasses/cartesian'
 import { Apply } from './typeclasses/apply'
 import { FlatMap, flatMap } from './typeclasses/flatmap'
@@ -55,6 +55,13 @@ export class Xstream<S extends Stream, I, A> {
     })))
   }
 
+  static from<F extends Stream, I, A, G extends FunctorInstances>(p: $<G, A>) {
+    return new Xstream<F, I, A>(new State((intent$: Subject<F, I>) => ({
+      s: intent$,
+      a: streamOps.from(p) as $<F, A>
+    })))
+  }
+
   toFantasyX<St>() {
     type itentStream = Subject<S, I>
     type updateStream = $<S, State<St, A>>

src/xs/array.ts

@@ -70,3 +70,7 @@ StreamOps.prototype.fromPromise = function(p) {
   }
   return [p.valueOf()]
 }
+
+StreamOps.prototype.from = function(fa) {
+  return [fa.valueOf()]
+}

src/xs/index.ts

@@ -1,4 +1,5 @@
 import { _, $ } from '../fantasy/typeclasses'
+import { FunctorInstances } from '../fantasy/typeclasses/functor'
 export interface S_<A> { }
 export type Stream = keyof S_<any>
 
@@ -17,6 +18,7 @@ export class StreamOps<F extends Stream> { }
 export interface StreamOps<F extends Stream> {
   empty<A>(): $<F, A>
   fromPromise<A>(p: Promise<A>): $<F, A>
+  from<G extends FunctorInstances, A>(fa: $<G, A>): $<F, A>
   just<A>(a: A): $<F, A>
   merge<A, B>(
     a: $<F, A>,

src/xs/most.ts

@@ -1,4 +1,4 @@
-import { Stream as MostStream, empty, just, combineArray, combine, flatMap, fromPromise } from 'most'
+import { Stream as MostStream, empty, just, combineArray, combine, flatMap, fromPromise, from } from 'most'
 import { sync, SyncSubject, Subject } from 'most-subject'
 import { Subscription, StreamOps } from '.'
 
@@ -51,7 +51,10 @@ StreamOps.prototype.subscribe = function <A>(fa: MostStream<A>, next: (v: A) =>
   }).subscribe({ next, error: x => console.error(x), complete }) as Subscription
 }
 
-StreamOps.prototype.fromPromise = fromPromise
+StreamOps.prototype.fromPromise = fromPromise;
+
+
+(<any>StreamOps.prototype.from) = from
 
 export const URI = 'Stream'
 export type URI = typeof URI

src/xs/rx.ts

@@ -8,6 +8,7 @@ import '@reactivex/rxjs/dist/cjs/add/operator/scan'
 import '@reactivex/rxjs/dist/cjs/add/operator/catch'
 import '@reactivex/rxjs/dist/cjs/add/operator/filter'
 import '@reactivex/rxjs/dist/cjs/add/observable/empty'
+import '@reactivex/rxjs/dist/cjs/add/observable/from'
 import '@reactivex/rxjs/dist/cjs/add/observable/of'
 import '@reactivex/rxjs/dist/cjs/add/observable/fromPromise'
 import '@reactivex/rxjs/dist/cjs/add/observable/combineLatest'
@@ -61,4 +62,6 @@ StreamOps.prototype.merge = function <A, B>(a: RxStream<A>, b: RxStream<B>): RxS
   return a.merge<A, B>(b)
 }
 
-StreamOps.prototype.fromPromise = RxStream.fromPromise
+StreamOps.prototype.fromPromise = RxStream.fromPromise;
+
+(<any>StreamOps.prototype.from) = RxStream.from