Underway

https://docs.rs/underway
Admin
Underway provides durable background jobs and step functions backed by PostgreSQL for deferred and...
Tokens:124,068
Snippets:920
Trust Score:9.1
Update:3 weeks ago
Show doc for...
Tokens:
Raw
### start method starts worker and scheduler

Source: https://docs.rs/underway/latest/underway/job/struct.Job.html

Starts both a worker and scheduler for the job and returns a JobHandle for graceful shutdown. The example shows calling start().await??.

```rust
let job = { /* A `Job`. */ };

job.start().await??;
```

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

### Scheduler::new usage example

Source: https://docs.rs/underway/latest/src/underway/scheduler.rs.html

Shows how to create a Scheduler with a queue and task. The example is marked no_run and includes hidden setup code for a PgPool and Queue. It demonstrates the Scheduler::new(queue.into(), task) call.

```rust
```rust,no_run
# use sqlx::{PgPool, Transaction, Postgres};
# use underway::{Task, task::Result as TaskResult, Queue};
use underway::Scheduler;

# struct ExampleTask;
# impl Task for ExampleTask {
#     type Input = ();
#     type Output = ();
#     async fn execute(
#         &self,
#         _: Transaction<'_, Postgres>,
#         _: Self::Input,
#     ) -> TaskResult<Self::Output> {
#         Ok(())
#     }
# }
# use tokio::runtime::Runtime;
# fn main() {
# let rt = Runtime::new().unwrap();
# rt.block_on(async {
# let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
# let queue = Queue::builder()
#    .name("example")
#    .pool(pool.clone())
#    .build()
#    .await?;
# /*
let queue = { /* A `Queue`. */ };
# */
# let task = ExampleTask;
# /*
let task = { /* An implementer of `Task`. */ };
# */
#

Scheduler::new(queue.into(), task);
# Ok::<(), Box<dyn std::error::Error>>(())
# });
# }
```
```

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

### Start a Job with job.start() (rust, no_run)

Source: https://docs.rs/underway/latest/src/underway/job.rs.html

Shown in the doc comment for `Job::start`, this example builds a `Job` with the `underway` crate and starts it with `job.start().await??`, which returns a `JobHandle`. The code is wrapped in `no_run` and uses hidden lines to set up a `PgPool`, a runtime, and a simple step that calls `To::done()`. The actual `Job` object is replaced by a placeholder comment in the displayed code.

```rust
/// # use sqlx::PgPool;
/// # use underway::{Job, To};
/// # use tokio::runtime::Runtime;
/// # fn main() {
/// # let rt = Runtime::new().unwrap();
/// # rt.block_on(async {
/// # let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
/// # let job = Job::<(), _>::builder()
/// #     .step(|_cx, _| async move { To::done() })
/// #     .name("example")
/// #     .pool(pool)
/// #     .build()
/// #     .await?;
/// # /*
/// let job = { /* A `Job`. */ };
/// # */
/// #

/// job.start().await??;
/// # Ok::<(), Box<dyn std::error::Error>>(())
/// # });
/// # }
```

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

### Scheduler::set_shutdown_token usage example

Source: https://docs.rs/underway/latest/src/underway/scheduler.rs.html

Shows how to set a shutdown token on a Scheduler using tokio_util::sync::CancellationToken. The example is marked no_run and includes hidden setup code for a PgPool, Queue, and Scheduler.

```rust
```rust,no_run
# use sqlx::{PgPool, Transaction, Postgres};
# use underway::{Task, task::Result as TaskResult, Queue, Scheduler};
use tokio_util::sync::CancellationToken;

