Ox

https://github.com/softwaremill/ox
Admin
Safe direct-style concurrency and resiliency for Scala on the JVM
Tokens:41,208
Snippets:507
Trust Score:9.9
License:Apache-2.0
Update:1 month ago
Show doc for...
Tokens:
Raw
================
LIBRARY RULES
================
- Use high-level concurrency operations like par(), race(), and timeout() when possible
- For manual concurrency, always use structured concurrency with supervised scopes to ensure proper cleanup of threads and resources
- Keep concurrency scopes as small as possible, creating short-lived scopes for single requests or jobs
- Use useCloseableInScope() for automatic resource management that ensures cleanup on scope exit
- Handle errors through either exceptions (for bugs/unexpected situations) or application errors as values using Either
- Prefer either blocks with .ok() for streamlined Either handling instead of manual pattern matching
- Use blocking operations freely, instead of using Futures
- Use Flows for defining asynchronous data processing pipelines
- Use .buffer() to create explicit asynchronous boundaries in flows when producers and consumers need decoupling
- Use Flow.usingEmit for custom flow creation but never share FlowEmit instances across threads
- Use channels for integrating with callback-based and reactive APIs within structured concurrency scopes
- Leverage retry() with exponential backoff and jitter for resilient error handling of transient failures
- Use repeat() with appropriate schedules for periodic task execution within supervised scopes
- Use .pipe, .tap, uninterruptible, .discard, debug utility functions

================
CODE SNIPPETS
================
### Example Source and mapParUnordered Usage

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/adr/0001-error-propagation-in-channels.md

Sets up an example `Source` of integers and applies the `mapParUnordered` operator with a mapping function `f` that is designed to fail when processing the third element.

```scala
val input: Source[Int] = Source.fromValues(1, 2, 3, 4)
def f(i: Int): Int = if ()
```

```scala
val result: Source[Int] = input.mapParUnordered(2)(f)
```

--------------------------------

### Basic Rate Limiter Usage

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/utils/rate-limiter.md

Demonstrates how to create and use a fixed window rate limiter with start time. Use `runBlocking` to wait for execution or `runOrDrop` to fail fast.

```scala
import ox.supervised
import ox.resilience.*
import scala.concurrent.duration.*

supervised: 
  val rateLimiter = RateLimiter.fixedWindowWithStartTime(2, 1.second)

  type T
  def operation: T = ???

  val blockedOperation: T = rateLimiter.runBlocking(operation)
  val droppedOperation: Option[T] = rateLimiter.runOrDrop(operation)
```

--------------------------------

### Kafka flow with publish and commit

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/integrations/kafka.md

Example demonstrating a Kafka flow that subscribes to a source topic, transforms the data, and uses KafkaDrain.runPublishAndCommit to handle publishing and offset commits.

```scala
import ox.kafka.{ConsumerSettings, KafkaDrain, KafkaFlow, ProducerSettings, SendPacket}
import ox.kafka.ConsumerSettings.AutoOffsetReset
import ox.*
import org.apache.kafka.clients.producer.ProducerRecord

val consumerSettings = ConsumerSettings.default("my_group")
  .bootstrapServers("localhost:9092").autoOffsetReset(AutoOffsetReset.Earliest)
val producerSettings = ProducerSettings.default.bootstrapServers("localhost:9092")
val sourceTopic = "source_topic"
val destTopic = "dest_topic"

supervised:
  // the consumer is shared between the subscribe & offset stages
  // its lifetime is bound to the current concurrency scope
  val consumer = consumerSettings.toThreadSafeConsumerWrapper
  KafkaFlow
    .subscribe(consumer, sourceTopic)
    .map(in => (in.value.toLong * 2, in))
    .map((value, original) => 
      SendPacket(ProducerRecord[String, String](destTopic, value.toString), original))
    .pipe(KafkaDrain.runPublishAndCommit(producerSettings, consumer))
```

--------------------------------

### Running flows with run methods

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/streaming/flows.md

Examples of executing flows using runToList, runForeach, and runDrain. These methods are blocking operations.

```scala
import ox.flow.Flow
import scala.concurrent.duration.*

Flow.fromValues(1, 2, 3).runToList() // List(1, 2, 3)
Flow.fromValues(1, 2, 3).runForeach(println)
Flow.tick(1.second, "x").runDrain() // never finishes
```

--------------------------------

### Using selectWithin with Channel Sources

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/streaming/selecting-from-channels.md

This example shows how to use `selectWithin` directly with channel sources. A `TimeoutException` is thrown if no data is available within the specified duration, treating the timeout as an error.

```scala
import ox.channels.*
import ox.supervised
import scala.concurrent.duration.*
import scala.concurrent.TimeoutException

supervised:
  val s1 = Channel.rendezvous[Int]
  val s2 = Channel.rendezvous[String] 
  
  try
    val result = selectWithin(50.millis)(s1, s2)
    println(s"Received: $result")
  catch
    case _: TimeoutException => println("No data available within timeout")
```

--------------------------------

### Advanced cron scheduling examples

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/integrations/cron4s.md

Demonstrates various ways to use CronSchedule with different operation types, custom stop strategies, error modes, and nested retry logic.

```scala
import ox.UnionMode
import ox.scheduling.cron.CronSchedule
import scala.concurrent.duration.*
import ox.resilience.{RetryConfig, retry}
import ox.scheduling.*
import cron4s.*

def directOperation: Int = ???
def eitherOperation: Either[String, Int] = ???
def unionOperation: String | Int = ???

val cronExpr: CronExpr = Cron.unsafeParse("10-35 2,4,6 * ? * *")

// various operation definitions - same syntax
repeat(CronSchedule.fromCronExpr(cronExpr))(directOperation)
repeatEither(CronSchedule.fromCronExpr(cronExpr))(eitherOperation)

// infinite repeats with a custom strategy
def customStopStrategy: Int => Boolean = ???
repeat(RepeatConfig(CronSchedule.fromCronExpr(cronExpr), customStopStrategy))(directOperation)

// custom error mode
repeatWithErrorMode(UnionMode[String])(RepeatConfig(CronSchedule.fromCronExpr(cronExpr)))(unionOperation)

// repeat with retry inside
repeat(RepeatConfig(CronSchedule.fromCronExpr(cronExpr))) {
  retry(Schedule.exponentialBackoff(100.millis).maxRetries(3))(directOperation)
}
```

