### Start Supervised Oban Instance Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Testing Provides API documentation for `start_supervised_oban!`, a function to start a supervised Oban instance under the test supervisor. It accepts all valid Oban options and registers the instance with a unique reference, preventing conflicts. ```APIDOC start_supervised_oban!(opts :: [Oban.option()]) :: Registry.key() (since 0.11.0) Starts a supervised Oban instance under the test supervisor, registering it with a unique reference. This prevents conflicts with other Oban instances or tests. It automatically adds sandbox allowances for plugins and queue producers, and the test process waits for clean shutdown. **Running Jobs:** By default, the supervised instance does not process jobs (`stage_interval` is `:infinity`). Set `stage_interval` to a low value to process jobs normally. **Options:** Accepts any option accepted by `Oban.start_link/1`, including shared repo-specific options. **Examples:** Start a basic supervised instance: ```elixir name = start_supervised_oban!(repo: MyApp.Repo) Oban.insert(name, MyWorker.new()) ``` Start with a single queue and polling every 10ms: ```elixir start_supervised_oban!(repo: MyApp.Repo, stage_interval: 10, queues: [alpha: 10]) ``` ``` -------------------------------- ### Fetch Dependencies Source: https://getoban.pro/docs/pro/1.6.2/installation Fetches all project dependencies, including the newly added Oban Pro package, after updating `mix.exs`. ```elixir $ mix deps.get ``` -------------------------------- ### Add Oban Hex Repository Source: https://getoban.pro/docs/pro/1.6.2/installation Adds the private Oban Pro hex repository to your system. Requires your OBAN_LICENSE_KEY for authentication. ```elixir mix hex.repo add oban https://repo.oban.pro \ --fetch-public-key SHA256:4/OSKi0NRF91QVVXlGAhb/BIMLnK8NHcx/EWs+aIWPc \ --auth-key $OBAN_LICENSE_KEY ``` -------------------------------- ### Oban Pro Testing Setup Source: https://getoban.pro/docs/pro/1.6.2/adoption Demonstrates how to switch from Oban.Testing to Oban.Pro.Testing in an ExUnit test case. This change is necessary to leverage the Pro version's enhanced features for testing. ```Elixir defmodule MyApp.Case do use ExUnit.CaseTemplate using do quote do - use Oban.Testing, repo: MyApp.Repo + use Oban.Pro.Testing, repo: MyApp.Repo end end end ``` -------------------------------- ### Delegate Oban Pro Migrations Source: https://getoban.pro/docs/pro/1.6.2/installation Configures the generated migration file to delegate the up and down functions to Oban.Pro.Migration for proper database schema management. ```elixir defmodule MyApp.Repo.Migrations.AddObanPro do use Ecto.Migration def up, do: Oban.Pro.Migration.up() def down, do: Oban.Pro.Migration.down() end ``` -------------------------------- ### Add Oban Pro Dependency Source: https://getoban.pro/docs/pro/1.6.2/installation Specifies Oban Pro as a dependency in your application's `mix.exs` file, pointing to the added 'oban' repository. ```elixir {:oban_pro, "~> 1.6", repo: "oban"} ``` -------------------------------- ### Generate Oban Pro Migration Source: https://getoban.pro/docs/pro/1.6.2/installation Generates a new Ecto migration file specifically for setting up Oban Pro's database tables and indexes. ```elixir mix ecto.gen.migration add_oban_pro ``` -------------------------------- ### Oban Pro Relay Usage Examples Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Relay Demonstrates how to insert jobs for asynchronous execution using `async/1` and then await their results with `await/1` or `await_many/1`. ```elixir relay = %{id: 123} |> MyApp.Worker.new() |> Oban.Pro.Relay.async() {:ok, result} = Oban.Pro.Relay.await(relay) {:ok, result} = Oban.Pro.Relay.await(relay, :timer.seconds(30)) relayed = 1..3 |> Enum.map(&DoubleWorker.new(%{int: &1})) |> Enum.map(&Oban.Pro.Relay.async/1) |> Oban.Pro.Relay.await_many() #=> [{:ok, 2}, {:ok, 4}, {:ok, 6}] ``` -------------------------------- ### Setup Oban Pro Testing Helpers (Individual Test) Source: https://getoban.pro/docs/pro/1.6.2/testing Imports `Oban.Pro.Testing` directly into an individual test module. This approach is useful if you prefer not to include helpers in every test case. ```Elixir defmodule MyApp.WorkerTest do use MyApp.Case, async: true use Oban.Pro.Testing, repo: MyApp.Repo end ``` -------------------------------- ### Run Database Migrations Source: https://getoban.pro/docs/pro/1.6.2/installation Executes all pending database migrations, including the one generated for Oban Pro, to apply schema changes. ```elixir mix ecto.migrate ``` -------------------------------- ### DynamicScaler Configuration Examples Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicScaler Demonstrates how to configure the DynamicScaler plugin with custom cloud modules, either as a tuple with options or by referencing the module directly if it pulls configuration from the environment. ```elixir {DynamicScaler, scalers: [range: 1..3, cloud: {MyApp.ASG, asg: "my-asg-name"}]} ``` ```elixir {DynamicScaler, scalers: [range: 1..3, cloud: MyApp.ASG]} ``` -------------------------------- ### Setup Oban Pro Testing Helpers (Module Use) Source: https://getoban.pro/docs/pro/1.6.2/testing Imports `Oban.Pro.Testing` into a custom test case module, providing access to advanced testing functions. This is a convenient way to include helpers in all tests within a module. ```Elixir defmodule MyApp.Case do use ExUnit.CaseTemplate using do quote do use Oban.Pro.Testing, repo: MyApp.Repo end end end ``` -------------------------------- ### Add DynamicPrioritizer Plugin Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicPrioritizer Demonstrates how to add the DynamicPrioritizer plugin to your Oban configuration in `config.exs`. This is the basic setup to enable the plugin's functionality. ```elixir config :my_app, Oban, plugins: [Oban.Pro.Plugins.DynamicPrioritizer] ... ``` -------------------------------- ### Dynamic Queue Configuration Examples Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicQueues Illustrates how queue definitions in Oban's configuration affect runtime persistence. Options not explicitly defined in the configuration will be retained when queues are updated at runtime. ```elixir queues: [ default: [limit: 10], ... ] ``` ```elixir # pause, global_limit, etc. will persist, but limit won't default: [limit: 10] ``` ```elixir # pause will persist, but limit and global_limit won't default: [limit: 10, global_limit: 20] ``` ```elixir # neither limits nor pausing will persist default: [limit: 10, global_limit: 20, paused: true] ``` -------------------------------- ### Initial DynamicCron Crontab Setup Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicCron Demonstrates how to provide an initial list of cron entries to the DynamicCron plugin using the `:crontab` option. This includes cron expressions, worker modules, and optional arguments like queue or max_attempts. ```elixir plugins: [{ Oban.Pro.Plugins.DynamicCron, crontab: [ {"* * * * *", MyApp.MinuteJob}, {"0 * * * *", MyApp.HourlyJob, queue: :scheduled}, {"0 0 * * *", MyApp.DailyJob, max_attempts: 1}, {"0 12 * * MON", MyApp.MondayWorker, tags: ["scheduled"]}, {"@daily", MyApp.AnotherDailyWorker} ] }] ``` -------------------------------- ### Oban Pro Workflow Basic Usage Example Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Workflow Demonstrates creating a workflow with sequential and parallel job execution using Oban Pro's Workflow behaviour. It shows how to add jobs with dependencies and insert the workflow into Oban. ```Elixir defmodule EchoWorkflow do use Oban.Pro.Decorator alias Oban.Pro.Workflow def insert do Workflow.new() |> Workflow.add(:a, new_echo(1)) |> Workflow.add(:b, new_echo(2), deps: :a) |> Workflow.add(:c, new_echo(3), deps: :b) |> Workflow.add(:d, new_echo(4), deps: :b) |> Workflow.add(:e, new_echo(5), deps: [:c, :d]) |> Oban.insert_all() end @job true def echo(value), do: IO.inspect(value) end ``` -------------------------------- ### Configure DynamicQueues Plugin Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicQueues Adds the DynamicQueues plugin to Oban's configuration and defines initial queues. This setup persists queue definitions to the database and supervises them across all connected nodes. ```Elixir config :my_app, Oban, plugins: [Oban.Pro.Plugins.DynamicQueues] ... ``` ```Elixir config :my_app, Oban, plugins: [{ Oban.Pro.Plugins.DynamicQueues, queues: [ default: 20, mailers: [global_limit: 20], events: [local_limit: 30, rate_limit: [allowed: 100, period: 60]] ] }] ... ``` -------------------------------- ### Oban Pro Workflow Sub-Workflows Example Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Workflow Demonstrates nesting workflows by adding one workflow as a dependency of another using `add_workflow/4`. This allows for modular composition of complex pipelines. ```Elixir alias MyApp.{WorkerA, WorkerB, WorkerC} alias Oban.Pro.Workflow extr_flow = Workflow.new(workflow_name: "extract") |> Workflow.add(:extract, WorkerA.new(%{source: "database"})) |> Workflow.add(:transform, WorkerB.new(%{type: "normalize"}), deps: :extract) note_flow = Workflow.new(workflow_name: "notify") |> Workflow.add(:prepare, WorkerA.new(%{template: "report"})) |> Workflow.add(:send, WorkerB.new(%{method: "email"}), deps: :prepare) # Example of adding a sub-workflow (assuming 'main_workflow' is defined elsewhere) # main_workflow = # Workflow.new(workflow_name: "main") # |> Workflow.add_workflow(:extract_stage, extr_flow, deps: :start_node) # |> Workflow.add_workflow(:notify_stage, note_flow, deps: :extract_stage) # |> Oban.insert_all() ``` -------------------------------- ### Oban Pro Batch Worker Example Source: https://getoban.pro/docs/pro/1.6.2/composition Demonstrates how to define an Oban Pro Batch worker in Elixir, including processing individual jobs and defining callbacks for batch completion or failure. ```elixir defmodule MyApp.ImageProcessor do use Oban.Pro.Worker, queue: :media @behaviour Oban.Pro.Batch alias MyApp.{Media, Notifications} @impl Oban.Pro.Worker def process(%Job{args: %{"image_id" => id, "operations" => operations}}) do with {:ok, image} <- Media.get_image(id), Media.process(image, operations) end end @impl Oban.Pro.Batch def batch_completed(%Job{meta: %{"batch_id" => batch_id, "account_id" => account_id}}) do with {:ok, images} <- Media.get_batch_images(batch_id), {:ok, archive_path} <- Media.create_archive(images), {:ok, url} <- Media.store_archive(archive_path, account_id) do Notifications.image_processing_complete(account_id, url) end end @impl Oban.Pro.Batch def batch_discarded(%Job{meta: %{"batch_id" => batch_id, "account_id" => account_id}}) do Notifications.processing_failed(account_id, batch_id) end end ``` -------------------------------- ### Implement Oban Pro Batch Handler Callbacks Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Batch Provides an example of implementing all optional batch handler callbacks: `batch_attempted/1`, `batch_cancelled/1`, `batch_completed/1`, `batch_discarded/1`, `batch_exhausted/1`, and `batch_retryable/1`. Each callback receives an Oban.Job struct and should return `:ok`. ```elixir defmodule MyApp.BatchWorker do use Oban.Pro.Worker @behaviour Oban.Pro.Batch @impl Oban.Pro.Worker def process(_job), do: :ok @impl Oban.Pro.Batch def batch_attempted(%Job{meta: %{"batch_id" => batch_id}}) do IO.inspect({:attempted, batch_id}) :ok end @impl Oban.Pro.Batch def batch_cancelled(%Job{meta: %{"batch_id" => batch_id}}) do IO.inspect({:cancelled, batch_id}) :ok end @impl Oban.Pro.Batch def batch_completed(%Job{meta: %{"batch_id" => batch_id}}) do IO.inspect({:completed, batch_id}) :ok end @impl Oban.Pro.Batch def batch_discarded(%Job{meta: %{"batch_id" => batch_id}}) do IO.inspect({:discarded, batch_id}) :ok end @impl Oban.Pro.Batch def batch_exhausted(%Job{meta: %{"batch_id" => batch_id}}) do IO.inspect({:exhausted, batch_id}) :ok end @impl Oban.Pro.Batch def batch_retryable(%Job{meta: %{"batch_id" => batch_id}}) do IO.inspect({:retryable, batch_id}) :ok end end ``` -------------------------------- ### Oban Pro Chains: Sequential Job Execution Source: https://getoban.pro/docs/pro/1.6.2/composition Chains ensure jobs for a specific resource execute in order, processing events exactly as they arrive or implementing sequential state transitions. This example shows how to configure chaining by account ID and a worker that processes payment webhooks sequentially. ```elixir use Oban.Pro.Worker, chain: [by: [:worker, args: :account_id]] defmodule MyApp.PaymentWebhookWorker do use Oban.Pro.Worker, queue: :webhooks, chain: [by: [args: :account_id]] alias MyApp.{Accounts, Payments, Transactions} @impl Oban.Pro.Worker def process(%Job{args: %{"account_id" => account_id, "event" => event, "data" => data}}) do with {:ok, account} <- Accounts.get(account_id) do case event do "payment.succeeded" -> Transactions.record_payment(account, data) "payment.refunded" -> Transactions.record_refund(account, data) "subscription.renewed" -> Payments.renew_subscription(account, data) end end end end ``` -------------------------------- ### Configure Oban Pro Chain Partitioning Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Worker Provides examples of configuring the `:by` option for Oban Pro chains to control how jobs are partitioned and sequenced. This allows fine-grained control over concurrency and order based on worker, arguments, or metadata. ```elixir # Chain by :worker (default behavior) use Oban.Pro.Worker, chain: [by: :worker] # Chain by a single args key use Oban.Pro.Worker, chain: [by: [args: :account_id]] # Chain by multiple args keys use Oban.Pro.Worker, chain: [by: [args: [:account_id, :cohort]]] # Chain by worker and a single args key use Oban.Pro.Worker, chain: [by: [:worker, args: :account_id]] # Chain by worker and multiple args keys use Oban.Pro.Worker, chain: [by: [:worker, args: [:account_id, :cohort]]] # Chain by a single meta key use Oban.Pro.Worker, chain: [by: [meta: :source_id]] ``` -------------------------------- ### Oban Pro: Global Concurrency Configuration Examples Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Engines.Smart Demonstrates how to configure global concurrency limits for queues in Oban Pro. Global concurrency limits the number of jobs running across all nodes in a cluster, complementing local limits. ```elixir # Execute 10 jobs concurrently across all nodes, with up to 10 on a single node my_queue: [global_limit: 10] # Execute 10 jobs concurrently, but only 3 jobs on a single node my_queue: [local_limit: 3, global_limit: 10] # Execute at most 1 job concurrently across all nodes my_queue: [global_limit: 1] ``` -------------------------------- ### Configure Oban Pro Testing Helpers with Options Source: https://getoban.pro/docs/pro/1.6.2/testing Configures `Oban.Pro.Testing` with additional options like `repo`, `prefix`, and `log`. The `repo` option is mandatory as it's injected into many generated helpers. ```Elixir using do quote do use Oban.Pro.Testing, repo: MyApp.Repo, prefix: "private", log: :debug end end ``` -------------------------------- ### Retrieve Recorded Job Results Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Worker Fetch the recorded output of a job using the `fetch_recorded/1` function. This example shows how to retrieve the result or handle cases where no result is recorded. ```elixir job = MyApp.Repo.get(Oban.Job, job_id) case MyApp.RecordedWorker.fetch_recorded(job) do {:ok, result} -> # Use the result {:error, :missing} -> # Nothing recorded yet end ``` -------------------------------- ### Oban Pro Cloud Callback: init Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Cloud Callback executed once at runtime to gather, normalize, and transform options for cloud hosting providers. It returns the initial configuration. ```APIDOC @callback init(opts()) :: conf() Executed once at runtime to gather, normalize, and transform options. ``` -------------------------------- ### Define Recorded Worker with Custom Limit Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Worker Configure a recorded worker with a specific output size limit. This example sets the limit to 128MB, overriding the default 64MB. ```elixir use Oban.Pro.Worker, recorded: [limit: 128_000_000] ``` -------------------------------- ### Run Basic Workflow Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Testing Demonstrates how to create and run a basic Oban workflow with sequential job dependencies. The workflow is built using a pipe operator, adding jobs `:a`, `:b`, and `:c` with specified dependencies. ```Elixir MyFlow.new_workflow() |> MyFlow.add(:a, MyFlow.new(%{id: 1})) |> MyFlow.add(:b, MyFlow.new(%{id: 2}), deps: [:a]) |> MyFlow.add(:c, MyFlow.new(%{id: 3}), deps: [:b]) |> run_workflow() ``` -------------------------------- ### Drain Jobs API Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Testing API documentation for the `drain_jobs` function, detailing its options for controlling queue draining behavior, concurrency, recursion, and result summarization. Includes usage examples for various scenarios. ```APIDOC drain_jobs(opts \\ []) Drains jobs from the Oban queue. ## Options * `:queue` - an atom specifying the queue to drain, or `:all` to drain jobs across all queues at once. Defaults to `:all` when no queue is provided. * `:with_limit` — the maximum number of jobs to fetch for draining at once. The limit only impacts how many jobs are fetched at once, *not* concurrency. When recursion is enabled this is how many jobs are processed per-iteration, and it defaults to `1`. Otherwise, there isn't a limit and all available jobs are fetched. * `:with_recursion` — whether to draining jobs recursively, or all in a single pass. Either way, jobs are processed sequentially, one at a time. Recursion is required when jobs insert other jobs (e.g. batches), or depend on the execution of other jobs (e.g. workflows). Defaults to `true`. * `:with_safety` — whether to silently catch errors when draining. When `false`, raised exceptions or unhandled exits are reraised (unhandled exits are wrapped in [`Oban.CrashError`](https://hexdocs.pm/oban/2.19.2/Oban.CrashError.html)). Defaults to `false`. * `:with_scheduled` — whether to include scheduled or retryable jobs when draining. In recursive mode, which is the default, this will include snoozed jobs, and may lead to an infinite loop if the job snoozes repeatedly. Defaults to `true`. * `:with_summary` — whether to summarize execution results with a map of counts by state, or return a list of each job that was drained. Defaults to `true`, which returns a summary map. See [shared options](#module-shared-repo-options) for additional repo-specific options. ## Examples Drain all available jobs across all queues: ```elixir assert %{completed: 3} = drain_jobs(queue: :all) ``` Drain and return all executed jobs without a count summary: ```elixir assert [%Oban.Job{}, %Oban.Job{}] = drain_jobs(with_summary: false) ``` Drain including a job that you expect to raise: ```elixir assert_raise RuntimeError, fn -> drain_jobs(queue: :risky) end ``` Drain without recursion to identify snoozed jobs: ```elixir assert %{scheduled: 3} = drain_jobs(with_recursion: false) ``` Drain without staging scheduled jobs: ```elixir assert %{completed: 1, scheduled: 0} = drain_jobs(with_scheduled: false) ``` Drain within a custom prefix: ```elixir assert %{completed: 3} = drain_jobs(queue: :default, prefix: "private") ``` Drain using a specific repo (necessary when calling this function directly): ```elixir assert %{completed: 3} = drain_jobs(queue: :default, repo: MyApp.Repo) ``` ``` -------------------------------- ### Run Workflow with Job Matching Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Testing Shows how to run a workflow and match on the returned jobs. It highlights that the execution order of jobs might differ from the insertion order, especially when dependencies are involved. ```Elixir workflow = \ MyFlow.new_workflow() |> MyFlow.add(:a, MyFlow.new(%{id: 1})) |> MyFlow.add(:b, MyFlow.new(%{id: 2}), deps: [:a]) |> MyFlow.add(:c, MyFlow.new(%{id: 3}), deps: [:b]) [_job_a, _job_b, _job_c] = run_workflow(workflow, with_summary: false) ``` -------------------------------- ### Get Specific Named Jobs Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Workflow Retrieve specific jobs from a workflow by their names or by providing a workflow ID. This allows targeted access to jobs within a workflow structure. ```Elixir [:job_a, :job_b] = Workflow.all_jobs(job, names: [:a, :b]) jobs = Workflow.all_jobs("some-uuid-1234-5678") ``` -------------------------------- ### Configure Oban Pro for Testing Source: https://getoban.pro/docs/pro/1.6.2/testing Sets Oban Pro to `:manual` testing mode in `test.exs`. This is crucial for testing Pro features like `Batch` and `Workflow` workers, especially when not using `:inline` mode. ```Elixir # test.exs config :my_app, Oban, testing: :manual ``` -------------------------------- ### Define Typespec for Structured Worker Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Worker Manually define a typespec for a structured worker to provide type hints and documentation. This example shows a basic worker with an ID and an active status. ```elixir defmodule MyApp.StructuredWorker do use Oban.Pro.Worker @type t :: %__MODULE__{id: integer(), active: boolean()} ``` -------------------------------- ### Oban Pro DynamicPartitioner API Types and Functions Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicPartitioner API documentation for Oban Pro's DynamicPartitioner, including type definitions for options and retention periods, and the signature and description for the `backfill_jobs/2` function. ```APIDOC Types: option() :: {:conf, Oban.Config.t()} | {:name, GenServer.name()} | {:retention, retention()} | {:schedule, String.t()} | {:timeout, timeout()} | {:timezone, String.t()} retention() :: [ completed: pos_integer(), cancelled: pos_integer(), discarded: pos_integer() ] Functions: backfill_jobs(conf_or_name, opts \ []) Backfill jobs from a standard table into a newly partitioned table. Parameters: conf_or_name: Configuration or name of the Oban instance. opts: Options for backfilling, including `:new_prefix`, `:old_prefix`, `:states`, `:batch_size`, `:sleep_ms`. ``` -------------------------------- ### Build and Insert Basic Workflow Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Workflow Demonstrates creating a new workflow, adding sequential jobs with dependencies, and inserting the entire workflow into Oban. It shows a simple linear execution flow. ```elixir alias MyApp.WorkerA alias MyApp.WorkerB alias MyApp.WorkerC Workflow.new() |> Workflow.add(:setup, WorkerC.new(%{mode: "initialize"})) |> Workflow.add_workflow(:extract, extr_flow, deps: :setup) |> Workflow.add_workflow(:notify, note_flow, deps: :extract) |> Workflow.add(:finalize, WorkerC.new(%{mode: "cleanup"}), deps: :notify) |> Oban.insert_all() ``` -------------------------------- ### Update SmartEngine Namespace Source: https://getoban.pro/docs/pro/1.6.2/v1-0 The SmartEngine has been moved to an `Engines` namespace for consistency with Oban's engines. Update your configuration to reference the new path. ```Elixir - engine: Oban.Pro.Queue.SmartEngine + engine: Oban.Pro.Engines.Smart ``` -------------------------------- ### Get All Workflow Jobs Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Workflow Retrieves all jobs associated with a workflow, with options to filter by upstream dependencies, specific names, or include sub-workflow jobs. Can be used with a job object or a workflow ID. ```Elixir def process(job) do job |> Workflow.all_jobs() |> do_things_with_jobs() :ok end ``` ```Elixir jobs = Workflow.all_jobs(job, only_deps: true) ``` ```Elixir [job_a, job_b] = Workflow.all_jobs(job, names: [:a, :b]) ``` ```Elixir [sub_a, sub_b] = Workflow.all_jobs(job, names: [:sub], with_subs: true) ``` ```Elixir jobs = Workflow.all_jobs("some-uuid-1234-5678") ``` ```Elixir [job_a, job_b] = Workflow.all_jobs("some-uuid-1234-5678", deps: [:a, :b]) ``` ```Elixir jobs = Workflow.all_jobs(MyApp.Oban, "some-uuid-1234-5678") ``` -------------------------------- ### Configure DynamicPartitioner Schedule and Buffer Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicPartitioner Configuration snippet to customize the `DynamicPartitioner` plugin's behavior. Allows setting a custom buffer period for pre-creating partitions and defining a specific schedule and timezone for partition management. ```Config plugins: [{ Oban.Pro.Plugins.DynamicPartitioner, buffer: 3, schedule: "0 5 * * *", timezone: "Europe/Paris" }] ``` -------------------------------- ### Force Job Cancellation on Deadline Expiry Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Worker Shows how to configure a worker to forcefully cancel itself if it starts processing but cannot complete before its deadline. This is achieved by setting the `force: true` option within the `deadline` configuration. ```Elixir defmodule FirmDeadlinedWorker do use Oban.Pro.Worker, deadline: [in: {10, :minutes}, force: true] # ... worker implementation ... end ``` -------------------------------- ### Update Batch Creation from new_batch to Oban.Pro.Batch.new Source: https://getoban.pro/docs/pro/1.6.2/v1-5 Illustrates the change in how new batches are created. The old `MyApp.BatchWorker.new_batch(jobs)` syntax is replaced by the more direct `Oban.Pro.Batch.new(jobs)` call, simplifying batch instantiation. ```elixir -MyApp.BatchWorker.new_batch(jobs) +Oban.Pro.Batch.new(jobs) ``` -------------------------------- ### Scaler Configuration: Cooldown Period Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicScaler Sets a custom cooldown period between scaling events. This example configures a 5-minute (300 seconds) cooldown to prevent rapid, successive scaling actions. ```Elixir scalers: [range: 1..3, cloud: MyApp.Cloud, cooldown: 300] ``` -------------------------------- ### Partitioning Schemes Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Engines.Smart Demonstrates different ways to define partition schemes for Oban Pro jobs. Partitions can be based on worker identity, arguments, or a combination thereof, helping to manage concurrency and distribute workloads effectively. ```ruby # Partition by worker alone partition: :worker # Partition by the `id` and `account_id` from args, ignoring the worker partition: [args: [:id, :account_id]] # Partition by worker and the `account_id` key from args partition: [:worker, args: :account_id] ``` -------------------------------- ### Oban Smart Engine Snooze Handling Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Engines.Smart Provides an example of how to check a job's meta for snooze information within the `process/1` function. This helps differentiate between actual attempts and snoozes. ```Elixir def process(job) do case job.meta do %{"orig_scheduled_at" => unix_microseconds, "snoozed" => snoozed} -> IO.inspect({snoozed, unix_microseconds}, label: "Times snoozed since") _ -> # This job has never snoozed before end end ``` -------------------------------- ### Get Workflow Status Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Workflow Retrieve detailed status information for a workflow, including its overall state, job counts, execution duration, and sub-workflow details. The output is a map containing comprehensive metrics. ```Elixir status = Workflow.status(job) ``` -------------------------------- ### Authorize Oban Pro Hex Repo in CI/CD Source: https://getoban.pro/docs/pro/1.6.2/ci_cd This snippet demonstrates how to add the Oban Pro Hex repository to your CI/CD environment. It uses the `mix hex.repo add` command to register the repository with its public key and an authentication key, typically stored as a secret in the CI service. This ensures secure access to Oban Pro packages during builds. ```shell - name: Authorize Oban run: | mix hex.repo add oban https://repo.oban.pro \ --fetch-public-key SHA256:4/OSKi0NRF91QVVXlGAhb/BIMLnK8NHcx/EWs+aIWPc \ --auth-key ${{secrets.oban_license_key}} ``` -------------------------------- ### Generate Flowchart with Mermaid Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Workflow This example demonstrates how to generate a flowchart representation of a workflow using the `to_mermaid` function from the Oban Pro library. It takes a workflow object as input and outputs a string formatted for Mermaid.js. ```Elixir Workflow.to_mermaid(workflow) ``` -------------------------------- ### Run Ecto Migration Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicPartitioner Executes the pending Ecto migrations, including the one created for the partitioned oban_jobs table. ```bash mix ecto.migrate ``` -------------------------------- ### Basic Oban Pro Migration Implementation Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Migration Shows how to implement the generated migration by delegating the `up/0` and `down/0` functions to Oban.Pro.Migration. This applies all necessary migrations for the database. ```elixir defmodule MyApp.Repo.Migrations.AddObanPro do use Ecto.Migration def up, do: Oban.Pro.Migration.up() def down, do: Oban.Pro.Migration.down() end ``` -------------------------------- ### Implement Workflow Cancellation Callback Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Workflow Provides an example of implementing the `after_cancelled/2` callback for workflows. This callback is invoked when a workflow job is cancelled due to upstream dependency changes, allowing for custom cleanup logic. ```elixir defmodule MyApp.Workflow do use Oban.Pro.Worker @behaviour Oban.Pro.Workflow require Logger @impl Oban.Pro.Workflow def after_cancelled(reason, job) do Logger.warn("Workflow job #{job.id} cancelled because a dependency was #{reason}") # Perform any cleanup required notify_monitoring_system(job, reason) :ok end end ``` -------------------------------- ### Elixir Cloud Module for AWS EC2 Scaling Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicScaler Example implementation of an Oban.Pro.Cloud behaviour for scaling AWS EC2 Auto Scaling Groups. It uses the ex_aws package to interact with the AWS API. ```elixir defmodule MyApp.ASG do @behaviour Oban.Pro.Cloud @impl Oban.Pro.Cloud def init(opts), do: Map.new(opts) @impl Oban.Pro.Cloud def scale(desired, conf) do params = %{ "Action" => "SetDesiredCapacity", "AutoScalingGroupName" => conf.asg, "DesiredCapacity" => desired, "Version" => "2011-01-01" } query = %ExAws.Operation.Query{ path: "", params: params, service: :autoscaling, action: :set_desired_capacity } with {:ok, _} <- ExAws.request(query), do: {:ok, conf} end end ``` -------------------------------- ### run_workflow: Oban Pro Test Helper Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Testing Inserts and executes a workflow synchronously within the test process. This helper augments the workflow with options optimized for testing but respects all standard workflow options. It accepts options for drain_jobs/1 and repo-specific options. ```APIDOC run_workflow(workflow, opts \ []) (since 0.11.0) @spec run_workflow( Oban.Pro.Workflow.t(), Oban.Pro.Workflow.new_opts() | drain_option() ) :: drain_result() Insert and execute a workflow synchronously, within the test process. This helper augments the workflow with options optimized for testing, but it will still respect all standard workflow options. ## Options Accepts all options for [`drain_jobs/1`](#drain_jobs/1), including the repo-specific options listed in [shared-options](#module-shared-repo-options). ``` -------------------------------- ### Generate Partitioned Oban Jobs Migration Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicPartitioner Generates a new Ecto migration file to create a partitioned oban_jobs table. This is the first step before configuring the DynamicPartitioner plugin. ```bash mix ecto.gen.migration add_partitioned_oban_jobs ``` -------------------------------- ### Put and Get Workflow Context Source: https://getoban.pro/docs/pro/1.6.2/changelog Illustrates using `Workflow.put_context/3` to share common data across jobs in a workflow and `Workflow.get_context/1` within a worker to access this shared data. ```elixir workflow = Workflow.new() |> Workflow.put_context(%{user_id: 123, app_version: "1.2.3"}) |> Workflow.add(:job_a, WorkerA.new(%{})) |> Workflow.add(:job_b, WorkerB.new(%{})) # Later in a worker: def process(job) do context = Workflow.get_context(job) # Use context map... end ``` -------------------------------- ### Define before_new Hook to Augment Job Arguments Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Worker Shows how to define a before_new/2 hook to modify or augment the arguments and options used when creating new jobs. This example injects a span_id into the job's meta. ```elixir defmodule MyApp.SpanHook do def before_new(args, opts) do opts = opts |> Keyword.put_new(:meta, %{}) |> put_in([:meta, :span_id], MyApp.Telemetry.gen_span_id()) {:ok, args, opts} end end ``` -------------------------------- ### Rate Limit with Partitioning Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Engines.Smart Shows how to configure rate limits that are applied per partition. This allows for granular control over how frequently jobs within specific partitions can be executed within a given time period. ```ruby # Allow one job per worker, every ten seconds, across the cluster local_limit: 10, rate_limit: [allowed: 1, period: 10, partition: :worker] ``` -------------------------------- ### Basic Scaler Configuration Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicScaler Shows a fundamental configuration for a DynamicScaler, specifying the node range and the cloud integration module. This sets up the minimum and maximum number of nodes and how to interact with the cloud provider. ```Elixir {DynamicScaler, scalers: [range: 1..5, cloud: MyApp.Cloud]} ``` -------------------------------- ### Get All Recorded Workflow Job Outputs Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Workflow Retrieves recorded outputs for workflow jobs, optionally filtered by name or relationship. Returns a map where keys are job names and values are their recorded outputs, or nil if unrecorded. ```Elixir def process(job) do %{"a" => job_a_return, "b" => job_b_return} = Workflow.all_recorded(job) :ok end ``` ```Elixir recorded = Workflow.all_recorded(job, only_deps: true) ``` ```Elixir %{"a" => _, "b" => _} = Workflow.all_recorded(job, names: [:a, :b]) ``` ```Elixir %{"a" => _, "b" => _} = Workflow.all_recorded("some-uuid-1234-5678", names: [:a, :b]) ``` -------------------------------- ### Configure Two Oban Instances for Transition Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicPartitioner Sets up two separate Oban instances during a migration: one for the original table and one for the new partitioned table. This allows processing of existing jobs while new jobs use the partitioned table. ```elixir config :my_app, Oban.Original, queues: [ default: 10, other_queue: 10, and_another: 10 ] config :my_app, Oban, prefix: "partitioned", queues: [ default: 10, other_queue: 10, and_another: 10 ], plugins: [ ... ] ``` -------------------------------- ### Oban Pro Batch Creation and Insertion Source: https://getoban.pro/docs/pro/1.6.2/composition Shows how to create a new Oban Pro Batch and insert multiple jobs into it using Elixir. This includes setting callback options and inserting the batch into Oban. ```elixir alias Oban.Pro.Batch alias MyApp.ImageProcessor image_ids |> Enum.map(&ImageProcessor.new(%{image_id: &1, operations: ~w(resize optimize watermark)})) |> Batch.new(callback_opts: [args: %{account_id: current_user.id}]) |> Oban.insert_all() ``` -------------------------------- ### Elixir Batch Processing with Cascading Sub-Workflows Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Workflow Illustrates using cascading functions for batch processing in Elixir. This example shows a fan-out pattern where multiple data sources are processed in parallel, and their results are aggregated in subsequent steps. ```Elixir defmodule MyApp.BatchETL do def insert(sources, date) do Workflow.new() |> Workflow.put_context(%{date: date}) |> Workflow.add_cascade(:sources, {sources, &extract_source/2}) |> Workflow.add_cascade(:transform, &transform_all/1, deps: :sources) |> Workflow.add_cascade(:load, &load_data/1, deps: :transform) |> Workflow.add_cascade(:notify, &send_notification/1, deps: [:transform, :load]) |> Oban.insert_all() end # Each `source` from the list is passed as the first argument def extract_source(source, context) do %{source: source, records: [1, 2, 3], extracted_at: DateTime.utc_now()} end def transform_all(%{sources: sources}) do total_records = Enum.sum_by(sources, fn {_key, result} -> length(result.records) end) transformed = sources |> Enum.flat_map(fn {_key, %{records: records}} -> records end) |> Enum.map(&(&1 * 2)) %{records: transformed, count: total_records} end def load_data(%{transform: %{records: records}}) do %{loaded: true, count: length(records)} end def send_notification(%{transform: transform, load: load, date: date}) do IO.puts("On #{date}, processed #{transform.count} records, loaded #{load.count}") end end ``` -------------------------------- ### Configure Chunk Worker for Larger Batches and Less Frequent Runs Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Workers.Chunk This example shows how to adjust the `size` and `timeout` options for a Chunk worker to process larger batches less frequently. Increasing these values can optimize throughput for specific workloads. ```elixir use Oban.Pro.Workers.Chunk, size: 500, timeout: :timer.seconds(5) ``` -------------------------------- ### Make Predeps Script Executable Source: https://getoban.pro/docs/pro/1.6.2/heroku Ensures the custom pre-dependency script is executable by the system. ```shell $chmod +x ./bin/predeps ```