Ox

Ox is a library designed for writing simple, expression-oriented code in a functional style with minimal syntax overhead. It focuses on high-performance, developer-friendly concurrency and streaming.

Key areas covered by Ox include:

• Streaming: Push-based backpressured streaming designed for direct-style usage. It includes a rich set of transformations, flexible source/sink definitions, and integration with Reactive Streams.
• Error Management: Tools for retries, timeouts, safe error propagation, and safe resource management.
• Concurrency: High-level operators, structured concurrency, safe low-level primitives, and communication mechanisms between concurrent computations.
• Scheduling & Timers: Support for managing execution timing.
• Resiliency: Implementation of patterns like circuit breakers, bulkheads, rate limiters, and backpressure.

A Flow[T] is a lazy, asynchronous data transformation pipeline that emits elements of type T.

Key characteristics:

• Lazy Evaluation: No elements are emitted and no effects are executed until the flow is explicitly run.
• Lifecycle: Flows can be finite or infinite. If infinite, running them will not end normally unless interrupted.
• Error Handling: Exceptions occurring during evaluation are thrown when the flow is run, after any cleanup logic has completed.

Structured concurrency is a programming approach where the lifetime of a thread is determined by the syntactic structure of the code. Instead of spawning threads that can outlive the function that created them (leading to 'leaked' threads), structured concurrency uses scopes.

Key principles include:

• Scopes: Concurrently running threads are started within a defined scope.
• Lifecycle Guarantee: A scope cannot finish until all threads started within it have finished (either successfully or due to an error).
• No Leaks: It is impossible to leak threads outside of a method, making threads an implementation detail rather than a side effect.
• Local Reasoning: It enables safer direct-style programming by allowing developers to reason about threading effects locally, similar to functional programming tenets.

Ox aims to combine the safety and composability of functional programming with the ease of use of imperative programming.

Key characteristics of Ox's approach:

• Concurrency Model: Uses Java 21 virtual threads combined with Go-like channels for inter-thread communication.
• Functional Core: Despite the imperative execution style, Ox maintains functional principles such as using immutable data structures, higher-order functions, typeclasses, and favoring function composition.
• Safety & Resiliency: Provides built-in support for error handling, resource management, scheduling, and safe resiliency utilities.
• Goal: To enable safe direct-style programming specifically optimized for the Scala 3 language.

A channel in Ox acts like a queue for sending and receiving values, but with advanced streaming capabilities including:

• Completion: A source can signal that it is done.
• Error Propagation: Errors can be signaled downstream.
• Select Operations: Support for selecting exactly one channel clause to complete (including send and receive operations).

Channels are lightweight and designed to work with Java 21+ Virtual Threads, making blocking operations (like .send or .receive) cheap and frequent.

Ox provides a high-level, functional API called Flow for defining streaming data transformation pipelines. Flows are lazily-evaluated; they describe a sequence of operations (like map, mapPar, grouped, async, and merge) but do not start processing data until a run method is invoked.

Flows are built on top of channels and structured concurrency (forks). They correspond to a "cold streams" model.

import ox.channels.BufferCapacity
import ox.flow.*

def invokeService(n: Int): String = ???

def sendParsedNumbers(incoming: Flow[String])(using BufferCapacity): Unit =
  incoming
    .mapConcat(_.split(" ").flatMap(_.toIntOption))
    .tap(n => println(s"Got: $n"))
    .mapPar(8)(invokeService)
    .runForeach(r => println("Result: $r"))

The supervised block provides a scope for structured concurrency. Within this scope, you can use fork to spawn new tasks. If any task within the scope fails (throws an exception), all other forks are interrupted (the "let it crash" model). The scope only ends once all forks have completed, ensuring no "leftover" tasks.

// Equivalent of par using structured concurrency
supervised {
  val f1 = fork { sleep(2.seconds); 1 }
  val f2 = fork { sleep(1.second); 2 }
  (f1.join(), f2.join())
}

// Error handling example
supervised {
  forkUser:
    sleep(1.second)
    println("Hello!")

  forkUser:
    sleep(500.millis)
    throw new RuntimeException("boom!")
}

Structured concurrency is a programming model where the lifetime of a thread is tied to the syntactic structure of the code. In Ox, this is implemented using scopes.

Key characteristics include:

• Scope Lifecycle: A scope only finishes once all threads started within it have finished (either successfully or with an error).
• No Thread Leaks: It is impossible to leak threads outside of the method or scope that started them.
• Local Reasoning: Concurrency becomes an implementation detail of a method rather than a side effect, allowing for safer direct-style programming and easier reasoning about threading effects.

Ox offers two complementary streaming APIs that can be mixed within a single pipeline:

You can convert between the two approaches as needed to suit the complexity of your task.

If any channel involved in a select is closed (either in an error state or done state), the select method throws a ChannelClosedException.

To handle this without exceptions, use the safe variants of the select methods (e.g., selectSafe). These return a union type that includes ChannelClosed as a possible result.

You can check the status of a channel using .isClosedForSend and .isClosedForReceive.