--------------------------------

### Circuit Breaker Usage Examples

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/utils/circuit-breaker.md

Demonstrates various ways to use the CircuitBreaker, including running operations directly, handling Either results, and integrating with retry mechanisms. It shows how to run operations that might drop or return specific error types.

```scala
import ox.UnionMode
import ox.resilience.*
import ox.scheduling.Schedule
import ox.supervised
import scala.concurrent.duration.*

def directOperation: Int = ???
def eitherOperation: Either[String, Int] = ???
def unionOperation: String | Int = ???

supervised:
  val circuitBreaker = CircuitBreaker()

  // various operation definitions
  circuitBreaker.runOrDrop(directOperation)
  circuitBreaker.runOrDropEither(eitherOperation)

  // custom error mode
  circuitBreaker.runOrDropWithErrorMode(UnionMode[String])(unionOperation)

  // retry with circuit breaker inside
  retryEither(Schedule.exponentialBackoff(100.millis).maxRetries(3)){
    circuitBreaker.runOrDrop(directOperation) match
      case Some(value) => Right(value)
      case None -> Left("Operation dropped")
  }
```

--------------------------------

### Rate Limiter Algorithms

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/utils/rate-limiter.md

Lists and describes the available rate limiting algorithms, including those based on start time and duration of operations.

```APIDOC
## Rate Limiter Algorithms

### Description
Lists and describes the available rate limiting algorithms, including those based on start time and duration of operations.

### Algorithms
- `StartTimeRateLimiterAlgorithm.FixedWindow(rate: Int, per: FiniteDuration)`: Maximum operations in fixed windows.
- `StartTimeRateLimiterAlgorithm.SlidingWindow(rate: Int, per: FiniteDuration)`: Maximum operations in any window.
- `StartTimeRateLimiterAlgorithm.LeakyBucket(maximum: Int, per: FiniteDuration)`: Token bucket capacity and refill interval.
- `DurationRateLimiterAlgorithm.FixedWindow(rate: Int, per: FiniteDuration)`: Operations spanning fixed windows, considering execution time.
- `DurationRateLimiterAlgorithm.SlidingWindow(rate: Int, per: FiniteDuration)`: Operations spanning any window, considering execution time.
```

--------------------------------

### Reading from stdin with abandonOnInterruptReads

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/structured-concurrency/interruptions.md

Example of wrapping System.in to make blocking reads interruptible within an Ox supervised scope.

```scala
import ox.*
import scala.concurrent.duration.*

// one process-wide wrapper: multiple wrappers over System.in would compete for input
lazy val stdin = abandonOnInterruptReads(System.in)

supervised {
  val firstByte: Option[Int] = timeoutOption(1.second)(stdin.read())
  println(s"Read: $firstByte")
}
```

--------------------------------

### Using selectOrClosedWithin with Channel Sources

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/streaming/selecting-from-channels.md

This example demonstrates using `selectOrClosedWithin` directly with channel sources. It returns a specified default value if no source provides data within the given timeout. This is suitable when you want to handle timeouts gracefully.

```scala
import ox.channels.*
import ox.supervised
import scala.concurrent.duration.*

supervised:
  val s1 = Channel.rendezvous[Int] 
  val s2 = Channel.rendezvous[String]
  
  // Returns -1 if no source has a value within 50ms
  val result = selectOrClosedWithin(50.millis, -1)(s1, s2)
  
  result match
    case value: Int if value == -1 => println("Timeout occurred")
    case value: Int                => println(s"Received int: $value")
    case value: String             => println(s"Received string: $value")
    case closed: ChannelClosed     => println(s"Channel closed: $closed")
```

--------------------------------

### Override OxApp Settings

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/utils/oxapp.md

Customize application settings by overriding the `settings` method. This example shows how to set a custom exit code for interrupted shutdowns.

```scala
import ox.*
import scala.concurrent.duration.*

object MyApp extends OxApp:
  override def settings: OxApp.Settings = OxApp.Settings.Default.copy(
    interruptedExitCode = ExitCode.Failure(130)
  )
  
  def run(args: Vector[String])(using Ox): ExitCode =
    sleep(60.seconds)
    ExitCode.Success
```

--------------------------------

### Unsupervised scope with `unsupervised`

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/structured-concurrency/fork-join.md

Use `unsupervised` to create a scope where only `forkUnsupervised` and `forkCancellable` can be started. Fork failures are not automatically handled and must be inspected via `Fork.join()`.

```scala
import ox.{forkUnsupervised, sleep, unsupervised}
import scala.concurrent.duration.*

unsupervised {
  val f1 = forkUnsupervised {
    sleep(1.second)
    1
  }
  val f2 = forkUnsupervised {
    sleep(2.seconds)
    2
  }
  (f1.join(), f2.join())
}
```

--------------------------------

### Equivalent to `par` using `supervised`

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/structured-concurrency/fork-join.md

Use `supervised` to create a concurrency scope. Forks started within this scope are managed by the scope's lifetime. The scope completes only after all forks have finished.

```scala
import ox.{fork, sleep, supervised}
import scala.concurrent.duration.*

supervised {
  val f1 = fork {
    sleep(2.seconds)
    1
  }

  val f2 = fork {
    sleep(1.second)
    2
  }

  (f1.join(), f2.join())
}
```

--------------------------------

### Creating basic Flows

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/streaming/flows.md

Demonstrates creating finite and infinite flows using the Flow companion object methods.

```scala
import ox.flow.Flow
import scala.concurrent.duration.*

Flow.fromValues(1, 2, 3) // a finite flow
Flow.tick(1.second, "x") // an infinite flow, emitting "x" every second
Flow.iterate(0)(_ + 1) // natural numbers
```

--------------------------------

### Create sources from values and flows

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/streaming/channels.md

Shows how to create sources from a sequence of values or by running flows, such as ticking at a specific interval or generating natural numbers.

```scala
import ox.supervised
import ox.channels.Source
import ox.flow.Flow

import scala.concurrent.duration.*

supervised:
  Source.fromValues(1, 2, 3)
  Flow.tick(1.second, "x").runToChannel()
  Flow.iterate(0)(_ + 1).runToChannel() // natural numbers
```