# struct ExampleTask;
# impl Task for ExampleTask {
#     type Input = ();
#     type Output = ();
#     async fn execute(
#         &self,
#         _: Transaction<'_, Postgres>,
#         _: Self::Input,
#     ) -> TaskResult<Self::Output> {
#         Ok(())
#     }
# }
# use tokio::runtime::Runtime;
# fn main() {
# let rt = Runtime::new().unwrap();
# rt.block_on(async {
# let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
# let queue = Queue::builder()
#    .name("example")
#    .pool(pool.clone())
#    .build()
#    .await?;
# let task = ExampleTask;
# let mut scheduler = Scheduler::new(queue.into(), task);
# /*
```

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

### Rust example: Enqueue multiple tasks with enqueue_multi

Source: https://docs.rs/underway/latest/src/underway/queue.rs.html

Shows how to use enqueue_multi to enqueue multiple tasks with input after five minutes. Requires a PgPool and a Queue instance. The example is wrapped in a no_run Rust block with hidden setup code.

```rust
```rust,no_run
# use sqlx::{PgPool, Transaction, Postgres};
# use underway::{Task, task::Result as TaskResult};
# use underway::Queue;

# struct ExampleTask;
# impl Task for ExampleTask {
#     type Input = ();
#     type Output = ();
#     async fn execute(
#         &self,
#         _: Transaction<'_, Postgres>,
#         _: Self::Input,
#     ) -> TaskResult<Self::Output> {
#         Ok(())
#     }
# }
# use tokio::runtime::Runtime;
# fn main() {
# let rt = Runtime::new().unwrap();
# rt.block_on(async {
# let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
# /*
let pool = { /* A `PgPool`. */ };
# */
# let queue = Queue::builder()
#     .name("example")
#     .pool(pool.clone())
#     .build()
#     .await?;
# /*
let queue = { /* A `Queue`. */ };
# */
# let task = ExampleTask;
# /*
let task = { /* An implementer of `Task`. */ };
# */
#

// Enqueue a new task with input after five minutes.
let _ = queue.enqueue_multi(&pool, &task, &[(), ()]).await?;
# Ok::<(), Box<dyn std::error::Error>>(())
# });
# }
```
```

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

### Queue Builder Usage Example

Source: https://docs.rs/underway/latest/src/underway/queue.rs.html

Shows how to construct a Queue using the builder pattern with a PgPool, set the queue name, and build it. The example is part of the documentation for the build method.

```rust
/// #     ) -> TaskResult<Self::Output> {
/// #         Ok(())
/// #     }
/// # }
/// use sqlx::PgPool;
/// use underway::Queue;
///
/// # use tokio::runtime::Runtime;
/// # fn main() {
/// # let rt = Runtime::new().unwrap();
/// # rt.block_on(async {
/// let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
/// let queue_builder = Queue::<ExampleTask>::builder()
///     .name("example")
///     .pool(pool.clone())
///     .build()
///     .await?;
/// # Ok::<(), Box<dyn std::error::Error>>(())
/// # });
/// # }
```

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

### Usage Example for run_migrations

Source: https://docs.rs/underway/latest/underway/fn.run_migrations.html

Example showing how to run migrations with `underway::run_migrations`. It sets up a database connection pool using `sqlx::PgPool` and an environment variable `DATABASE_URL`, then calls the function.

```rust
use std::env;

use sqlx::PgPool;

// Set up the database connection pool.
let database_url = &env::var("DATABASE_URL")?;
let pool = PgPool::connect(database_url).await?;

