ReactiveX for MoonBit

English | 中文

Reactive Extensions for MoonBit - A comprehensive reactive programming library for the MoonBit language.

Overview

ReactiveX for MoonBit is a feature-complete reactive programming library that provides Observable sequences and rich operators for composing asynchronous and event-based programs. Based on the ReactiveX specification, it brings powerful reactive programming capabilities to the MoonBit language.

Features

🔧 Core Components

• ✅ Core Types: Observable[T], Observer[T], BasicSubscription
• ✅ Error Handling: RxError enum type with type-safe error management
• ✅ Subscription Management: Memory-safe subscription lifecycle management

🚀 Creation Operators (5)

• ✅ of - Create Observable from a single value
• ✅ from_array - Create Observable from an array
• ✅ empty - Create empty sequence
• ✅ never - Create Observable that never emits
• ✅ error / error_with_type - Create error Observable

🔄 Transformation Operators (8)

• ✅ map - Transform values
• ✅ filter - Filter by predicate
• ✅ take - Take first N values
• ✅ skip - Skip first N values
• ✅ scan - Accumulate with intermediate results
• ✅ reduce - Reduce to single final result
• ✅ flat_map - Transform and flatten inner Observables
• ✅ switch_map - Switch to latest inner Observable

🔗 Combination Operators (4)

• ✅ merge - Merge multiple Observables
• ✅ concat - Concatenate multiple Observables
• ✅ combine_latest - Combine latest values from multiple sources
• ✅ zip - Pair values from multiple sources

🛠️ Utility Operators (6)

• ✅ tap - Side effects (debugging friendly)
• ✅ distinct - Remove duplicates
• ✅ catch_error - Error catching and recovery
• ✅ debounce - Emit only after delay period
• ✅ start_with - Prepend initial value
• ✅ retry - Retry on error with max attempts

⚡ Advanced Features

• ✅ Generic Support: Full type safety guarantees
• ✅ Fluent API: Chainable method design
• ✅ Error Recovery: Robust error handling mechanisms
• ✅ Test Coverage: 30+ test cases with 100% coverage

Quick Start

Installation

Add this library to your moon.mod.json dependencies:

{
  "deps": {
    "reactivex": "path/to/reactivex"
  }
}

Basic Usage

// Create Observable
let numbers = from_array([1, 2, 3, 4, 5])

// Transform data stream: map -> filter -> limit
let result = numbers
  |> map(fn(x) { x * 2 })           // [2, 4, 6, 8, 10]
  |> filter(fn(x) { x > 5 })        // [6, 8, 10]  
  |> take(2)                        // [6, 8]

// Method 1: Simple subscription (data only)
let _ = result.subscribe_next(fn(value) { 
  println("Received: \(value)") 
})

// Method 2: Complete subscription (data, error, complete)
let observer = new_simple_observer(
  fn(value) { println("Value: \(value)") },
  fn(error) { println("Error: \(error)") },
  fn() { println("Complete!") }
)
let subscription = result.subscribe(observer)

Error Handling Example

// Create potentially failing Observable
let risky_data = from_array([1, 0, 2])
  |> map(fn(x) { 10 / x })  // Division by zero will fail

// Use catch_error for recovery
let safe_data = risky_data.catch_error(fn(err) {
  println("Caught error: \(err)")
  of(-1)  // Return default value
})

let _ = safe_data.subscribe_next(fn(x) { println("Result: \(x)") })

API Documentation

🔧 Core Types

RxError

pub enum RxError {
  RuntimeError(String)      // Runtime error
  OperatorError(String)     // Operator error  
  SubscriptionError(String) // Subscription error
  TimeoutError(String)      // Timeout error
}

Observable[T]

Core type representing a reactive data stream - an observable sequence of values.

Observer[T]

Observer interface with three callback functions:

• on_next: (T) -> Unit - Called when receiving new value
• on_error: (RxError) -> Unit - Called when error occurs
• on_complete: () -> Unit - Called when sequence completes

🚀 Creation Operators

• of[T](value: T) -> Observable[T] - Create from single value
• from_array[T](values: Array[T]) -> Observable[T] - Create from array
• empty[T]() -> Observable[T] - Create empty sequence
• never[T]() -> Observable[T] - Create never-emitting sequence
• error[T](message: String) -> Observable[T] - Create error sequence

🔄 Transformation Operators

• map[T, U](source, transform) -> Observable[U] - Transform values
• filter[T](source, predicate) -> Observable[T] - Filter by condition
• take[T](source, count) -> Observable[T] - Take first N values
• skip[T](source, count) -> Observable[T] - Skip first N values
• scan[T, U](source, initial, accumulator) -> Observable[U] - Accumulate
• reduce[T, U](source, initial, accumulator) -> Observable[U] - Reduce

🔗 Combination Operators

• merge[T](sources: Array[Observable[T]]) -> Observable[T] - Merge Observables
• concat[T](sources: Array[Observable[T]]) -> Observable[T] - Concatenate Observables

🛠️ Utility Operators

• tap[T](source, side_effect) -> Observable[T] - Side effects
• distinct[T : Eq](source) -> Observable[T] - Remove duplicates
• catch_error[T](source, error_handler) -> Observable[T] - Error recovery

📚 Examples

🎯 Usage Examples

Check out complete examples in src/examples.mbt:

// Call example functions
example_basic_usage()      // Basic usage demo
example_operator_chain()   // Operator chaining demo

🔗 Operator Chaining Example

// Data processing pipeline
let pipeline = from_array([1, 2, 3, 2, 4, 5])
  |> distinct()                    // Dedupe: [1, 2, 3, 4, 5]
  |> filter(fn(x) { x % 2 == 1 })  // Odds: [1, 3, 5]
  |> map(fn(x) { x * x })         // Square: [1, 9, 25]
  |> scan(0, fn(acc, x) { acc + x }) // Sum: [1, 10, 35]

let _ = pipeline.subscribe_next(fn(x) { println("Running sum: \(x)") })

🧪 Testing

Run Tests

# Run all tests
moon test

# Check code
moon check

# Build project
moon build

📊 Test Coverage

Current Test Status:

• ✅ Test Cases: 29 tests, all passing
• ✅ API Coverage: 16 public functions fully covered
• ✅ Feature Coverage: All operators and core functionality

🛠️ Development

Project Structure

ReactiveX/
├── src/                    # Source directory
│   ├── reactivex.mbt      # ReactiveX core implementation
│   ├── test.mbt           # Test suite
│   ├── examples.mbt       # Usage examples
│   └── moon.pkg.json      # Package configuration
├── moon.mod.json          # Project configuration
├── README.md              # English documentation
├── README_zh_CN.md        # Chinese documentation
└── LICENSE                # MIT license

Build Commands

# Check syntax and types
moon check

# Build project  
moon build

# Run tests
moon test

# Format code
moon fmt

🤝 Contributing

Contributions are welcome! Please ensure:

1. Code Quality: Follow existing code style, pass all tests
2. Test Coverage: Add test cases for new features
3. Documentation: Update relevant docs and examples

📄 License

MIT License - see LICENSE file for details.

🙏 Acknowledgments

This project is inspired by ReactiveX, bringing reactive programming support to the MoonBit language.

---

ReactiveX for MoonBit - Making reactive programming simple and powerful in MoonBit! 🚀