--------------------------------

### Compile and test commands

Source: https://github.com/softwaremill/ox/blob/master/README.md

Standard sbt commands for compiling and testing the project.

```bash
sbt compile
sbt test
```

--------------------------------

### Creating and using an Actor with `ask`

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/utils/actors.md

Demonstrates creating a stateful actor and invoking its methods using the `ask` method, which blocks until the invocation completes. Exceptions are propagated to the caller.

```scala
import ox.supervised
import ox.channels.*

class Stateful:
  private var counter: Int = 0
  def increment(delta: Int): Int =
    counter += delta
    counter

supervised:
  val ref = Actor.create(new Stateful)

  ref.ask(_.increment(5)) // blocks until the invocation completes
  ref.ask(_.increment(4)) // returns 9
```

--------------------------------

### Basic Rate Limiter Usage

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/utils/rate-limiter.md

Demonstrates how to create and use a fixed window rate limiter. `runBlocking` waits until the operation can be executed, while `runOrDrop` returns `None` if the operation cannot be executed immediately.

```APIDOC
## Basic Rate Limiter Usage

### Description
Demonstrates how to create and use a fixed window rate limiter. `runBlocking` waits until the operation can be executed, while `runOrDrop` returns `None` if the operation cannot be executed immediately.

### Method
```scala
import ox.supervised
import ox.resilience.*
import scala.concurrent.duration.*

supervised: 
  val rateLimiter = RateLimiter.fixedWindowWithStartTime(2, 1.second)

  type T
  def operation: T = ???

  val blockedOperation: T = rateLimiter.runBlocking(operation)
  val droppedOperation: Option[T] = rateLimiter.runOrDrop(operation)
```

### Notes
A rate limiter must be created within an `Ox` concurrency scope. The operation can be provided directly using a by-name parameter (`f: => T`).
```

--------------------------------

### Create channels with different buffer types

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/streaming/channels.md

Demonstrates creating rendezvous, buffered, and unlimited channels with specified buffer sizes.

```scala
import ox.channels.*
val c1 = Channel.rendezvous[String]
val c2 = Channel.buffered[String](5)
val c3 = Channel.unlimited[String]
```

--------------------------------

### Basic AdaptiveRetry Usage

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/retries.md

Demonstrates basic retry configurations with different schedules like immediate, fixed interval, and exponential backoff, all with a default ResultPolicy and a maximum of 3 retries.

```scala
import ox.UnionMode
import ox.resilience.AdaptiveRetry
import ox.resilience.{ResultPolicy, RetryConfig}
import ox.scheduling.{Jitter, Schedule}
import scala.concurrent.duration.*

def directOperation: Int = ???
def eitherOperation: Either[String, Int] = ???
def unionOperation: String | Int = ???

val adaptive = AdaptiveRetry.default

// various configs with custom schedules and default ResultPolicy
adaptive.retry(Schedule.immediate.maxRetries(3))(directOperation)
adaptive.retry(Schedule.fixedInterval(100.millis).maxRetries(3))(directOperation)
adaptive.retry(Schedule.exponentialBackoff(100.millis).maxRetries(3))(directOperation)
adaptive.retry(Schedule.exponentialBackoff(100.millis).maxRetries(3).jitter()
  .maxInterval(5.minutes))(directOperation)
```

--------------------------------

### Schedule Creation

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/retries.md

Various methods are available to construct `Schedule` instances, from immediate retries to exponential backoff.

```APIDOC
## Schedule Creation

### Description
This section details the factory methods for creating `Schedule` objects, which dictate the retry timing and limits.

### Factory Methods
* `Schedule.immediate`
* `Schedule.fixedInterval(interval: FiniteDuration)`
* `Schedule.intervals(intervals: FiniteDuration*)`
* `Schedule.exponentialBackoff(initialDelay: FiniteDuration)`
* `Schedule.fibonacciBackoff(initialDelay: FiniteDuration)`
* `Schedule.decorrelatedJitter(min: FiniteDuration)`
* `Schedule.computed[S](initialState: S, compute: S => (S, Option[FiniteDuration]))`

### Customization Methods
* `.maxRetries(maxRetries: Int)`
* `.maxAttempts(maxAttempts: Int)`
* `.maxInterval(maxInterval: FiniteDuration)`
* `.jitter(jitter: Jitter)`
* `.maxRetriesByCumulativeDelay(upTo: FiniteDuration)`
* `.withInitialDelay(interval: FiniteDuration)`
* `.withNoInitialDelay()`
* `.andThen(other: Schedule)`
```

--------------------------------

### Race computations with raceEither

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/high-level-concurrency/race.md

Use `raceEither` to race computations that return `Either` types. It returns the first `Right` value encountered, or the first `Left` if all computations result in `Left`. The third computation is cancelled in this example.

```scala
import ox.{raceEither, sleep}
import scala.concurrent.duration.*

raceEither({
  sleep(200.millis)
  Left(-1)
}, {
  sleep(500.millis)
  Right("ok")
}, {
  sleep(1.second)
  Right("also ok")
})
```

--------------------------------

### Transforming a Channel with map

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/streaming/transforming-channels.md

Use the `.map` method to transform values received from a source channel. This starts a new virtual thread to process values and send them to a new channel. The new channel is returned as the result of `.map`.

```scala
import ox.supervised
import ox.channels.{Channel, Source}

supervised:
  val c = Channel.rendezvous[String]
  val c2: Source[Int] = c.map(s => s.length())
```

--------------------------------

### Basic Circuit Breaker Usage

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/utils/circuit-breaker.md

Demonstrates the basic usage of the CircuitBreaker by running an operation and potentially dropping it if the breaker is open. Requires importing `ox.supervised` and `ox.resilience`.

```scala
import ox.supervised
import ox.resilience.*

supervised:
  val circuitBreaker = CircuitBreaker()

  type T
  def operation: T = ???

  val operationResult: Option[T] = circuitBreaker.runOrDrop(operation)
```

--------------------------------

### Structured concurrency and supervision

Source: https://github.com/softwaremill/ox/blob/master/README.md

Demonstrates the use of supervised scopes and fork/join for structured concurrency.

