# Path: oslash/observable.py
# class Observable(Generic[TSource]):
# 
#     """The Rx Observable Monad.
# 
#     The Rx Observable monad is based on the Continuation monad
#     representing suspended computations in continuation-passing style
#     (CPS).
#     """
# 
#     def __init__(self, subscribe: Callable[[Callable], Any]) -> None:
#         """Observable constructor.
# 
#         Keyword arguments:
#         subscribe -- A callable that takes a callable (on_next)
#         """
#         self._get_value = lambda: subscribe
# 
#     @classmethod
#     def unit(cls, x: TSource) -> 'Observable[TSource]':
#         """x -> Observable x"""
#         return cls(lambda on_next: on_next(x))
#     just = unit
# 
#     def map(self, mapper: Callable[[TSource], TResult]) -> 'Observable[TResult]':
#         r"""Map a function over an observable.
# 
#         Haskell: fmap f m = Cont $ \c -> runCont m (c . f)
#         """
#         source = self
#         return Observable(lambda on_next: source.subscribe(compose(on_next, mapper)))
# 
#     def bind(self, fn: Callable[[TSource], 'Observable[TResult]']) -> 'Observable[TResult]':
#         r"""Chain continuation passing functions.
# 
#         Haskell: m >>= k = Cont $ \c -> runCont m $ \a -> runCont (k a) c
#         """
#         source = self
#         return Observable(lambda on_next: source.subscribe(lambda a: fn(a).subscribe(on_next)))
#     flat_map = bind
# 
#     def filter(self, predicate: Callable[[TSource], bool]) -> 'Observable[TSource]':
#         """Filter the on_next continuation functions"""
#         source = self
# 
#         def subscribe(on_next):
#             def _next(x):
#                 if predicate(x):
#                     on_next(x)
# 
#             return source.subscribe(_next)
#         return Observable(subscribe)
# 
#     @staticmethod
#     def call_cc(fn: Callable) -> 'Observable':
#         r"""call-with-current-continuation.
# 
#         Haskell: callCC f = Cont $ \c -> runCont (f (\a -> Cont $ \_ -> c a )) c
#         """
#         def subscribe(on_next):
#             return fn(lambda a: Observable(lambda _: on_next(a))).subscribe(on_next)
# 
#         return Observable(subscribe)
# 
#     def subscribe(self, on_next: Callable[[TSource], None]) -> Any:
#         return self._get_value()(on_next)
# 
#     def __or__(self, func):
#         """Use | as operator for bind.
# 
#         Provide the | operator instead of the Haskell >>= operator
#         """
#         return self.bind(func)
# 
#     def __eq__(self, other) -> bool:
#         return self.subscribe(identity) == other.subscribe(identity)
# 
# Path: oslash/util/fn.py
# A = TypeVar("A")
# B = TypeVar("B")
# C = TypeVar("C")
# D = TypeVar("D")
# E = TypeVar("E")
# F = TypeVar("F")
# G = TypeVar("G")
# def compose() -> Callable[[A], A]:  # pylint: disable=function-redefined
# def compose(op1: Callable[[A], B]) -> Callable[[A], B]:  # pylint: disable=function-redefined
# def compose(op2: Callable[[B], C], op1: Callable[[A], B]) -> Callable[[A], C]:  # pylint: disable=function-redefined
# def compose(
#     op3: Callable[[C], D], op2: Callable[[B], C], op1: Callable[[A], B]  # pylint: disable=function-redefined
# ) -> Callable[[A], D]:
# def compose(
#     op4: Callable[[D], E],  # pylint: disable=function-redefined
#     op3: Callable[[C], D],
#     op2: Callable[[B], C],
#     op1: Callable[[A], B],
# ) -> Callable[[A], E]:
# def compose(
#     op5: Callable[[E], F],  # pylint: disable=function-redefined
#     op4: Callable[[D], E],
#     op3: Callable[[C], D],
#     op2: Callable[[B], C],
#     op1: Callable[[A], B],
# ) -> Callable[[A], F]:
# def compose(
#     op1: Callable[[A], B],  # pylint: disable=function-redefined,too-many-arguments
#     op2: Callable[[B], C],
#     op3: Callable[[C], D],
#     op4: Callable[[D], E],
#     op5: Callable[[E], F],
#     op6: Callable[[F], G],
# ) -> Callable[[A], G]:
# def compose(*funcs: Callable) -> Callable:  # type: ignore
#     def _compose(source: Any) -> Any:


import unittest
from oslash.observable import Observable
from oslash.util import identity, compose


# pure = Cont.pure
unit = Observable.unit
just = Observable.just
call_cc = Observable.call_cc


class TestObservable(unittest.TestCase):
    def test_observable_just(self):
        stream = Observable.just(42)

        def on_next(x):
            return "OnNext(%s)" % x

        self.assertEqual("OnNext(42)", stream.subscribe(on_next))

    def test_observable_just_map(self):
        stream = Observable.just(42).map(lambda x: x*10)

        def on_next(x):
            return "OnNext(%s)" % x

        self.assertEqual("OnNext(420)", stream.subscribe(on_next))

    def test_observable_just_flatmap(self):

