🥮 mooncakes.io

lunar-reactor

Asynchronous reactive streams library for Moonbit

async
reactive
streams
moon add quelgar/lunar-reactor@0.1.0
Download zip
Author
Version
0.1.0
License
Apache-2.0
Repository
Last updated
9 hours ago
Downloads
2

Dependencies

README

#Lunar Reactor

Asynchronous reactive streams for Moonbit.

The main thing that Lunar Reactor offers is the Observable type, which provides a push-based asynchronous stream of values with backpressure.

Inspired by the Monix and ZIO Scala libraries.

See examples/fs_examples.mbt for a simple example of using an observable to read and write files with UTF-8 encoding.

#Observables

The type Observable[T, E] represents a stream of values of type T that can fail with an error of type E. Observables can be modified in various ways, such as applying filters or mapping values. Observables are lazy and only start emitting values when they are subscribed to by an Observer.

[!NOTE] Currently Lunar Reactor only offers "cold" observables, which means that the stream of values is generated anew for every subscription by an observer.

#Observers

An observer is used to subscribe to an observable, which activates the observable. The observable will make calls to the three methods of the observer: on_next, on_error and on_complete. All are async.

[!NOTE] While the Observer trait is the fundamental way to use an Observable, it is typically only used within the internal implementations of Observables. For application code, Observable is not very convenient to use directly, and it is generally preferable to consume an observable with a Consumer, see below.

The order of calls is guaranteed to be: 0 or more calls to on_next, followed by either a single call to on_error or a single call to on_complete. No more calls will be made to the observer after either on_error or on_complete has been called. The only exception to this is if the observer returns Stop from on_next, see below.

#on_next

The on_next method accepts a value of type T and returns an Ack. It is called whenever the observable has a value to emit. The Ack return value has two cases: Continue and Stop. If Stop is returned then the subscription is effectively cancelled and the observer will not be called on any of its methods again.

As on_next is async, the observable will wait for the returned Ack before emitting the next value. This allows the observer to apply backpressure to the observable.

#on_error

Provides an error of type E to the observer, indicating the observable failed. After this method is called, the observable will not emit any more values and the observer will not be called again.

#on_complete

Indicates to the observer that the observable has completed successfully. After this method is called, the observable will not emit any more values and the observer will not be called again.

#Consumers

A Consumer[T, E, R] is a type that can consumer from an Observable[T, E] and produces either a result of type R when the observable completes successfully, or an error of type E when the observable or consumer itself fails. A consumer is typically a much more convenient way for application code to use observables than using an observer directly.

Passing a consumer to the consume method of an observable will subscribe to the observable and start the stream. As consume is async, it will block until the observable completes or fails, and will either return the result of type R or raise an error of type E.

There's also the consume_result method that returns a Result[R, E] instead of raising.

async test "consuming an observable" {
  let stream = Observable::from_array(["a", "b", "c"])
  let result = stream.consume(Consumer::count())
  assert_eq(result, 3)
}

#Pipelines

Pipelines are essentially functions from one observable to another. They aren't strictly necessary, as anything they can do could be done directly to an observable, but they provide a convenient way to encapsulate and compose useful transformations.

Pipelines use attached to observers via the pipe method, and can be composed together using and_then.

async test "using a pipeline" {
  let pipeline = Pipeline::map(i => i * 2)
    .and_then(Pipeline::drop_while(i => i < 7))
  let stream = Observable::range_between(start=0, end=10, step=1)
    .pipe(pipeline)
  let result = stream.consume(Consumer::collect_all())
  assert_eq(result, [8, 10, 12, 14, 16, 18])
}

#Cancellation

All the fundamental methods of the system, subscribe, on_next, on_error, on_complete, and consume are all async. This means that they can be cancelled. The only error that these methods are allowed to raise is Cancelled, which is how subscriptions are aborted in response to the task running the subscription being cancelled.

When implementing these methods, whenever you call an async function, you may need to check for cancellation. For functions that can only raise due to cancellation, you can use the capture_cancellation utility function to convert such a raise to a raise of Cancelled.

#To Do

  • Javascript interop

#
Cancelled

source
pub suberror Cancelled {
  Cancelled
}

#
NoError

source
pub suberror NoError

