class

Crabbit::Environment

Owns Stream connections and creates producers, consumers, and management requests.

An environment lazily creates and reuses connections according to the current stream metadata. Closing it also closes all producers, consumers, super-stream resources, and pooled connections created through it.

environment = Crabbit::Environment.connect(
  "rabbitmq-stream://guest:guest@localhost:5552/%2f"
)
begin
  environment.create_stream("events") unless environment.stream_exists?("events")
ensure
  environment.close
end

Constructors

connect(uri : String = Configuration::DEFAULT_URI, **options) : self

Parses uri and returns an environment.

Named options are forwarded to Configuration.parse.

Source

new(configuration : Configuration = Configuration.new, compression_codecs : CompressionCodecs = CompressionCodecs.new)

Creates an environment without opening a connection.

Connections are established lazily by management and messaging methods. Pass a custom registry to replace or extend sub-entry compression codecs.

Source

Instance methods

Idempotently closes all resources and network connections.

Source

closed?

Returns whether this environment was closed.

Source

compression_codecs

Returns the compression-codec registry used by producers and consumers.

Source

configuration

Returns the immutable connection configuration.

Source

consumer(stream : String, options : ConsumerOptions = ConsumerOptions.new) : Consumer

Creates a pull consumer for stream.

Retrieve deliveries with Consumer#receive, Consumer#receive?, or Consumer#each.

Source

consumer(stream : String, options : ConsumerOptions = ConsumerOptions.new, &handler : Delivery -> ) : Consumer

Creates a callback consumer for stream.

The block runs on handler fibers and each delivery is marked processed after the block returns, even when it raises.

Source

create_stream(stream : String, options : StreamOptions = StreamOptions.new) : Nil

Creates stream with the supplied retention and placement options.

Raises BrokerError when RabbitMQ rejects the operation, including when the stream already exists.

Source

create_super_stream(name : String, partitions : Enumerable(String), binding_keys : Enumerable(String), arguments : Hash(String, String) = {} of String => String) : Nil

Creates a super stream from matching partition and binding-key lists.

Each partition must have a corresponding binding key. arguments are passed to RabbitMQ unchanged.

Source

delete_stream(stream : String) : Nil

Deletes stream.

Source

delete_super_stream(name : String) : Nil

Deletes the super stream name and its partitions.

Source

metadata(streams : Enumerable(String)) : Array(StreamMetadata)

Returns metadata for every requested stream in input order.

Per-stream failures are represented by StreamMetadata#response_code.

Source

partitions(super_stream : String) : Array(String)

Returns the ordered partition stream names of super_stream.

Source

producer(stream : String, options : ProducerOptions = ProducerOptions.new) : Producer

Creates a producer for stream.

Source

query_offset(reference : String, stream : String) : UInt64 | Nil

Returns the stored consumer offset, or nil when none exists.

Source

query_publisher_sequence(reference : String, stream : String) : UInt64

Returns the last publishing ID recorded for a named publisher.

RabbitMQ returns zero when no publisher sequence exists.

Source

resolve_offset(stream : String, offset : OffsetSpecification, properties : Hash(String, String) = {} of String => String) : UInt64

Resolves a relative or timestamp offset to an absolute stream offset.

properties are forwarded to RabbitMQ's Resolve Offset Specification command and are useful for broker extensions.

Source

route(routing_key : String, super_stream : String) : Array(String)

Returns the partitions selected by routing_key and broker bindings.

Source

store_offset(reference : String, stream : String, offset : UInt64) : Nil

Stores offset for the consumer reference and stream.

Source

stream_exists?(stream : String) : Bool

Returns whether stream currently exists.

Errors other than ResponseCode::StreamDoesNotExist are raised.

Source

stream_stats(stream : String) : StreamStats

Returns broker statistics for stream.

Source

super_stream_consumer(super_stream : String, options : ConsumerOptions, &handler : Delivery -> ) : SuperStreamConsumer

Creates callback consumers for all current and future partitions of a super stream.

Source

super_stream_producer(super_stream : String, options : SuperStreamProducerOptions) : SuperStreamProducer

Creates a producer that routes messages across a super stream.

Source