Asynchronous publisher for one RabbitMQ stream.
Producers batch queued messages, enforce ProducerOptions#max_unconfirmed,
and recover indefinitely until closed. Publishing queues work and returns a
PublishHandle; it does not wait for a broker confirmation.
Named producers resume the broker sequence and support deduplication.
Unnamed producers provide at-least-once recovery and can produce duplicates
when a connection fails with an unknown outcome.
Constructors
new(environment :
Environment, stream :
String, options :
ProducerOptions)
Creates and declares a producer.
Applications normally call Environment#producer so the environment can
own and close the resource.
Instance methods
close
Idempotently deletes the publisher and fails unresolved handles.
closed?
Returns whether this producer was permanently closed.
last_publishing_id
Queries RabbitMQ for the last publishing ID of this named producer.
Raises ConfigurationError for an unnamed producer.
open?
Returns whether this producer is currently ready to publish.
options
Returns the immutable producer options.
publish(message :
Message, filter :
String |
Nil = nil, *, publishing_id :
UInt64 |
Nil = nil) :
PublishHandle Publishes an AMQP message and returns immediately with a handle.
filter selects the server-side filter value and takes precedence over
ProducerOptions#filter_value_extractor. publishing_id overrides the
automatically allocated monotonically increasing ID.
publish(message :
Message, filter :
String |
Nil = nil, *, publishing_id :
UInt64 |
Nil = nil, &callback :
Confirmation -> ) :
PublishHandle Publishes an AMQP message and invokes the block on completion.
Returns the same handle that can also be awaited.
publish(message :
RawMessage, filter :
String |
Nil = nil, *, publishing_id :
UInt64 |
Nil = nil) :
PublishHandle Publishes already encoded AMQP bytes without re-encoding them.
publish(message :
RawMessage, filter :
String |
Nil = nil, *, publishing_id :
UInt64 |
Nil = nil, &callback :
Confirmation -> ) :
PublishHandle Publishes already encoded AMQP bytes and invokes the block on completion.
publish(bytes :
Bytes, filter :
String |
Nil = nil, *, publishing_id :
UInt64 |
Nil = nil) :
PublishHandle Copies bytes, wraps them in one AMQP Data section, and publishes the message.
The caller may mutate the original slice after this method returns.
publish(bytes :
Bytes, filter :
String |
Nil = nil, *, publishing_id :
UInt64 |
Nil = nil, &callback :
Confirmation -> ) :
PublishHandle Wraps bytes in one AMQP Data section, publishes it, and invokes the
block on completion.
publish(value :
String, filter :
String |
Nil = nil, *, publishing_id :
UInt64 |
Nil = nil) :
PublishHandle Encodes value as one AMQP Data section and publishes it.
publish(value :
String, filter :
String |
Nil = nil, *, publishing_id :
UInt64 |
Nil = nil, &callback :
Confirmation -> ) :
PublishHandle Encodes value as one AMQP Data section, publishes it, and invokes the
block on completion.
publish_confirmed(message :
Message |
RawMessage |
Bytes |
String, timeout :
Time::Span = options.confirm_timeout) :
Confirmation Publishes one message and waits for its final confirmation.
This convenience method does not raise negative confirmations; inspect
Confirmation#confirmed and Confirmation#error.
state
Returns the current lifecycle state.
stream
Returns the target stream name.
unconfirmed_count
Returns the number of logical messages awaiting a final outcome.
wait_for_confirms(timeout :
Time::Span = options.confirm_timeout) :
Nil Waits until every currently tracked publish has resolved.
Raises TimeoutError if unresolved messages remain after timeout.