```scala
// equivalent of par
supervised {
  val f1 = fork { sleep(2.seconds); 1 }
  val f2 = fork { sleep(1.second); 2 }
  (f1.join(), f2.join())
}
```

--------------------------------

### Simplified OxApp Entry Point

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/utils/oxapp.md

A simplified variant of OxApp for cases where command line arguments are not needed. The run function expects no arguments beyond the root Ox capability and handles exceptions by printing a stack trace and returning an exit code of 1.

```scala
import ox.*

object MyApp extends OxApp.Simple:
  def run(using Ox): Unit = println("All done!")
```

--------------------------------

### KafkaFlow.subscribe and KafkaDrain.runPublishAndCommit

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/integrations/kafka.md

Demonstrates how to consume messages from a source topic, transform them, and publish to a destination topic while managing offset commits.

```APIDOC
## Kafka Integration: Publish and Commit

### Description
This pattern allows for transactional-like processing where messages from a source topic are consumed, processed, and published to a destination topic. Offsets for the source topic are committed only after the destination messages are successfully published.

### Usage
1. Define `ConsumerSettings` and `ProducerSettings`.
2. Use `KafkaFlow.subscribe` to create a stream of `ReceivedMessage`.
3. Map the stream to `SendPacket` objects containing the `ProducerRecord` to send and the original `ReceivedMessage` to commit.
4. Use `KafkaDrain.runPublishAndCommit` to execute the pipeline.

### Data Structure: SendPacket
- **send** (List[ProducerRecord[K, V]]) - The list of messages to be published to the destination topic.
- **commit** (List[ReceivedMessage[_, _]]) - The list of received messages to be committed after publication.

### Example
```scala
supervised:
  val consumer = consumerSettings.toThreadSafeConsumerWrapper
  KafkaFlow
    .subscribe(consumer, sourceTopic)
    .map(in => (in.value.toLong * 2, in))
    .map((value, original) => 
      SendPacket(ProducerRecord[String, String](destTopic, value.toString), original))
    .pipe(KafkaDrain.runPublishAndCommit(producerSettings, consumer))
```
```

--------------------------------

### Creating Flows from Channels

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/streaming/flows.md

Shows how to create a flow from a channel Source within a supervised scope.

```scala
import ox.channels.Channel
import ox.flow.Flow
import ox.{fork, supervised}

val ch = Channel.bufferedDefault[Int]
supervised:
  fork:
    ch.send(1)
    ch.send(15)
    ch.send(-2)
    ch.done()

  Flow.fromSource(ch) // TODO: transform the flow further & run
```

--------------------------------

### Define Application Entry Point with OxApp

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/utils/oxapp.md

Use OxApp to define your application's main run function. The application's main run function is executed on a virtual thread, with a root Ox capability provided. Handles command line arguments and returns an ExitCode.

```scala
import ox.*
import scala.concurrent.duration.*

object MyApp extends OxApp:
  def run(args: Vector[String])(using Ox): ExitCode =
    forkUser {
      sleep(500.millis)
      println("Fork finished!")
    }
    println(s"Started app with args: ${args.mkString(", ")}!")
    ExitCode.Success
```

--------------------------------

### Manage multiple resources with resourceScope

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/utils/resources.md

Demonstrates using resourceScope to manage multiple AutoCloseable resources, ensuring they are closed in reverse order when the scope completes.

```scala
import ox.{resourceScope, useCloseableInScope}
import java.io.{FileReader, FileWriter}

def process(): Unit = resourceScope {
  val in = useCloseableInScope(new FileReader("in.txt"))
  val out = useCloseableInScope(new FileWriter("out.txt"))
  // both closed when the scope completes, out first
  out.write(in.read())
}
```

--------------------------------

### Transforming a channel into a flow

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/streaming/flows.md

Demonstrates creating a flow from a source channel, applying transformations, and running the result back to a channel. This requires a concurrency scope as .runToChannel() initiates a background fork.

```scala
import ox.Ox
import ox.channels.{BufferCapacity, Source}
import ox.flow.Flow

def transformChannel(ch: Source[String])(using Ox, BufferCapacity): Source[Int] =
  Flow.fromSource(ch)
    .mapConcat(_.split(" "))
    .mapConcat(_.toIntOption)
    .filter(_ % 2 == 0)
    .runToChannel()
```

--------------------------------

### Using ForkLocal for Context Propagation

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/structured-concurrency/fork-local.md

Demonstrates creating a ForkLocal, setting its value within a scope, and observing propagation across nested forks. Values are retrieved using `.get()` and can be temporarily overridden within a scope using `supervisedWhere`.

```scala
import ox.{ForkLocal, fork, supervised}

val v = ForkLocal("a")
supervised {
  println(v.get()) // "a"
  fork {
    v.supervisedWhere("x") {
      println(v.get()) // "x"
      fork {
        println(v.get()) // "x"
      }.join()
    }
  }.join()
  println(v.get()) // "a"
}
```

--------------------------------

### Custom Rate Limiter Algorithms

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/utils/rate-limiter.md

Explains how to extend the `RateLimiterAlgorithm` interface to implement custom algorithms or modify existing ones, including methods for state updates.

```APIDOC
## Custom Rate Limiter Algorithms

### Description
Explains how to extend the `RateLimiterAlgorithm` interface to implement custom algorithms or modify existing ones, including methods for state updates.

### Interface
The `RateLimiterAlgorithm` interface is modeled like a `Semaphore` with `acquire` and `tryAcquire` methods. It also includes `runOperation` for operations considering execution span.

### State Update Methods
- `def update(): Unit`: Updates the internal state of the rate limiter.
- `def getNextUpdate: Long`: Returns the time in nanoseconds after which a new `update` needs to be called.
```

--------------------------------

### Basic Repeat with Schedule

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/repeat.md

Use `repeat` with a `Schedule` and a direct operation returning a value.

```scala
import ox.scheduling.repeat

repeat(schedule)(operation)
```

--------------------------------

### API Shorthands for Rate Limiter Configuration

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/utils/rate-limiter.md

Provides convenient shorthand methods for creating RateLimiter instances with specific algorithms.

