class Crabbit::Environment

• Crabbit::Environment
• Reference
• Object

Overview

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

Defined in:

crabbit/environment.cr

Constructors

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

  Parses uri and returns an environment.

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

  Creates an environment without opening a connection.

Instance Method Summary

• #close : Nil

  Idempotently closes all resources and network connections.

• #closed? : Bool

  Returns whether this environment was closed.

• #compression_codecs : CompressionCodecs

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

• #configuration : Configuration

  Returns the immutable connection configuration.

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

  Creates a pull consumer for stream.

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

  Creates a callback consumer for stream.

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

  Creates stream with the supplied retention and placement options.

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

• #delete_stream(stream : String) : Nil

  Deletes stream.

• #delete_super_stream(name : String) : Nil

  Deletes the super stream name and its partitions.

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

  Returns metadata for every requested stream in input order.

• #partitions(super_stream : String) : Array(String)

  Returns the ordered partition stream names of super_stream.

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

  Creates a producer for stream.

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

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

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

  Returns the last publishing ID recorded for a named publisher.

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

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

  Returns the partitions selected by routing_key and broker bindings.

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

  Stores offset for the consumer reference and stream.

• #stream_exists?(stream : String) : Bool

  Returns whether stream currently exists.

• #stream_stats(stream : String) : StreamStats

  Returns broker statistics for stream.

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

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

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

  Creates a producer that routes messages across a super stream.

Constructor Detail

Instance Method Detail