// Run migrations.
underway::run_migrations(&pool).await?;
```

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

### Worker::with_concurrency_limit example

Source: https://docs.rs/underway/latest/src/underway/worker.rs.html

Shows how to create a worker and then set its concurrency limit. The example is truncated in the source; the `with_concurrency_limit` call is not shown.

```rust
use sqlx::{PgPool, Transaction, Postgres};
use underway::{Task, task::Result as TaskResult, Queue, Worker};
struct ExampleTask;
impl Task for ExampleTask {
    type Input = ();
    type Output = ();
    async fn execute(
        &self,
        _: Transaction<'_, Postgres>,
        _: Self::Input,
    ) -> TaskResult<Self::Output> {
        Ok(())
    }
}
use tokio::runtime::Runtime;
fn main() {
    let rt = Runtime::new().unwrap();
    rt.block_on(async {
        let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
        let queue = Queue::builder()
           .name("example")
           .pool(pool.clone())
           .build()
           .await?;
        let task = ExampleTask;
        let mut worker = Worker::new(queue.into(), task);
        /*
```

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

### Job::builder() example

Source: https://docs.rs/underway/latest/underway/job/struct.Job.html

Creates a new job builder. The example shows building a job with a step, name, and pool, then awaiting the build. Requires a PgPool.

```rust
use underway::{Job, To};

let pool = { /* A `PgPool`. */ };

let job = Job::<(), ()>::builder()
    .step(|_cx, _| async move { To::done() })
    .name("example")
    .pool(pool)
    .build()
    .await?;
```

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

### Run worker in a separate task with Worker::run

Source: https://docs.rs/underway/latest/src/underway/worker.rs.html

Example from the `run` method documentation. Shows how to spawn a worker in a separate Tokio task using `tokio::spawn`. The worker is created from a queue and a task, and the example includes hidden setup code for a PostgreSQL pool and queue.

```rust
/// #     }
/// # }
/// # use tokio::runtime::Runtime;
/// # fn main() {
/// # let rt = Runtime::new().unwrap();
/// # rt.block_on(async {
/// # let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
/// # let queue = Queue::builder()
/// #    .name("example")
/// #    .pool(pool)
/// #    .build()
/// #    .await?;
/// # let task = ExampleTask;
/// # let worker = Worker::new(queue.into(), task);
/// # /*
/// let worker = { /* A `Worker`. */ };
/// # */
/// #
///
/// // Run the worker in a separate task.
/// tokio::spawn(async move { worker.run().await });
/// # Ok::<(), Box<dyn std::error::Error>>(())
/// # });
/// # }
```

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

### Dequeue a task with Queue::dequeue

Source: https://docs.rs/underway/latest/src/underway/queue.rs.html

Shows how to enqueue a task and then dequeue it using the `Queue::dequeue` method. The example uses a `PgPool` and a custom `Task` implementation, and expects a pending task to be available. It is marked `no_run` and includes hidden setup code for compilation.

```rust
```rust,no_run
# use sqlx::{PgPool, Transaction, Postgres};
# use underway::{Task, task::Result as TaskResult};
# use underway::Queue;
# struct ExampleTask;
# impl Task for ExampleTask {
#     type Input = ();
#     type Output = ();
#     async fn execute(
#         &self,
#         _: Transaction<'_, Postgres>,
#         _: Self::Input,
#     ) -> TaskResult<Self::Output> {
#         Ok(())
#     }
# }
# use tokio::runtime::Runtime;
# fn main() {
# let rt = Runtime::new().unwrap();
# rt.block_on(async {
# let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
# /*
let pool = { /* A `PgPool`. */ };
# */
# let queue = Queue::builder()
#     .name("example")
#     .pool(pool.clone())
#     .build()
#     .await?;
# /*
let queue = { /* A `Queue`. */ };
# */
# let task = ExampleTask;
# /*
let task = { /* An implementer of `Task`. */ };
# */
#

// Enqueue a new task.
queue.enqueue(&pool, &task, &()).await?;

// Dequeue the enqueued task.
let mut tx = pool.begin().await?;
let pending_task = queue
    .dequeue()
    .await?
    .expect("There should be a pending task.");
# Ok::<(), Box<dyn std::error::Error>>(())
# });
# }
```
```

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

### Job::enqueue() example

Source: https://docs.rs/underway/latest/underway/job/struct.Job.html

Enqueues a job using a connection from the queue's pool. The example shows enqueueing a job with input ().

```rust
let job = { /* A `Job`. */ };

// Enqueue a new job with the given input.
job.enqueue(&()).await?;
```

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

### RetryPolicy builder initial_interval_ms example

Source: https://docs.rs/underway/latest/src/underway/task/retry_policy.rs.html

Shows how to set the initial interval before the first retry in milliseconds. The example sets initial_interval_ms to 2_500.

```rust
use underway::task::RetryPolicy;

// Set an initial interval of two and a half seconds.
let retry_policy_builder = RetryPolicy::builder().initial_interval_ms(2_500);
```

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

### Run worker with custom polling interval using Worker::run_every

Source: https://docs.rs/underway/latest/src/underway/worker.rs.html

Example from the `run_every` method documentation. Shows how to spawn a worker in a separate task with a custom polling interval of ten seconds using `run_every(10.seconds())`. Includes hidden setup code for a PostgreSQL pool and queue.

```rust
/// # use sqlx::{PgPool, Transaction, Postgres};
/// # use underway::{Task, task::Result as TaskResult, Queue, Worker};
/// use jiff::ToSpan;
///
/// # struct ExampleTask;
/// # impl Task for ExampleTask {
/// #     type Input = ();
/// #     type Output = ();
/// #     async fn execute(
/// #         &self,
/// #         _: Transaction<'_, Postgres>,
/// #         _: Self::Input,
/// #     ) -> TaskResult<Self::Output> {
/// #         Ok(())
/// #     }
/// # }
/// # use tokio::runtime::Runtime;
/// # fn main() {
/// # let rt = Runtime::new().unwrap();
/// # rt.block_on(async {
/// # let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
/// # let queue = Queue::builder()
/// #    .name("example")
/// #    .pool(pool)
/// #    .build()
/// #    .await?;
/// # let task = ExampleTask;
/// # let worker = Worker::new(queue.into(), task);
/// # /*
/// let worker = { /* A `Worker`. */ };
/// # */
/// #
///
/// // Increase the polling interval to every ten seconds.
/// tokio::spawn(async move { worker.run_every(10.seconds()).await });
/// # Ok::<(), Box<dyn std::error::Error>>(())
/// # });
/// # }
```

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

### Job::schedule() example

Source: https://docs.rs/underway/latest/underway/job/struct.Job.html

Schedules a job using a connection from the queue's pool. The example parses a cron expression with a timezone and schedules the job.

```rust
let job = { /* A `Job`. */ };

let every_minute = "0 * * * *[America/Los_Angeles]".parse()?;
job.schedule(&every_minute, &()).await?;
```

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

### process_next_task documentation and example

Source: https://docs.rs/underway/latest/src/underway/worker.rs.html

Documentation for the process_next_task method, including its behavior on timeout, retry policy, and error conditions. The example shows a minimal Task implementation with an execute method that takes a transaction and input, returning a TaskResult.

```rust
    /// Processes the next available task in the queue.
    ///
    /// When a task is found, its execute method will be invoked with the
    /// dequeued input.
    ///
    /// If the execution exceeds the task's timeout then the execute future is
    /// cancelled. This can also result in the task being marked failed if its
    /// retry policy has been exhausted.
    ///
    /// Tasks may also fail for other reasons during execution. Unless a failure
    /// is [explicitly fatal](crate::task::Error::Fatal) or no more retries are
    /// left, then the task will be re-queued in a
    /// [`Pending`](crate::task::State::Pending) state.
    ///
    /// # Errors
    ///
    /// This function will return an error if a new transaction cannot be
    /// acquired from the queue pool.
    ///
    /// It also has the same error conditions as [`Queue::dequeue`] as this is
    /// used internally.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// # use sqlx::{PgPool, Transaction, Postgres};
    /// # use underway::{Task, task::Result as TaskResult, Queue, Worker};
    /// # struct ExampleTask;
    /// # impl Task for ExampleTask {
    /// #     type Input = ();
    /// #     type Output = ();
    /// #     async fn execute(
    /// #         &self,
    /// #         _: Transaction<'_, Postgres>,
    /// #         _: Self::Input,
    /// #     ) -> TaskResult<Self::Output> {
```

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

### RetryPolicy builder build example

Source: https://docs.rs/underway/latest/src/underway/task/retry_policy.rs.html

Shows how to build a custom retry policy with max_attempts and backoff_coefficient. The example sets max_attempts to 20 and backoff_coefficient to 3.0.

```rust
use underway::task::RetryPolicy;

// Build a custom retry policy.
let retry_policy = RetryPolicy::builder()
    .max_attempts(20)
    .backoff_coefficient(3.0)
    .build();
```

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

### Job::schedule_using() example

Source: https://docs.rs/underway/latest/underway/job/struct.Job.html

Schedules a job using a provided executor, allowing the same transaction. The example uses a transaction from a PgPool and a weekly schedule with a timezone.

```rust
let pool = { /* A `PgPool`. */ };

// Our own transaction.
let mut tx = pool.acquire().await?;

let job = { /* A `Job`. */ };

// Schedule weekly using the transaction we already have.
let weekly = "@weekly[America/Los_Angeles]".parse()?;
job.schedule_using(&mut *tx, &weekly, &()).await?;
```

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

### Order receipts job example

Source: https://docs.rs/underway/latest/src/underway/lib.rs.html

Defines a multi-step job that generates a PDF receipt and then emails it. The example is incomplete in the source; it shows the input structs for the steps but not the full job definition.

```rust
use std::env;

use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use underway::{Job, To};

#[derive(Deserialize, Serialize)]
struct GenerateReceipt {
    // An order we want to generate a receipt for.
    order_id: i32,
}

#[derive(Deserialize, Serialize)]
struct EmailReceipt {
    // An object store key to our receipt PDF.
    receipt_key: String,
}
```

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

### RetryPolicy builder backoff_coefficient example

Source: https://docs.rs/underway/latest/src/underway/task/retry_policy.rs.html

Shows how to set the backoff coefficient to control delay growth. The example sets backoff_coefficient to 1.5.

```rust
use underway::task::RetryPolicy;

// Set a backoff coefficient of one point five.
let retry_policy_builder = RetryPolicy::builder().backoff_coefficient(1.5);
```

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

### is_err_and example

Source: https://docs.rs/underway/latest/underway/task/type.Result.html

Shows how to use is_err_and to check if a Result is Err and the error value matches a predicate. Includes examples with different error kinds and ownership.

```rust
use std::io::{Error, ErrorKind};

let x: Result<u32, Error> = Err(Error::new(ErrorKind::NotFound, "!"));
assert_eq!(x.is_err_and(|x| x.kind() == ErrorKind::NotFound), true);

let x: Result<u32, Error> = Err(Error::new(ErrorKind::PermissionDenied, "!"));
assert_eq!(x.is_err_and(|x| x.kind() == ErrorKind::NotFound), false);

let x: Result<u32, Error> = Ok(123);
assert_eq!(x.is_err_and(|x| x.kind() == ErrorKind::NotFound), false);

let x: Result<u32, String> = Err("ownership".to_string());
assert_eq!(x.as_ref().is_err_and(|x| x.len() > 1), true);
println!("still alive {:?}", x);
```

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

### inspect example

Source: https://docs.rs/underway/latest/underway/task/type.Result.html

Calls a function with a reference to the contained value if `Ok`, and returns the original result. The example parses a string, inspects the value, and then maps it to a cube.

```rust
let x: u8 = "4"
    .parse::<u8>()
    .inspect(|x| println!("original: {x}"))
    .map(|x| x.pow(3))
    .expect("literal `4` should parse as a `u8`");
```

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

### RetryPolicy builder example with max_attempts and backoff_coefficient

Source: https://docs.rs/underway/latest/src/underway/task/retry_policy.rs.html

Shows how to create a RetryPolicy with a custom max_attempts and backoff_coefficient using the builder. The example sets max_attempts to 10 and backoff_coefficient to 4.0.

```rust
use underway::task::RetryPolicy;

let retry_policy = RetryPolicy::builder()
    .max_attempts(10)
    .backoff_coefficient(4.0)
    .build();
```

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

### iter_mut example

Source: https://docs.rs/underway/latest/underway/task/type.Result.html

Returns a mutable iterator over the possibly contained value. The example shows modifying the contained value when `Ok` and getting `None` when `Err`.

```rust
let mut x: Result<u32, &str> = Ok(7);
match x.iter_mut().next() {
    Some(v) => *v = 40,
    None => {},
}
assert_eq!(x, Ok(40));

let mut x: Result<u32, &str> = Err("nothing!");
assert_eq!(x.iter_mut().next(), None);
```

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

### as_bytes example

Source: https://docs.rs/underway/latest/underway/job/struct.JobId.html

Borrows the underlying 16-byte array of the UUID. The example creates a UUID from a byte array, gets its bytes, and verifies that the borrowed bytes point to the original array.

```rust
let bytes1 = [
    0xa1, 0xa2, 0xa3, 0xa4,
    0xb1, 0xb2,
    0xc1, 0xc2,
    0xd1, 0xd2, 0xd3, 0xd4, 0xd5, 0xd6, 0xd7, 0xd8,
];
let uuid1 = Uuid::from_bytes_ref(&bytes1);

let bytes2 = uuid1.as_bytes();
let uuid2 = Uuid::from_bytes_ref(bytes2);

assert_eq!(uuid1, uuid2);

assert!(std::ptr::eq(
    uuid2 as *const Uuid as *const u8,
    &bytes1 as *const [u8; 16] as *const u8,
));
```

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

### Build a job with queue

Source: https://docs.rs/underway/latest/src/underway/job.rs.html

Shows how to finalize a job builder into a Job. The example creates a queue with name "example" and a PgPool, adds a step that returns To::done(), assigns the queue, and calls build() to produce the Job. This is part of the build method documentation.

```rust
/// # use sqlx::PgPool;
/// use underway::{Job, To, Queue};
///
/// # use tokio::runtime::Runtime;
/// # fn main() {
/// # let rt = Runtime::new().unwrap();
/// # rt.block_on(async {
/// # let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
/// # /*
/// let pool = { /* A `PgPool`. */ };
/// # */
/// #
/// let queue = Queue::builder().name("example").pool(pool).build().await?;
///
/// // Build the job.
/// let job = Job::<(), ()>::builder()
///     .step(|_cx, _| async move { To::done() })
///     .queue(queue)
///     .build();
/// # Ok::<(), Box<dyn std::error::Error>>(())
/// # });
/// # }
```

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

### Get UUID fields in little-endian order

Source: https://docs.rs/underway/latest/underway/job/struct.JobId.html

Shows how to use to_fields_le to get the four field values of a UUID in little-endian order. The example parses a UUID and asserts the expected little-endian fields.

```rust
use uuid::Uuid;

let uuid = Uuid::parse_str("a1a2a3a4-b1b2-c1c2-d1d2-d3d4d5d6d7d8")?;

assert_eq!(
    uuid.to_fields_le(),
    (
        0xa4a3a2a1,
        0xb2b1,
        0xc2c1,
        &[0xd1, 0xd2, 0xd3, 0xd4, 0xd5, 0xd6, 0xd7, 0xd8],
    )
);
```

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

### Set queue on job builder

Source: https://docs.rs/underway/latest/src/underway/job.rs.html

Shows how to set a queue on a job builder. The example creates a queue with name "example" and a PgPool, then builds a job with a single step that returns To::done() and assigns the queue. This is part of the Builder's queue method documentation.

```rust
/// # use sqlx::PgPool;
/// use underway::{Job, Queue, To};
///
/// # use tokio::runtime::Runtime;
/// # fn main() {
/// # let rt = Runtime::new().unwrap();
/// # rt.block_on(async {
/// # let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
/// # /*
/// let pool = { /* A `PgPool`. */ };
/// # */
/// #
/// let queue = Queue::builder().name("example").pool(pool).build().await?;
///
/// // Set a queue.
/// let job = Job::<(), ()>::builder()
///     .step(|_cx, _| async move { To::done() })
///     .queue(queue);
/// # Ok::<(), Box<dyn std::error::Error>>(())
/// # });
/// # }
```

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

### Run a scheduler with job.scheduler().run()

Source: https://docs.rs/underway/latest/src/underway/job.rs.html

Shows how to create and run a scheduler for a job using job.scheduler().run(). Includes imports and setup in comments.

```rust
/// # use sqlx::PgPool;
/// # use underway::{Job, To};
/// # use tokio::runtime::Runtime;
/// # fn main() {
/// # let rt = Runtime::new().unwrap();
/// # rt.block_on(async {
/// # let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
/// # let job = Job::<(), _>::builder()
/// #     .step(|_cx, _| async move { To::done() })
/// #     .name("example")
/// #     .pool(pool)
/// #     .build()
/// #     .await?;
/// # /*
/// let job = { /* A `Job`. */ };
/// # */
/// #

let job_scheduler = job.scheduler();
job_scheduler.run().await?; // Run the scheduler directly.
/// # Ok::<(), Box<dyn std::error::Error>>(())
/// # });
/// # }
```

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

### Get UUID version

Source: https://docs.rs/underway/latest/underway/job/struct.JobId.html

Parses a UUID and retrieves its version as an Option<Version>. The example expects the MD5 version.

```rust
let my_uuid = Uuid::parse_str("02f09a3f-1624-3b1d-8409-44eff7708208")?;

assert_eq!(Some(Version::Md5), my_uuid.get_version());
```

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

### Create a new builder with Queue::builder()

Source: https://docs.rs/underway/latest/underway/queue/struct.Builder.html

Example for `Builder::new()`, showing typical usage through `Queue::<ExampleTask>::builder()`.

```rust
use underway::Queue;

let queue_builder = Queue::<ExampleTask>::builder();
```

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

### Get UUID version number

Source: https://docs.rs/underway/latest/underway/job/struct.JobId.html

Parses a UUID and retrieves its version number as a usize. The example expects version 3.

```rust
let my_uuid = Uuid::parse_str("02f09a3f-1624-3b1d-8409-44eff7708208")?;

assert_eq!(3, my_uuid.get_version_num());
```

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

### Schedule a job with a cron expression

Source: https://docs.rs/underway/latest/underway/job/index.html

Schedules a job with a cron-like expression. The example sets a weekly schedule with a timezone and then starts the job.

```rust
use underway::{Job, To};

let pool = { /* A `PgPool`. */ };

let job = Job::builder()
    .step(|_cx, _: ()| async move { To::done() })
    .name("scheduled-job")
    .pool(pool)
    .build()
    .await?;

// Sets a weekly schedule with the given input.
let weekly = "@weekly[America/Los_Angeles]".parse()?;
job.schedule(&weekly, &()).await?;

job.start();
```

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

### Create a queue from a Job using job.queue()

Source: https://docs.rs/underway/latest/src/underway/job.rs.html

Shows how to obtain a queue reference from a job. Comments show required imports and setup.

```rust
/// # let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
/// # let job = Job::<(), _>::builder()
/// #     .step(|_cx, _| async move { To::done() })
/// #     .name("example")
/// #     .pool(pool)
/// #     .build()
/// #     .await?;
/// # /*
/// let job = { /* A `Job`. */ };
/// # */
/// #

let job_queue = job.queue();
/// # Ok::<(), Box<dyn std::error::Error>>(())
/// # });
/// # }
```

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

### Get UUID variant

Source: https://docs.rs/underway/latest/underway/job/struct.JobId.html

Shows how to parse a UUID string and check its variant. The example expects the parsed UUID to have the RFC4122 variant.

```rust
let my_uuid = Uuid::parse_str("02f09a3f-1624-3b1d-8409-44eff7708208")?;

assert_eq!(Variant::RFC4122, my_uuid.get_variant());
```

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

### Build a job with a preconstructed queue and enqueue

Source: https://docs.rs/underway/latest/src/underway/job.rs.html

Shows how to create a queue first and then reuse it for building a job, avoiding an await on job build. Requires a PgPool and DATABASE_URL.

```rust
# use sqlx::PgPool;
use underway::{Job, Queue, To};

# use tokio::runtime::Runtime;
# fn main() {
# let rt = Runtime::new().unwrap();
# rt.block_on(async {
# let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
# /*
let pool = { /* A `PgPool`. */ };
# */
#

// We've defined a queue directly.
let queue = Queue::builder()
    .name("example-job")
    .pool(pool)
    .build()
    .await?;

let job = Job::builder()
    .step(|_cx, _| async move { To::done() })
    .queue(queue)
    .build();

job.enqueue(&()).await?;
# Ok::<(), Box<dyn std::error::Error>>(())
# });
# }
```

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

### ok example

Source: https://docs.rs/underway/latest/underway/task/type.Result.html

Demonstrates converting a Result to an Option, returning Some(value) for Ok and None for Err.

```rust
let x: Result<u32, &str> = Ok(2);
assert_eq!(x.ok(), Some(2));

let x: Result<u32, &str> = Err("Nothing here");
assert_eq!(x.ok(), None);
```

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

### JobHandle shutdown example

Source: https://docs.rs/underway/latest/underway/job/struct.JobHandle.html

Shows how to start a job and gracefully shut down its worker and scheduler using the shutdown method. The job variable is a placeholder for a Job instance.

```rust
let job = { /* A `Job`. */ };

let job_handle = job.start();

// Gracefully stop worker and scheduler.
job_handle.shutdown().await?;
```

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

### Run a worker with job.worker().run()

Source: https://docs.rs/underway/latest/src/underway/job.rs.html

Shows how to create and run a worker for a job using job.worker().run(). Includes imports and setup in comments.

```rust
/// # use sqlx::PgPool;
/// # use underway::{Job, To};
/// # use tokio::runtime::Runtime;
/// # fn main() {
/// # let rt = Runtime::new().unwrap();
/// # rt.block_on(async {
/// # let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
/// # let job = Job::<(), _>::builder()
/// #     .step(|_cx, _| async move { To::done() })
/// #     .name("example")
/// #     .pool(pool)
/// #     .build()
/// #     .await?;
/// # /*
/// let job = { /* A `Job`. */ };
/// # */
/// #

let job_worker = job.worker();
job_worker.run().await?; // Run the worker directly.
/// # Ok::<(), Box<dyn std::error::Error>>(())
/// # });
/// # }
```

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

### Test: Reschedule task for retry (partial)

Source: https://docs.rs/underway/latest/src/underway/queue.rs.html

Starts a test for rescheduling a task for retry by building a queue and enqueueing a task. The snippet is incomplete as it cuts off mid-test, but shows the initial setup.

```rust
    #[sqlx::test]
    async fn reschedule_task_for_retry(pool: PgPool) -> sqlx::Result<(), Error> {
        let queue = Queue::builder()
            .name("test_reschedule")
            .pool(pool.clone())
            .build()
            .await?;

        let input = serde_json::json!({ "key": "value" });
        let task = TestTask;

        // Enqueue a task
        let task_id = queue.enqueue(&pool, &task, &input).await?;
```

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

### Set database pool with .pool()

Source: https://docs.rs/underway/latest/underway/queue/struct.Builder.html

Example showing the `pool` method, which accepts a `PgPool` and transitions the builder to the `PoolSet` state. It uses `PgPool::connect` with an async `await` and `?` operator.

```rust
use sqlx::PgPool;
use underway::Queue;

let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
let queue_builder = Queue::<ExampleTask>::builder()
    .name("example")
    .pool(pool.clone());
```

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

### Schedule a task with queue.schedule()

Source: https://docs.rs/underway/latest/underway/queue/struct.Queue.html

Shows how to create a schedule for the queue using the `schedule` method. The example parses a crontab-like schedule string and passes it along with the input. After scheduling, a scheduler instance must be run for the schedule to fire.

```rust
let pool = { /* A `PgPool`. */ };
let queue = { /* A `Queue`. */ };

// Set a schedule on the queue with the given input.
let daily = "@daily[America/Los_Angeles]".parse()?;
queue.schedule(&pool, &daily, &()).await?;
```

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

### Build the queue with .build()

Source: https://docs.rs/underway/latest/underway/queue/struct.Builder.html

Example showing the async `build` method on the `PoolSet` builder state. It returns `Result<Queue<T>, Error>` and is awaited with `?`.

```rust
use sqlx::PgPool;
use underway::Queue;

let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
let queue_builder = Queue::<ExampleTask>::builder()
    .name("example")
    .pool(pool.clone())
    .build()
    .await?;
```

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

### Worker::run_every example

Source: https://docs.rs/underway/latest/underway/worker/struct.Worker.html

Runs the worker with a configurable delay between polls. This example sets the polling interval to ten seconds.

```rust
use jiff::ToSpan;

let worker = { /* A `Worker`. */ };

// Increase the polling interval to every ten seconds.
tokio::spawn(async move { worker.run_every(10.seconds()).await });
```

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

### as_u128 example

Source: https://docs.rs/underway/latest/underway/job/struct.JobId.html

Packs the UUID bytes directly into a u128. The example parses a UUID and asserts the resulting u128 value.

```rust
let uuid = Uuid::parse_str("a1a2a3a4-b1b2-c1c2-d1d2-d3d4d5d6d7d8")?;

assert_eq!(
    uuid.as_u128(),
    0xa1a2a3a4b1b2c1c2d1d2d3d4d5d6d7d8,
);
```

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

### RetryPolicy builder max_interval_ms example

Source: https://docs.rs/underway/latest/src/underway/task/retry_policy.rs.html

Shows how to set the maximum interval between retries in milliseconds. The example sets max_interval_ms to 300_000.

```rust
use underway::task::RetryPolicy;

// Set a maximum interval of five minutes.
let retry_policy_builder = RetryPolicy::builder().max_interval_ms(300_000);
```

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

### Queue::builder() example

Source: https://docs.rs/underway/latest/underway/queue/struct.Queue.html

Creates a new Queue using the builder pattern. Requires a PgPool and a task type that implements Task. The builder sets the queue name, dead letter queue name, and pool before building.

```rust
use underway::Queue;

let pool = { /* A `PgPool`. */ };
let queue: Queue<ExampleTask> = Queue::builder()
    .name("example")
    .dead_letter_queue("example_dlq")
    .pool(pool)
    .build()
    .await?;
```

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

### Send welcome email job with underway

Source: https://docs.rs/underway/latest/index.html

Defines a WelcomeEmail input struct and a job with a single step that simulates sending an email. The job is built with a name and pool, enqueued with sample data, and started. Requires DATABASE_URL environment variable and running migrations.

```rust
use std::env;

use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use underway::{Job, To};

// This is the input we'll provide to the job when we enqueue it.
#[derive(Deserialize, Serialize)]
struct WelcomeEmail {
    user_id: i32,
    email: String,
    name: String,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Set up the database connection pool.
    let database_url = &env::var("DATABASE_URL").expect("DATABASE_URL should be set");
    let pool = PgPool::connect(database_url).await?;

    // Run migrations.
    underway::run_migrations(&pool).await?;

    // Build the job.
    let job = Job::builder()
        .step(
            |_cx,
             WelcomeEmail {
                 user_id,
                 email,
                 name,
             }| async move {
                // Simulate sending an email.
                println!("Sending welcome email to {name} <{email}> (user_id: {user_id})");
                // Returning this indicates this is the final step.
                To::done()
            },
        )
        .name("welcome-email")
        .pool(pool)
        .build()
        .await?;

    // Here we enqueue a new job to be processed later.
    job.enqueue(&WelcomeEmail {
        user_id: 42,
        email: "ferris@example.com".to_string(),
        name: "Ferris".to_string(),
    })
    .await?;

    // Start processing enqueued jobs.
    job.start().await??;

    Ok(())
}
```

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

### RetryPolicy builder example with backoff_coefficient only

Source: https://docs.rs/underway/latest/src/underway/task/retry_policy.rs.html

Shows how to create a RetryPolicy with a custom backoff_coefficient using the builder. The example sets backoff_coefficient to 3.0.

```rust
use underway::task::RetryPolicy;

let retry_policy = RetryPolicy::builder().backoff_coefficient(3.0).build();
```

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

### as_ref example

Source: https://docs.rs/underway/latest/underway/task/type.Result.html

Shows converting a Result to a Result of references, leaving the original intact.

```rust
let x: Result<u32, &str> = Ok(2);
assert_eq!(x.as_ref(), Ok(&2));

let x: Result<u32, &str> = Err("Error");
assert_eq!(x.as_ref(), Err(&"Error"));
```

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

### unwrap_or_else method example

Source: https://docs.rs/underway/latest/underway/task/type.Result.html

Returns the contained Ok value or computes it from a closure. The example defines a function count that returns the length of a string.

```rust
fn count(x: &str) -> usize { x.len() }
```

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

### Scheduler::new example

Source: https://docs.rs/underway/latest/underway/struct.Scheduler.html

Creates a new Scheduler with a queue and a task. The queue is converted into an Arc<Queue<T>> via .into().

```rust
use underway::Scheduler;

let queue = { /* A `Queue`. */ };
let task = { /* An implementer of `Task`. */ };

Scheduler::new(queue.into(), task);
```

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

### iter example

Source: https://docs.rs/underway/latest/underway/task/type.Result.html

Returns an iterator over the possibly contained value. The example shows that the iterator yields one value for `Ok` and none for `Err`.

```rust
let x: Result<u32, &str> = Ok(7);
assert_eq!(x.iter().next(), Some(&7));

let x: Result<u32, &str> = Err("nothing!");
assert_eq!(x.iter().next(), None);
```

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

### to_u128_le example

Source: https://docs.rs/underway/latest/underway/job/struct.JobId.html

Returns a 128-bit little-endian value. The entire UUID is reversed, not just the fields. The example shows the resulting value for a parsed UUID.

```rust
let uuid = Uuid::parse_str("a1a2a3a4-b1b2-c1c2-d1d2-d3d4d5d6d7d8")?;

assert_eq!(
    uuid.to_u128_le(),
    0xd8d7d6d5d4d3d2d1c2c1b2b1a4a3a2a1,
);
```

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

### Set queue name with .name()

Source: https://docs.rs/underway/latest/underway/queue/struct.Builder.html

Example showing the `name` method, which accepts an `impl Into<String>` and transitions the builder to the `NameSet` state.

```rust
use underway::Queue;

let queue_builder = Queue::<ExampleTask>::builder().name("my-queue");
```

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

### Define a job with a closure step

Source: https://docs.rs/underway/latest/underway/job/index.html

Shows the minimal setup for a job with a single step using a closure that returns `To::done()`.

```rust
use underway::{Job, To};

let job_builder = Job::<(), ()>::builder().step(|_cx, _| async move { To::done() });
```
Tokens:
Raw