An error type with no constructors, used to indicate an Observable that never raises an error.

#
Ack

source
pub(all) enum Ack {
  Continue
  Stop
} derive(
Debug)

#
CollectToFixedResult

source
pub struct CollectToFixedResult[T] {
  view : ArrayView[T]
  overflowed : Bool
}

#
Consumer

source
type Consumer[T, E, R]

#
Consumer::Consumer

source
fn[T, E, R] Consumer::Consumer(observer : Observer[T, E], result : async () -> Result[R, E] raise Cancelled) -> Consumer[T, E, R]

#
Consumer::all

source
fn[T, E] Consumer::all(predicate : (T) -> Bool) -> Consumer[T, E, Bool]

#
Consumer::any

source
fn[T, E] Consumer::any(predicate : (T) -> Bool) -> Consumer[T, E, Bool]

#
Consumer::collect_all

source
fn[T, E] Consumer::collect_all(size_hint? : Int) -> Consumer[T, E, Array[T]]

#
Consumer::collect_to_fixed

source
fn[T, E] Consumer::collect_to_fixed(fixed : FixedArray[T]) -> Consumer[T, E, CollectToFixedResult[T]]

#
Consumer::contains

source
fn[T : Eq, E] Consumer::contains(item : T) -> Consumer[T, E, Bool]

#
Consumer::contramap

source
fn[T, E, R, U] Consumer::contramap(self : Consumer[T, E, R], f : (U) -> T) -> Consumer[U, E, R]

#
Consumer::count

source
fn[T, E] Consumer::count() -> Consumer[T, E, UInt64]

#
Consumer::drain

source
fn[T, E] Consumer::drain() -> Consumer[T, E, Unit]

#
Consumer::each

source
fn[T, E : Error] Consumer::each(effect : async (T) -> Unit raise E) -> Consumer[T, E, Unit]

#
Consumer::find

source
fn[T, E] Consumer::find(predicate : (T) -> Bool) -> Consumer[T, E, T?]

#
Consumer::from_observer

source
fn[T, E] Consumer::from_observer(observer : Observer[T, E]) -> Consumer[T, E, Unit]

#
Consumer::hash

source
fn[T : Hash, E] Consumer::hash(seed? : Int) -> Consumer[T, E, Int]

#
Consumer::make

source
fn[T, E, R] Consumer::make(on_next~ : async (T) -> Ack raise Cancelled, provide_result~ : async () -> Result[R, E] raise Cancelled) -> Consumer[T, E, R]

#
Consumer::make_nofail

source
fn[T, E, R] Consumer::make_nofail(on_next~ : async (T) -> Ack raise Cancelled, provide_result~ : async () -> R raise Cancelled) -> Consumer[T, E, R]

#
Consumer::map_result

source
fn[T, E, R, R2] Consumer::map_result(self : Consumer[T, E, R], f : (R) -> R2) -> Consumer[T, E, R2]

#
Consumer::observer

source
fn[T, E, R] Consumer::observer(self : Consumer[T, E, R]) -> Observer[T, E]

#
Consumer::result

source
async fn[T, E, R] Consumer::result(self : Consumer[T, E, R]) -> Result[R, E] raise Cancelled

#
Consumer::stop

source
fn[T, E] Consumer::stop() -> Consumer[T, E, Unit]

#
Consumer::zip

source
fn[T, E, R1, R2] Consumer::zip(self : Consumer[T, E, R1], other : Consumer[T, E, R2]) -> Consumer[T, E, (R1, R2)]

#
Observable

source
type Observable[T, E]

#
Observable::Observable

source
fn[T, E] Observable::Observable(subscribe : async (Observer[T, E]) -> Unit raise Cancelled) -> Observable[T, E]

#
Observable::buffer_tumbling

source
fn[T, E] Observable::buffer_tumbling(self : Observable[T, E], size : Int) -> Observable[
Vector[T], E]

#
Observable::catch_error

source
fn[T, E] Observable::catch_error(self : Observable[T, E], f : (E) -> Observable[T, E]) -> Observable[T, E]

#
Observable::consume

source
async fn[T, E : Error, R] Observable::consume(self : Observable[T, E], consumer : Consumer[T, E, R]) -> R