```APIDOC
## API Shorthands for Rate Limiter Configuration

### Description
Provides convenient shorthand methods for creating RateLimiter instances with specific algorithms.

### Shorthands
- `RateLimiter.fixedWindowWithStartTime(maxOperations: Int, window: FiniteDuration)`
- `RateLimiter.slidingWindowWithStartTime(maxOperations: Int, window: FiniteDuration)`
- `RateLimiter.leakyBucket(maxTokens: Int, refillInterval: FiniteDuration)`
- `RateLimiter.fixedWindowWithDuration(maxOperations: Int, window: FiniteDuration)`
- `RateLimiter.slidingWindowWithDuration(maxOperations: Int, window: FiniteDuration)
```

--------------------------------

### Custom Retry Schedules

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/retries.md

Shows how to configure retry mechanisms with different schedules like fixed interval, exponential backoff, and jitter, along with maximum retry intervals.

```scala
retry(Schedule.fixedInterval(100.millis).maxRetries(3))(directOperation)
retry(Schedule.exponentialBackoff(100.millis).maxRetries(3))(directOperation)
retry(Schedule.exponentialBackoff(100.millis).maxRetries(3).jitter().maxInterval(5.minutes))(
  directOperation)
```

--------------------------------

### AdaptiveRetry with Custom Result Policies

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/retries.md

Shows how to define custom success and retry conditions using ResultPolicy for operations returning Int. It also demonstrates retrying Either operations based on custom conditions.

```scala
// result policies
// custom success
adaptive.retry(RetryConfig[Throwable, Int](
  Schedule.immediate.maxRetries(3), ResultPolicy.successfulWhen(_ > 0)))(directOperation)
// fail fast on certain errors
adaptive.retry(RetryConfig[Throwable, Int](
  Schedule.immediate.maxRetries(3), ResultPolicy.retryWhen(_.getMessage != "fatal error")))(
    directOperation)
adaptive.retryEither(RetryConfig(
  Schedule.immediate.maxRetries(3), ResultPolicy.retryWhen(_ != "fatal error")))(
    eitherOperation)
```

--------------------------------

### Basic Retry Operations

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/retries.md

Demonstrates the basic usage of retry for different operation signatures (direct, Either, union types) with immediate retry schedules.

```scala
import ox.UnionMode
import ox.resilience.{retry, retryEither, retryWithErrorMode, ResultPolicy, RetryConfig}
import ox.scheduling.{Jitter, Schedule}
import scala.concurrent.duration.*

def directOperation: Int = ???
def eitherOperation: Either[String, Int] = ???
def unionOperation: String | Int = ???

// various operation signatures/error modes - same syntax
retry(Schedule.immediate.maxRetries(3))(directOperation)
retryEither(Schedule.immediate.maxRetries(3))(eitherOperation)
```

--------------------------------

### Resource Release with OxApp

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/utils/oxapp.md

Demonstrates how resources are released when the application is interrupted using OxApp. The `releaseAfterScope` ensures cleanup actions are performed.

```scala
import ox.*

object MyApp extends OxApp:
  def run(args: Vector[String])(using Ox): ExitCode =
    releaseAfterScope:
      println("Releasing ...")
    println("Waiting ...")
    never
```

--------------------------------

### Channel rendezvous selection

Source: https://github.com/softwaremill/ox/blob/master/README.md

Demonstrates creating rendezvous channels and using the select clause to perform operations.

```scala
val c = Channel.rendezvous[Int]
val d = Channel.rendezvous[Int]
select(c.sendClause(10), d.receiveClause)
```

--------------------------------

### Transforming flows with pre-defined stages

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/streaming/flows.md

Demonstrates chaining transformation stages like map, filter, take, zip, and interleave on a Flow instance. Transformations are lazy and do not execute until the flow is run.

```scala
import ox.flow.Flow

Flow.fromValues(1, 2, 3, 5, 6)
  .map(_ * 2)
  .filter(_ % 2 == 0)
  .take(3)
  .zip(Flow.repeat("a number"))
  .interleave(Flow.repeat((0, "also a number")))
  // etc., TODO: run the flow
```

--------------------------------

### Create an OxApp

Source: https://github.com/softwaremill/ox/blob/master/README.md

Defines an application that shuts down cleanly upon receiving termination signals.

```scala
object MyApp extends OxApp:
  def run(args: Vector[String])(using Ox): ExitCode =
    // ... your app's code ...
    // might use fork {} to create top-level background threads
    ExitCode.Success
```

--------------------------------

### Basic Retry Syntax

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/retries.md

The basic syntax for using the retry mechanism. It takes a schedule and an operation to perform.

```scala
import ox.resilience.retry

retry(schedule)(operation)
```

--------------------------------

### Configure Circuit Breaker Instance

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/utils/circuit-breaker.md

Instantiate a CircuitBreaker with default or custom configuration. Custom configurations allow fine-tuning of failure thresholds, slow call settings, windowing, and state transition timings.

```scala
import ox.supervised
import ox.resilience.*
import scala.concurrent.duration.*

supervised:
  // using default config
  CircuitBreaker()
  // or
  CircuitBreaker(CircuitBreakerConfig.default)
  
  // custom config
  val config = CircuitBreakerConfig(
    failureRateThreshold = PercentageThreshold(50),
    slowCallThreshold = PercentageThreshold(50),
    slowCallDurationThreshold = 10.seconds,
    slidingWindow = SlidingWindow.CountBased(100),
    minimumNumberOfCalls = 20,
    waitDurationOpenState = 10.seconds,
    halfOpenTimeoutDuration = 0.millis,
    numberOfCallsInHalfOpenState = 10
  )
  
  // providing config for CircuitBreaker instance
  CircuitBreaker(config)
```

--------------------------------

### Infinite Retries

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/retries.md

Illustrates how to configure retries for an indefinite number of attempts using fixed interval or exponential backoff schedules with jitter.

```scala
retry(Schedule.fixedInterval(100.millis))(directOperation)
retry(Schedule.exponentialBackoff(100.millis).jitter(Jitter.Full).maxInterval(5.minutes))(
  directOperation)
```

--------------------------------

### Logging flow elements with .tap

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/streaming/flows.md

Use the .tap method to log elements emitted by a flow without integrating external logging libraries.

```scala
import ox.flow.Flow

Flow.fromValues(1, 2, 3)
  .tap(n => println(s"Received: $n"))
  .runToList()
