Gain insights into data processing and concurrency in Elixir. Explore GenServer, GenStage, Flow, and Broadway to enhance performance, set up pipelines, and handle increased workloads efficiently.

Concurrent Data Processing in Elxir by The Pragmatic Programmer-01.png

myDokcer.tar.gz

ch1_1

ch_3

ch_3.1

ch1_1-task supervisor

ch_4.1

ch_5.1 RabbitMQ

Ch_5.2 RabbitMQ

Most engineers don’t think that concurrency is distinct from programming. Data processing includes transforming information in some way, like reporting, data aggregation, or analytics. On the other hand, concurrency involves significantly improving performance.

This course is designed to help you navigate the world of concurrency tools available in the Elixir ecosystem. You’ll explore popular modules and libraries like GenServer, GenStage, Flow, and Broadway. You’ll be able to ingest data, set up pipelines, and run concurrent processes with Elixir.

By the end of the course, you’ll have a wide range of new techniques to simplify your product, improve code performance, and make your application more resilient to increased workloads.

Concurrent Data Processing in Elixir

# Update consumer demand

Now, for our `scraper` project, let’s lower the demand to something more suitable:

#scraper/lib/page_consumer.ex
def init(initial_state) do
  Logger.info("PageConsumer init")
  sub_opts = [{PageProducer, min_demand: 0, max_demand: 3}]
  {:consumer, initial_state, subscribe_to: sub_opts}
end

With `min_demand` set to `0` and `max_demand` set to `3`, we ensure that the consumer takes at least one event when available, up to 3 at a time. However, our producer still has to supply the consumers with events, and right now, it doesn’t do anything. Let’s fix this.

Going back to `PageProducer`, we have implemented the `handle_demand/2` callback to respond to demand and produce events. Remember that `handle_demand/2` is always called when the producer receives a demand from a consumer. Since consumers always demand events when they’re free, this callback is useful when we want to respond to consumer demand immediately and keep consumers busy.

# Expose an API

Instead, we would like to expose an API that developers can use to request pages to be scraped. Consumers will have to wait patiently until work is available for them in this particular use case. Therefore, `handle_demand/2` isn’t very useful for what we’re trying to achieve.

## Dispatch events

Thankfully, there are several ways for a producer to dispatch events. Most callbacks for `:producer` and `:producer_consumer` stages have the ability to do so.

Earlier, we mentioned that `GenStage` is built on top of `GenServer`. This means that the callbacks we covered previously are also available for `GenStage` processes:
* `handle_call/3`

* `handle_cast/2` 

* `handle_info/3`

They are called when we invoke `GenStage.call/3`, `GenStage.cast/2` or `Process.send/3`, respectively. However, the return signatures of those callbacks have an important difference from their `GenServer` counterparts. Here are two examples of return tuples allowed for `GenStage`:

```elixir
{:reply, reply, [event], new_state}
{:noreply, [event], new_state}
```

Notice the extra element in the tuple that holds a list of event values. These callbacks work the same as the `handle_demand/3` callback. This is great news because it gives us a lot of flexibility when we dispatch events.

## Implement our API

Let’s implement our API in `PageProducer`:


We'll update the consumer demand in our scraper project and then build an API to send requests to scrape pages.

Getting Started

Easy Concurrency with the Task Module

Long-Running Processes Using GenServer

Data Processing Pipelines with GenStage

Process Collections with Flow

Data Ingestion Pipelines with Broadway

Concluding the Course

Revisiting the Producer

Update consumer demand