A long-lived connection to /jobs/take, delivering jobs to an owner
process as they arrive.
This is the low-level consumer. Most applications want Zizq.Worker,
which runs one of these and handles concurrency, acknowledgement and
shutdown. Use this directly only if you need to drive the loop
yourself.
{:ok, _pid} = Zizq.Stream.Take.start_link(client: MyApp.Zizq, prefetch: 10)
receive do
{:zizq_stream, _pid, {:job, job}} -> handle(job)
endMessages
The owner (by default whoever called start_link/1) receives:
{:zizq_stream, pid, {:connected, url}}{:zizq_stream, pid, {:job, %Zizq.Job{}}}{:zizq_stream, pid, {:disconnected, reason}}— reason is aZizq.Erroror:closedfor a clean end of stream
Flow control
The server sends at most :prefetch unacknowledged jobs before it
pauses, so acknowledging is what asks for more. That bounds this
process's mailbox by construction: it cannot be flooded faster than
jobs are being completed, however far behind the consumer falls.
Reconnection
Disconnects are expected on a connection meant to live forever, so they are retried with exponential backoff rather than crashing the process. A response the server will answer identically next time — a rejected queue name, say — is not retried; the process stops instead, because retrying could only loop.
Summary
Functions
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec start_link(keyword()) :: GenServer.on_start()
Start a stream.
Options
:owner(pid/0) - Process to deliver messages to. Defaults to the caller.:queues(list ofString.t/0) - Queues to take from. Empty means every queue. The default value is[].:prefetch(pos_integer/0) - Maximum unacknowledged jobs the server will send before pausing. Omitted by default, so the server's own default applies.:worker_id(String.t/0) - Identifies this consumer in the server's logs. Assigned by the server if omitted.:name(term/0) - Optional GenServer name.