```

--------------------------------

### Create a buffered channel

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/streaming/channels.md

Creates a buffered channel using the default buffer size. Other buffer sizes, as well as rendezvous and unlimited channels, can also be created.

```scala
import ox.channels.*
val c = Channel.bufferedDefault[String]
```

--------------------------------

### Schedule: Intervals from List

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/retries.md

Creates a schedule using a predefined list of intervals between retries.

```scala
Schedule.intervals(intervals: FiniteDuration*)
```

--------------------------------

### Schedule: Immediate Retries

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/retries.md

Creates a schedule that retries immediately after a failure.

```scala
Schedule.immediate
```

--------------------------------

### Repeat with Initial Delay

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/repeat.md

Repeats an operation with a 100ms fixed interval and an initial delay of 50ms before the first attempt, up to 3 attempts.

```scala
repeat(Schedule.fixedInterval(100.millis).maxAttempts(3).withInitialDelay(50.millis))(directOperation)
```

--------------------------------

### Schedule Customization: And Then

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/retries.md

Chains another schedule to be executed after the current one is exhausted.

```scala
schedule.andThen(other: Schedule)
```

--------------------------------

### Create a basic cron schedule

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/integrations/cron4s.md

Use CronSchedule.unsafeFromString to create a schedule from a cron expression string for use with the repeat API.

```scala
import ox.scheduling.cron.*
import cron4s.*

repeat(CronSchedule.unsafeFromString("10-35 2,4,6 * ? * *"))(operation)
```

--------------------------------

### Use InheritableMDC.supervisedWhere

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/integrations/mdc-logback.md

Demonstrates setting inheritable MDC values that propagate to forks, while standard MDC values remain local to the current thread.

```scala
import org.slf4j.MDC

import ox.fork
import ox.logback.InheritableMDC

InheritableMDC.supervisedWhere("a" -> "1", "b" -> "2") {
  MDC.put("c", "3") // not inherited

  fork {
    MDC.get("a") // "1"
    MDC.get("b") // "2"
    MDC.get("c") // null
  }.join()

  MDC.get("a") // "1"
  MDC.get("b") // "2"
  MDC.get("c") // "3"
}
```

--------------------------------

### Run Flow to File

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/streaming/io.md

Write a Flow[Chunk[Byte]] to a file. Ox handles closing the file.

```scala
import ox.flow.Flow
import java.nio.file.Paths

Flow.fromIterable(List("text1,", "text2"))
  .encodeUtf8
  .runToFile(Paths.get("/path/to/my/target/file.txt"))
```

--------------------------------

### Publish to Kafka topic

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/integrations/kafka.md

Use KafkaDrain.runPublish to publish data to a Kafka topic as a sink.

```scala
import ox.flow.Flow
import ox.kafka.{ProducerSettings, KafkaDrain}
import ox.pipe
import org.apache.kafka.clients.producer.ProducerRecord

val settings = ProducerSettings.default.bootstrapServers("localhost:9092")
Flow
  .fromIterable(List("a", "b", "c"))
  .map(msg => ProducerRecord[String, String]("my_topic", msg))
  .pipe(KafkaDrain.runPublish(settings))
```

--------------------------------

### Schedule Customization: With Initial Delay

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/retries.md

Sets a specific initial delay for the first retry.

```scala
schedule.withInitialDelay(interval: FiniteDuration)
```

--------------------------------

### Register release logic with releaseAfterScope

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/utils/resources.md

Registers a release action for a resource to be executed when the current scope completes.

```scala
import ox.{supervised, releaseAfterScope}

case class MyResource(c: Int)

def release(resource: MyResource): Unit =
  println(s"releasing ${resource.c} ...")

supervised {
  val resource1 = MyResource(10)
  releaseAfterScope(release(resource1))
  println(s"Using $resource1 ...")
}
```

--------------------------------

### Create and transform a flow

Source: https://github.com/softwaremill/ox/blob/master/README.md

Demonstrates functional transformations on a stream of data using the Flow API.

```scala
Flow.iterate(0)(_ + 1) // natural numbers
  .filter(_ % 2 == 0)
  .map(_ + 1)
  .intersperse(5)
  // compute the running total
  .mapStateful(0) { (state, value) =>
    val newState = state + value
    (newState, newState)
  }
  .take(10)
  .runForeach(n => println(n.toString))
```

--------------------------------

### Basic Retry Usage

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/retries.md

The basic syntax for using the retry mechanism involves providing a schedule and the operation to be retried.

```APIDOC
## Basic Retry Usage

### Description
This demonstrates the fundamental way to apply retries to an operation using a predefined schedule.

### Method Signature
```scala
import ox.resilience.retry

retry(schedule)(operation)
```

### Parameters
* `schedule`: A `Schedule` instance defining the retry strategy (e.g., delays, number of attempts).
* `operation`: A by-name parameter `f: => T` representing the operation to be executed and potentially retried.
```

--------------------------------

### Use resourceScope inside a fork

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/utils/resources.md

Illustrates extracting a resource scope to a method to safely use it within a concurrency scope.

```scala
import ox.{forkDiscard, resourceScope, supervised, useCloseableInScope}
import java.io.FileReader

def handleRequest(): Unit = resourceScope {
  val in = useCloseableInScope(new FileReader("in.txt"))
  println(s"Processing the request using: ${in.read()}")
}

supervised {
  forkDiscard(handleRequest())
}
```

--------------------------------

### Add Kafka dependency

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/integrations/kafka.md

Include the Ox Kafka library in your project dependencies.

```scala
"com.softwaremill.ox" %% "kafka" % "1.0.6"
```

--------------------------------

### Repeat with Immediate Schedule and Max Attempts

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/repeat.md

Repeats an operation immediately up to a maximum of 3 attempts.

```scala
repeat(Schedule.immediate.maxAttempts(3))(directOperation)
```

--------------------------------

### Using Actor `tell` for background invocation

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/utils/actors.md

Shows how to use the `tell` method to schedule an invocation for background processing. This method returns immediately after the invocation is sent to the actor's channel, without waiting for execution. Exceptions from `tell` invocations are propagated to the actor's enclosing scope.

```scala
import ox.{never, supervised}
import ox.channels.*

class Stateful:
  def work(howHard: Int): Unit = throw new RuntimeException("boom!")
  def close(): Unit = println("Closing")  

