Purescript + rxjs playground

RxPlayground.purs

module RxPlayground where

import Debug.Trace (trace, print)
import Control.Monad.Eff

import Data.Array ((..))

foreign import data Observable :: * -> *
foreign import data Subject :: * -> *
foreign import data Observer :: * -> *
foreign import data Disposable :: *

foreign import dispose
  """
  function dispose(disposable) {
    return function() {
      disposable.dispose();
      return {};
    };
  }
  """ :: forall eff. Disposable -> Eff eff Unit

foreign import createObservable
  """
  function createObservable(f) {
    return Rx.Observable.create(function(observer) {
      return f(observer)();
    });
  }
  """ :: forall v rx eff. (Observer v -> Eff eff (Eff eff Unit)) -> Observable v

foreign import fromArray
  """
  function fromArray(xs) {
    return Rx.Observable.fromArray(xs);
  }
  """ :: forall v. [v] -> Observable v

foreign import newSubject
  """
  function newSubject() {
    return new Rx.Subject();
  }
  """ :: forall v eff. Eff eff (Subject v)

foreign import toObservable
  """
  function toObservable(subj) { return subj; }
  """ :: forall v. Subject v -> Observable v

foreign import subscribe
  """
  function subscribe(obs, nextFn, errFn, cmplFn) {
    return function(nextFn) {
      return function(errFn) {
        return function(cmplFn) {
          return function() {
            return obs.subscribe(function(v) { nextFn(v)(); },
                                 function(err) { errFn(err)(); },
                                 cmplFn);
          };
        };
      };
    };
  }
  """ :: forall err v eff. Observable v
      -> (v -> Eff eff Unit)
      -> ({ | err } -> Eff eff Unit)
      -> Eff eff Unit
      -> Eff eff Disposable


foreign import genericOnNext
  """
  function genericOnNext(o) {
    return function(v) {
      return function() {
        o.onNext(v);
        return {};
      };
    };
  }
  """ :: forall val ob eff. ob val -> val -> Eff eff Unit

foreign import genericOnErr
  """
  function genericOnErr(o) {
    return function(v) {
      return function() {
        o.onError(v);
        return {};
      };
    };
  }
  """ :: forall err val ob eff. ob val -> err -> Eff eff Unit

foreign import genericOnCmpl
  """
  function genericOnCmpl(o) {
    return function() {
      o.onCompleted();
      return {};
    };
  }
  """ :: forall val ob eff. ob val -> Eff eff Unit

foreign import scan
  """
  function scan(stepFn) {
    return function(zero) {
      return function(obs) {
        return obs.scan(stepFn, zero);
      };
    };
  }
  """ :: forall val st. (st -> val -> st) -> st -> Observable val -> Observable st

foreign import map
  """
  function map(mapFn) {
    return function(obs) {
      return obs.map(mapFn);
    };
  }
  """ :: forall a b. (a -> b) -> Observable a -> Observable b

foreign import filter
  """
  function filter(predFn) {
    return function(obs) {
      return obs.filter(predFn);
    };
  }
  """ :: forall a. (a -> Boolean) -> Observable a -> Observable a

class IObserver ob where
  onNext  :: forall val eff. ob val -> val -> Eff eff Unit
  onError :: forall val err eff. ob val -> err -> Eff eff Unit
  onCompleted :: forall val eff. ob val -> Eff eff Unit

instance subjectIObserver :: IObserver Subject where
  onNext = genericOnNext
  onError = genericOnErr
  onCompleted = genericOnCmpl

instance observerIObserver :: IObserver Observer where
  onNext = genericOnNext
  onError = genericOnErr
  onCompleted = genericOnCmpl

main2 = do
  subject <- newSubject
  disp <- subscribe (toObservable subject)
                    (\v -> trace $ "Hello World: " ++ show v)
                    (\err -> trace "ERROR!")
                    (trace "done")
  onNext subject 123
  onCompleted subject
  return {}

main = do
  let observable =
        scan (+) 0 $ fromArray (1..10)
        -- filter (\x -> x > 5) $ fromArray [1..10]
        -- createObservable $ \observer -> do
        --   trace "HELLO THERE!"
        --   onNext observer 10
        --   onCompleted observer
        --   return $ trace "Observable was disposed"
  disp <-
    subscribe observable (\v -> trace $ "Hello World " ++ show v)
                         (\err -> trace "error")
                         (trace "done")

  -- dispose disp
  return {}