BlockingQueue is a simple queue implemented as a GenServer. It has a fixed maximum length established when it is created.
The queue is designed to decouple, but limit, the latency between a producer and
consumer. When pushing to a full queue the push
operation blocks
preventing the producer from making progress until the consumer catches up.
Likewise, when calling pop
on an empty queue the call blocks until there
is work to do.
Add a dependency in your mix.exs:
deps: [{:blocking_queue, "~> 1.0"}]
A simple example:
{:ok, pid} = BlockingQueue.start_link(5)
BlockingQueue.push(pid, "Hi")
BlockingQueue.pop(pid) # should return "Hi"
The queue is designed to be used from more complex examples in which the producer and consumer are in separate processes and run asynchronously to each other.
An example of an infinite stream:
{:ok, pid} = BlockingQueue.start_link(:infinity)
BlockingQueue.push(pid, "Hi")
BlockingQueue.pop(pid) # should return "Hi"
An example using the Stream
API
{:ok, pid} = BlockingQueue.start_link(5)
[1, 2, 3]
|> BlockingQueue.push_stream(pid)
BlockingQueue.pop_stream(pid)
|> Enum.take(3) # Should return [1, 2, 3]
Just fork the repo, make your change, and send me a pull request.
Or, feel free to file and issue and start a discussion about a new feature you have in mind.