#
Observable::consume_result

source
async fn[T, E, R] Observable::consume_result(self : Observable[T, E], consumer : Consumer[T, E, R]) -> Result[R, E] raise Cancelled

#
Observable::debug_dump

source

#
Observable::drop

source
fn[T, E] Observable::drop(self : Observable[T, E], count : UInt) -> Observable[T, E]

#
Observable::empty

source
fn[T, E] Observable::empty() -> Observable[T, E]

#
Observable::fail

source
fn[T, E] Observable::fail(error : E) -> Observable[T, E]

#
Observable::filter

source
fn[T, E] Observable::filter(self : Observable[T, E], predicate : (T) -> Bool) -> Observable[T, E]

#
Observable::flat_map

source
fn[T, E, U] Observable::flat_map(self : Observable[T, E], f : (T) -> Observable[U, E]) -> Observable[U, E]

Concatenates the inner Observables produced by f in order into a single flat Observable.

#
Observable::from_array

source
fn[T] Observable::from_array(array : ArrayView[T]) -> Observable[T, NoError]

#
Observable::from_async

source
fn[T, E : Error] Observable::from_async(task : async () -> T raise E) -> Observable[T, E]

#
Observable::from_async_generator

source
fn[T, E : Error] Observable::from_async_generator(generator : async () -> T? raise E) -> Observable[T, E]

#
Observable::from_bytes

source
fn Observable::from_bytes(bytes : BytesView) -> Observable[Byte, NoError]

#
Observable::from_generator

source
fn[T, E : Error] Observable::from_generator(generator : () -> T? raise E) -> Observable[T, E]

#
Observable::from_iter

source
fn[T] Observable::from_iter(iter : Iter[T]) -> Observable[T, NoError]

#
Observable::from_value

source
fn[T, E] Observable::from_value(value : T) -> Observable[T, E]

#
Observable::map

source
fn[T, E, U] Observable::map(self : Observable[T, E], f : (T) -> U) -> Observable[U, E]

#
Observable::map_error

source
fn[T, E1, E2] Observable::map_error(self : Observable[T, E1], f : (E1) -> E2) -> Observable[T, E2]

#
Observable::never

source
fn[T, E] Observable::never() -> Observable[T, E]

#
Observable::panic_on_error

source
fn[T, E] Observable::panic_on_error(self : Observable[T, E]) -> Observable[T, NoError]

#
Observable::pipe

source
fn[T1, T2, E] Observable::pipe(self : Observable[T1, E], pipeline : Pipeline[T1, T2, E]) -> Observable[T2, E]

#
Observable::range_between

source
fn[T : Compare + Add + Eq] Observable::range_between(start~ : T, end~ : T, step~ : T) -> Observable[T, NoError]

#
Observable::range_from

source
fn[T : Add] Observable::range_from(start~ : T, step~ : T) -> Observable[T, NoError]

#
Observable::recover

source
fn[T, E] Observable::recover(self : Observable[T, E], f : (E) -> T) -> Observable[T, NoError]

#
Observable::refine_error

source
fn[T, E1, E2] Observable::refine_error(self : Observable[T, E1], f : (E1) -> E2?) -> Observable[T, E2]

#
Observable::subscribe

source
async fn[T, E] Observable::subscribe(self : Observable[T, E], observer : Observer[T, E]) -> Unit raise Cancelled

Subscribes the given observer to this observable, blocking until the observable completes, errors, or is cancelled.

#
Observable::take

source
fn[T, E] Observable::take(self : Observable[T, E], count : UInt) -> Observable[T, E]

#
Observable::tap

source
fn[T, E : Error] Observable::tap(self : Observable[T, E], effect : async (T) -> Unit raise E) -> Observable[T, E]

#
Observable::tap_error

source
fn[T, E : Error] Observable::tap_error(self : Observable[T, E], effect : async (E) -> Unit raise E) -> Observable[T, E]

#
Observable::widen_error

source
fn[T, E] Observable::widen_error(self : Observable[T, NoError]) -> Observable[T, E]

#
Observer

source
type Observer[T, E]

#
Observer::Observer

source
fn[T, E] Observer::Observer(on_next~ : async (T) -> Ack raise Cancelled, on_complete~ : async () -> Unit raise Cancelled, on_error~ : async (E) -> Unit raise Cancelled) -> Observer[T, E]

