class Crabbit::Producer

• Crabbit::Producer
• Reference
• Object

Overview

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.

Defined in:

crabbit/producer.cr

Constructors

• .new(environment : Environment, stream : String, options : ProducerOptions)

  Creates and declares a producer.

Instance Method Summary

• #close : Nil

  Idempotently deletes the publisher and fails unresolved handles.

• #closed? : Bool

  Returns whether this producer was permanently closed.

• #last_publishing_id : UInt64

  Queries RabbitMQ for the last publishing ID of this named producer.

• #open? : Bool

  Returns whether this producer is currently ready to publish.

• #options : ProducerOptions

  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.

• #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.

• #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

  Wraps bytes in one AMQP Data section and publishes the message.

• #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.

• #state : ResourceState

  Returns the current lifecycle state.

• #stream : String

  Returns the target stream name.

• #unconfirmed_count : Int32

  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.

Constructor Detail

Instance Method Detail