# Path: oslash/observable.py
# class Observable(Generic[TSource]):
# 
#     """The Rx Observable Monad.
# 
#     The Rx Observable monad is based on the Continuation monad
#     representing suspended computations in continuation-passing style
#     (CPS).
#     """
# 
#     def __init__(self, subscribe: Callable[[Callable], Any]) -> None:
#         """Observable constructor.
# 
#         Keyword arguments:
#         subscribe -- A callable that takes a callable (on_next)
#         """
#         self._get_value = lambda: subscribe
# 
#     @classmethod
#     def unit(cls, x: TSource) -> 'Observable[TSource]':
#         """x -> Observable x"""
#         return cls(lambda on_next: on_next(x))
#     just = unit
# 
#     def map(self, mapper: Callable[[TSource], TResult]) -> 'Observable[TResult]':
#         r"""Map a function over an observable.
# 
#         Haskell: fmap f m = Cont $ \c -> runCont m (c . f)
#         """
#         source = self
#         return Observable(lambda on_next: source.subscribe(compose(on_next, mapper)))
# 
#     def bind(self, fn: Callable[[TSource], 'Observable[TResult]']) -> 'Observable[TResult]':
#         r"""Chain continuation passing functions.
# 
#         Haskell: m >>= k = Cont $ \c -> runCont m $ \a -> runCont (k a) c
#         """
#         source = self
#         return Observable(lambda on_next: source.subscribe(lambda a: fn(a).subscribe(on_next)))
#     flat_map = bind
# 
#     def filter(self, predicate: Callable[[TSource], bool]) -> 'Observable[TSource]':
#         """Filter the on_next continuation functions"""
#         source = self
# 
#         def subscribe(on_next):
#             def _next(x):
#                 if predicate(x):
#                     on_next(x)
# 
#             return source.subscribe(_next)
#         return Observable(subscribe)
# 
#     @staticmethod
#     def call_cc(fn: Callable) -> 'Observable':
#         r"""call-with-current-continuation.
# 
#         Haskell: callCC f = Cont $ \c -> runCont (f (\a -> Cont $ \_ -> c a )) c
#         """
#         def subscribe(on_next):
#             return fn(lambda a: Observable(lambda _: on_next(a))).subscribe(on_next)
# 
#         return Observable(subscribe)
# 
#     def subscribe(self, on_next: Callable[[TSource], None]) -> Any:
#         return self._get_value()(on_next)
# 
#     def __or__(self, func):
#         """Use | as operator for bind.
# 
#         Provide the | operator instead of the Haskell >>= operator
#         """
#         return self.bind(func)
# 
#     def __eq__(self, other) -> bool:
#         return self.subscribe(identity) == other.subscribe(identity)
# 
# Path: oslash/util/fn.py
# A = TypeVar("A")
# B = TypeVar("B")
# C = TypeVar("C")
# D = TypeVar("D")
# E = TypeVar("E")
# F = TypeVar("F")
# G = TypeVar("G")
# def compose() -> Callable[[A], A]:  # pylint: disable=function-redefined
# def compose(op1: Callable[[A], B]) -> Callable[[A], B]:  # pylint: disable=function-redefined
# def compose(op2: Callable[[B], C], op1: Callable[[A], B]) -> Callable[[A], C]:  # pylint: disable=function-redefined
# def compose(
#     op3: Callable[[C], D], op2: Callable[[B], C], op1: Callable[[A], B]  # pylint: disable=function-redefined
# ) -> Callable[[A], D]:
# def compose(
#     op4: Callable[[D], E],  # pylint: disable=function-redefined
#     op3: Callable[[C], D],
#     op2: Callable[[B], C],
#     op1: Callable[[A], B],
# ) -> Callable[[A], E]:
# def compose(
#     op5: Callable[[E], F],  # pylint: disable=function-redefined
#     op4: Callable[[D], E],
#     op3: Callable[[C], D],
#     op2: Callable[[B], C],
#     op1: Callable[[A], B],
# ) -> Callable[[A], F]:
# def compose(
#     op1: Callable[[A], B],  # pylint: disable=function-redefined,too-many-arguments
#     op2: Callable[[B], C],
#     op3: Callable[[C], D],
#     op4: Callable[[D], E],
#     op5: Callable[[E], F],
#     op6: Callable[[F], G],
# ) -> Callable[[A], G]:
# def compose(*funcs: Callable) -> Callable:  # type: ignore
#     def _compose(source: Any) -> Any:


# Path: tests/test_observable.py
import unittest
from oslash.observable import Observable
from oslash.util import identity, compose



# pure = Cont.pure
unit = Observable.unit
just = Observable.just
call_cc = Observable.call_cc


class TestObservable(unittest.TestCase):
    def test_observable_just(self):
        stream = Observable.just(42)

        def on_next(x):
            return "OnNext(%s)" % x

        self.assertEqual("OnNext(42)", stream.subscribe(on_next))

    def test_observable_just_map(self):
        stream = Observable.just(42).map(lambda x: x*10)

        def on_next(x):
            return "OnNext(%s)" % x

        self.assertEqual("OnNext(420)", stream.subscribe(on_next))

    def test_observable_just_flatmap(self):

        stream = Observable.just(42).flat_map(lambda x: just(x*10))