Zizq.Stream.Take (Zizq v0.6.1)

Copy Markdown View Source

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)
end

Messages

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 a Zizq.Error or :closed for 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.

Start a stream.

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

start_link(opts)

@spec start_link(keyword()) :: GenServer.on_start()

Start a stream.

Options

  • :client (atom/0) - Required. Name of a running Zizq client.

  • :owner (pid/0) - Process to deliver messages to. Defaults to the caller.

  • :queues (list of String.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.