supervised:
  val ref = Actor.create(new Stateful, Some(_.close()))

  // fire-and-forget, exception causes the scope to close
  ref.tell(_.work(5)) 
  
  // preventing the scope from closing
  never
```

--------------------------------

### Custom Result Policies

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/retries.md

Demonstrates how to define custom success and failure conditions for retries using `ResultPolicy`. This includes defining success based on a return value and retrying only for specific error messages.

```scala
// custom success
retry(RetryConfig[Throwable, Int](
  Schedule.immediate.maxRetries(3), ResultPolicy.successfulWhen(_ > 0)))(directOperation)
// fail fast on certain errors
retry(RetryConfig[Throwable, Int](
  Schedule.immediate.maxRetries(3), ResultPolicy.retryWhen(_.getMessage != "fatal error")))(
    directOperation)
retryEither(RetryConfig(
  Schedule.immediate.maxRetries(3), ResultPolicy.retryWhen(_ != "fatal error")))(
    eitherOperation)
```

--------------------------------

### Reading from Kafka and committing offsets with runCommit

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/integrations/kafka.md

Uses a thread-safe consumer wrapper to subscribe to a topic, process messages in parallel, and commit offsets using KafkaDrain.runCommit.

```scala
import ox.kafka.{ConsumerSettings, KafkaDrain, KafkaFlow, CommitPacket}
import ox.kafka.ConsumerSettings.AutoOffsetReset
import ox.*

val consumerSettings = ConsumerSettings.default("my_group")
  .bootstrapServers("localhost:9092").autoOffsetReset(AutoOffsetReset.Earliest)
val sourceTopic = "source_topic"

supervised:
  // the consumer is shared between the subscribe & offset stages
  // its lifetime is bound to the current concurrency scope
  val consumer = consumerSettings.toThreadSafeConsumerWrapper
  KafkaFlow
    .subscribe(consumer, sourceTopic)
    .mapPar(10) { in => 
      // process the message, e.g. send an HTTP request
      CommitPacket(in)
    }
    .pipe(KafkaDrain.runCommit(consumer))
```

--------------------------------

### Executing flow pipelines on the calling thread

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/streaming/flows.md

Shows a pipeline that processes elements one-by-one on the thread that invokes the runToList method.

```scala
import ox.flow.Flow

Flow.fromValues(1, 2, 3, 5, 6)
  .map(_ * 2)
  .filter(_ % 2 == 0)
  .runToList()
```

--------------------------------

### Integrate flow with imperative API

Source: https://github.com/softwaremill/ox/blob/master/README.md

Uses the Flow.usingEmit helper to bridge imperative code with the flow API.

```scala
def readNextBatch(): List[String] = ???
Flow.usingEmit { emit =>
  forever:
    readNextBatch().foreach(emit.apply)
}
```

--------------------------------

### Add cron dependency

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/integrations/cron4s.md

Include this dependency in your build file to use the Ox cron module.

```scala
"com.softwaremill.ox" %% "cron" % "1.0.6"
```

--------------------------------

### Add ox core dependency

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/info/dependency.md

Configuration snippets for including the ox core library in sbt and scala-cli projects.

```scala
// sbt dependency
"com.softwaremill.ox" %% "core" % "1.0.6"

// scala-cli dependency
//> using dep com.softwaremill.ox::core:1.0.6
```

--------------------------------

### Create Flow from File

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/streaming/io.md

Obtain a Flow of byte chunks read from a file. Ox handles closing the file.

```scala
import ox.flow.Flow
import java.nio.file.Paths

Flow
  .fromFile(Paths.get("/path/to/my/file.txt"))
  .linesUtf8
  .map(_.toUpperCase)
  .runToList() // List("FILE_LINE1", "FILE_LINE2")
```

--------------------------------

### Register resources in a concurrency scope

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/utils/resources.md

Shows how resources registered within a supervised concurrency scope are released after all forks finish.

```scala
import ox.{supervised, useInScope}

case class MyResource(c: Int)

def acquire(c: Int): MyResource =
  println(s"acquiring $c ...")
  MyResource(c)

def release(resource: MyResource): Unit =
  println(s"releasing ${resource.c} ...")

supervised {
  val resource1 = useInScope(acquire(10))(release)
  val resource2 = useInScope(acquire(20))(release)
  println(s"Using $resource1 ...")
  println(s"Using $resource2 ...")
}
```

--------------------------------

### Pipe and tap control flow

Source: https://github.com/softwaremill/ox/blob/master/README.md

Uses pipe and tap to chain function calls and side effects using dot-syntax.

```scala
def compute: Int = ???
def computeMore(v: Int): Long = ???
compute
  .pipe(2 * _)
  .tap(println)
  .pipe(computeMore)
```

--------------------------------

### Creating Flows using FlowEmit

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/streaming/flows.md

Creates a flow by providing custom logic to emit elements. Note that FlowEmit instances are thread-unsafe and must not be shared across threads.

```scala
import ox.flow.Flow

def isNoon(): Boolean = ???

Flow.usingEmit: emit =>
  emit(1)
  for i <- 4 to 50 do emit(i)
  if isNoon() then emit(42)
```

--------------------------------

### repeatUntil

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/utils/control-flow.md

Repeatedly evaluates the given code block until it returns true.

```APIDOC
## repeatUntil

### Description
Repeatedly evaluates the given code block, until it returns true.

### Signature
`repeatUntil(block: => Boolean): Unit`
```

--------------------------------

### Repeat with Custom Continuation Strategy

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/repeat.md

Repeats an operation indefinitely with a 100ms fixed interval, using a custom strategy to decide whether to continue after a successful result. The loop stops if the operation returns an error.

```scala
def customStopStrategy: Int => Boolean = ???
repeat(RepeatConfig(Schedule.fixedInterval(100.millis))    .copy(shouldContinueOnResult = customStopStrategy))(directOperation)
```

--------------------------------

### Read from Kafka topic

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/integrations/kafka.md

Use KafkaFlow.subscribe to consume messages from a Kafka topic with specified consumer settings.

```scala
import ox.kafka.{ConsumerSettings, KafkaFlow, ReceivedMessage}
import ox.kafka.ConsumerSettings.AutoOffsetReset

