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

Add Flow to a GenStage Pipeline

Explore how to enhance a GenStage data-processing pipeline by integrating the Flow library in Elixir. Understand how to rewrite producer-consumer modules using Flow's abstractions, simplify code by removing manual process management, and configure supervision trees to support concurrent workflows. This lesson helps you efficiently manage event flows and improve pipeline clarity and performance.

We'll cover the following...

The scraper project

The scraper project has the following components in the data-processing pipeline:

  • It has one PageProducer process of the :producer type.

  • It has two OnlinePageConsumerProducer processes of the :producer_consumer type.

  • It has one PageConsumerSupervisor process of the :consumer type.

  • It has up to two PageConsumer processes started on demand by PageConsumerSupervisor.

Flow with the GenStage

To demonstrate how Flow works with GenStage, we’ll rewrite the original OnlinePageConsumerProducer implementation using Flow. When working with GenStage, two groups of functions are available to use. The first group is made to work with already running stages:

  • It works with the from_stages/2 stage to receive events from :producer stages.

  • It works with the through_stages/3 stage to send events to :producer_consumer stages and receive what they send in turn.

  • It works with the into_stages/3 stage to send events to :consumer or :producer_consumer stages.

All functions in this group require a list of the process ids (PIDs) to connect to the already running processes.

The second group of functions is useful when we want Flow ...