A GenStage producer that polls a Queutils.BlockingQueue at a fixed interval, emitting any events on the queue.

Usage

Add it to your application supervisor's start/2 function, after the queue it pulls from, like this:

def start(_type, _args) do
  children = [
    ...
    {Queutils.BlockingQueue, name: MessageQueue, max_length: 10_000},
    {Queutils.BlockingQueueProducer, name: MessageProducer},
    ...
  ]

  opts = [strategy: :one_for_one, name: MyApplication.Supervisor]
  Supervisor.start_link(children, opts)
end

The subscribe a consumer to it, like any other GenStage producer.

def init(_opts) do
  {:consumer, :my_consumer_state, [subscribe_to: MessageProducer]}
end

Options

• :name - the ID of the queue. This will be the first argument to the push/2 function. Default is BlockingQueueProducer.
• :max_length - The maximum number of messages that this process will store until it starts blocking. Default is 1,000.
• :dispatcher - The GenStage dispatcher that this producer should use. Default is GenStage.DemandDispatcher.