Search⌘ K
AI Features
Home
Courses
Concurrent Data Processing in Elixir

The Details of Broadway Callbacks

Explore how to implement key Broadway callbacks like handle_message, prepare_messages, and handle_failed to create a data ingestion pipeline in Elixir. Understand the configuration of producers and processors, set up concurrency options, and integrate the pipeline into your application's supervision tree for fault-tolerant concurrent processing.

We'll cover the following...

Now that we have the tickets project set up, we can start writing some code. Once a message comes from the message broker, we want to use it to create a ticket in the database for the customer and send a confirmation email to them.

Broadway callbacks

To accomplish this, we will use the Broadway behavior to configure and start the data-ingestion pipeline. There are four callbacks available for us to implement:

  • handle_message/3

  • prepare_messages/2

  • handle_failed/2

  • handle_batch/4

This section will focus on the first three callbacks in the list.

Create bookings_pipeline.ex

First, we create a new file, bookings_pipeline.ex, in the lib folder. Then, we define the BookingsPipeline module in the following way:

defmodule BookingsPipeline do
  use Broadway

end

We also add use Broadway to bring in the Broadway behavior.

Define start_link

Let’s add the start_link/1 function now. Here’s an outline of how the implementation looks:

def start_link(_args) do
  options = [
    name:
...