#
Observer::before_on_error

source
fn[T, E : Error] Observer::before_on_error(self : Observer[T, E], effect : async (E) -> Unit raise E) -> Observer[T, E]

#
Observer::before_on_next

source
fn[T, E : Error] Observer::before_on_next(self : Observer[T, E], effect : async (T) -> Unit raise E) -> Observer[T, E]

#
Observer::contramap

source
fn[T, E, U] Observer::contramap(self : Observer[T, E], f : (U) -> T) -> Observer[U, E]

#
Observer::contramap_halt

source
fn[T, E, U] Observer::contramap_halt(self : Observer[T, E], f : (U) -> T?) -> Observer[U, E]

#
Observer::debug_dump

source

#
Observer::on_complete

source
async fn[T, E] Observer::on_complete(self : Observer[T, E]) -> Unit raise Cancelled

#
Observer::on_error

source
async fn[T, E] Observer::on_error(self : Observer[T, E], e : E) -> Unit raise Cancelled

#
Observer::on_next

source
async fn[T, E] Observer::on_next(self : Observer[T, E], t : T) -> Ack raise Cancelled

#
Observer::replace_on_next

source
fn[T, E, U] Observer::replace_on_next(self : Observer[T, E], on_next : async (U) -> Ack raise Cancelled) -> Observer[U, E]

#
Pipeline

source
type Pipeline[T1, T2, E]

#
Pipeline::and_then

source
fn[T1, T2, T3, E] Pipeline::and_then(self : Pipeline[T1, T2, E], next : Pipeline[T2, T3, E]) -> Pipeline[T1, T3, E]

#
Pipeline::apply

source
fn[T1, T2, E] Pipeline::apply(self : Pipeline[T1, T2, E], observable : Observable[T1, E]) -> Observable[T2, E]

#
Pipeline::bytes_to_view

source
fn[E] Pipeline::bytes_to_view() -> Pipeline[Bytes, BytesView, E]

#
Pipeline::chunk_bytes

source
fn[E] Pipeline::chunk_bytes(chunk_size : Int) -> Pipeline[Byte, Bytes, E]

#
Pipeline::drop_while

source
fn[T, E] Pipeline::drop_while(predicate : (T) -> Bool) -> Pipeline[T, T, E]

#
Pipeline::flat_map

source
fn[T1, T2, E] Pipeline::flat_map(f : (T1) -> Observable[T2, E]) -> Pipeline[T1, T2, E]

#
Pipeline::make

source
fn[T1, T2, E] Pipeline::make(f : (Observable[T1, E]) -> Observable[T2, E]) -> Pipeline[T1, T2, E]

#
Pipeline::map

source
fn[T1, T2, E] Pipeline::map(f : (T1) -> T2) -> Pipeline[T1, T2, E]

#
Pipeline::split_lines

source
fn[E] Pipeline::split_lines(max_line_length? : Int) -> Pipeline[Char, String, E]

#
Pipeline::split_on

source
fn[T : Eq, E] Pipeline::split_on(separator : T) -> Pipeline[T, 
Vector[T], E]

#
Pipeline::string_to_chars

source
fn[E] Pipeline::string_to_chars() -> Pipeline[String, Char, E]

#
Pipeline::unchunk_bytes

source
fn[E] Pipeline::unchunk_bytes() -> Pipeline[Bytes, Byte, E]

#
Resource

source
type Resource[T]

#
Resource::Resource

source
fn[T] Resource::Resource(acquire : async () -> T, release : (T) -> Unit) -> Resource[T]

#
Resource::acquire

source
async fn[T] Resource::acquire(self : Resource[T]) -> T

#
Resource::release

source
fn[T] Resource::release(self : Resource[T], resource : T) -> Unit

#
Resource::use_resource

source
async fn[T, U] Resource::use_resource(self : Resource[T], f : async (T) -> U) -> U

#
capture_cancellation

source
async fn[T] capture_cancellation(effect : async () -> T) -> T raise Cancelled

Source Files

Powered by MoonBit

Site sourceReport issuePackagesBuild queueSkillsStatistics

© 2026 mooncakes.io