val settings = ConsumerSettings.default("my_group").bootstrapServers("localhost:9092")
  .autoOffsetReset(AutoOffsetReset.Earliest)
val topic = "my_topic"
  
val source = KafkaFlow.subscribe(settings, topic)
  .runForeach { (msg: ReceivedMessage[String, String]) => ??? }
```

--------------------------------

### Add mdc-logback dependency

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/integrations/mdc-logback.md

Include the mdc-logback library in your Scala project dependencies.

```scala
"com.softwaremill.ox" %% "mdc-logback" % "1.0.6"
```

--------------------------------

### repeatWhile

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/utils/control-flow.md

Repeatedly evaluates the given code block as long as it returns true.

```APIDOC
## repeatWhile

### Description
Repeatedly evaluates the given code block, as long as it returns true.

### Signature
`repeatWhile(block: => Boolean): Unit`
```

--------------------------------

### Joining and unwrapping forked Either computations

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/basics/error-handling.md

Join a forked computation that returns an `Either` and unwrap its result using a single `.ok()` invocation within an `either` block.

```scala
import ox.{either, fork, Fork, supervised}
import ox.either.ok

val v1: Either[Int, String] = ???

supervised:
  val forkedResult: Fork[Either[Int, String]] = fork(either(v1.ok()))

  val result: Either[Int, String] = either:
    forkedResult.ok()
```

--------------------------------

### Add otel-context dependency

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/integrations/otel-context.md

Include the ox otel-context library in your project dependencies.

```scala
"com.softwaremill.ox" %% "otel-context" % "1.0.6"
```

--------------------------------

### Unwrapping Options with .ok() in an either block

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/basics/error-handling.md

Unwrap `Option` values within an `either` block using `.ok()`. If an `Option` is `None`, the computation short-circuits with `Unit` as the error type.

```scala
import ox.either
import ox.either.ok

val v1: Option[String] = ???
val v2: Option[Int] = ???

val result: Either[Unit, String] = either:
  v1.ok() * v2.ok()
```

--------------------------------

### Error handling in structured concurrency

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/index.md

Shows how errors in a fork are handled within a supervised scope.

```scala
supervised {
  forkUser:
    sleep(1.second)
    println("Hello!")

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

--------------------------------

### Schedule: Fibonacci Backoff

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/retries.md

Creates a schedule where the delay between retries follows the Fibonacci sequence.

```scala
Schedule.fibonacciBackoff(initialDelay: FiniteDuration)
```

--------------------------------

### Configure PropagatingVirtualThreadFactory in OxApp

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/integrations/otel-context.md

Set the PropagatingVirtualThreadFactory in your OxApp settings to enable context propagation across virtual threads.

```scala
import ox.*
import ox.otel.context.PropagatingVirtualThreadFactory

object MyApp extends OxApp:
  override def settings: OxApp.Settings = OxApp.Settings.Default.copy(
    threadFactory = Some(PropagatingVirtualThreadFactory())
  )
  
  def run(args: Vector[String])(using Ox): ExitCode = ExitCode.Success
```

--------------------------------

### Schedule: Exponential Backoff

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/retries.md

Creates a schedule where the delay between retries increases exponentially.

```scala
Schedule.exponentialBackoff(initialDelay: FiniteDuration)
```

--------------------------------

### Catching all exceptions in an either.catchAll block

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/basics/error-handling.md

Use an `either.catchAll` block to catch any non-fatal exceptions occurring within the block and represent them as `Throwable` in the resulting `Either`. Both `.ok()` and `.fail()` can be used inside.

```scala
import ox.either
import ox.either.ok

def doWork(): Either[Exception, Boolean] = ???

val result: Either[Throwable, String] = either.catchAll:
  if doWork().ok() then "ok" else throw new RuntimeException("not ok")
```

--------------------------------

### Specifying Buffer Capacity for Transformations

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/streaming/transforming-channels.md

Override the default buffer capacity (16) for transformation stages by providing a `BufferCapacity` given. This allows tuning the performance and memory usage of channel transformations.

```scala
(v: Source[Int]).map(_ + 1)(using BufferCapacity(10))
```

--------------------------------

### Repeat with Fixed Interval Schedule

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/repeat.md

Repeats an operation with a fixed 100ms interval between attempts, up to a maximum of 3 attempts.

```scala
repeat(Schedule.fixedInterval(100.millis).maxAttempts(3))(directOperation)
```

--------------------------------

### Schedule Customization: No Initial Delay

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/retries.md

Removes any initial delay for the first retry.

```scala
schedule.withNoInitialDelay()
```

--------------------------------

### Result Policy: Default

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/scheduling/retries.md

Uses the default result policy settings for success and retry conditions.

```scala
ResultPolicy.default[E, T]
```

--------------------------------

### Repeat a computation

Source: https://github.com/softwaremill/ox/blob/master/README.md

Repeats a computation according to a fixed interval schedule.

```scala
def computationR: Int = ???
repeat(Schedule.fixedInterval(100.millis))(computationR)
```

--------------------------------

### Error handling in structured concurrency

Source: https://github.com/softwaremill/ox/blob/master/README.md

Shows how exceptions in a supervised scope cause other forks to be interrupted.

```scala
supervised {
  forkUser:
    sleep(1.second)
    println("Hello!")

  forkUser:
    sleep(500.millis)
    throw new RuntimeException("boom!")
}
// on exception, all other forks are interrupted ("let it crash")
// the scope ends & re-throws only when all forks complete (no "leftovers")
```

--------------------------------

### Publish to Kafka as a mapping stage

Source: https://github.com/softwaremill/ox/blob/master/generated-doc/out/integrations/kafka.md

Use mapPublish to publish data to a Kafka topic while processing the resulting RecordMetadata within a flow.

```scala
import ox.flow.Flow
import ox.kafka.ProducerSettings
import ox.kafka.KafkaStage.*
import org.apache.kafka.clients.producer.{ProducerRecord, RecordMetadata}

val settings = ProducerSettings.default.bootstrapServers("localhost:9092")
val metadatas: Flow[RecordMetadata] = Flow
  .fromIterable(List("a", "b", "c"))
  .map(msg => ProducerRecord[String, String]("my_topic", msg))
  .mapPublish(settings)

// process & run the metadatas flow further
```
Tokens:
Raw