struct NATS::JetStream::Streams

Overview

A stream in NATS JetStream represents the history of messages pertaining to a given domain. When you publish a message to a subject that a stream is monitoring, the stream then adds that message to its history in the order it was published.

Defined in:

jetstream/streams.cr

Constructors

Instance Method Summary

Constructor Detail

def self.new(nats : NATS::Client) #

[View source]

Instance Method Detail

def create(config : StreamConfig) : Stream #

[View source]
def create(storage : StreamConfig::Storage, retention : StreamConfig::RetentionPolicy | Nil = nil, discard : StreamConfig::DiscardPolicy | Nil = nil, compression : StreamConfig::Compression = :none, persist_mode : StreamConfig::PersistMode | Nil = nil, **kwargs) : Stream #

Create a stream of the given storage type and with the given properties, which are passed unmodified to NATS::JetStream::StreamConfig.new.


[View source]
def delete(stream : JetStream::Stream) #

Delete the given stream


[View source]
def delete(stream : String) #

Delete the stream with the given name


[View source]
def direct_get(stream : String, *, last_by_subject : String) : JetStream::StreamGetMsgResponse | Nil #

The #get_msg API involves some administrative overhead and usually routes to the stream's primary node in the NATS cluster. The #direct_get method uses the DIRECT.GET API which allows any server hosting the stream (including replicas) to respond with the message data. Specifying last_by_subject allows you to get the last message in a stream that was published to a specific subject. This is used by NATS::KV internally to fetch values for a given key from replicas for the KV's backing stream.

NOTE In order to use this API, the stream must have been created with allow_direct set to true. For performance reasons, the client does not perform this check.


[View source]
def direct_get(stream : String, *, sequence : Int, next_by_subject : String | Nil = nil) #

The #get_msg API involves some administrative overhead and usually routes to the stream's primary node in the NATS cluster. The #direct_get method uses the DIRECT.GET API which allows any server hosting the stream (including replicas) to respond with the message data. Specifying last_by_subject allows you to get the last message in a stream that was published to a specific subject. This is used by NATS::KV internally to fetch values for a given key from replicas for the KV's backing stream.

NOTE In order to use this API, the stream must have been created with allow_direct set to true. For performance reasons, the client does not perform this check.


[View source]
def get_msg(stream : String, *, last_by_subject : String) #

[View source]
def get_msg(stream : String, *, sequence : Int, next_by_subject : String | Nil = nil) #

[View source]
def info(name : String) : Stream | Nil #

Get the current state of the stream with the given name


[View source]
def list(subject : String | Nil = nil, offset : Int | Nil = nil, limit : Int | Nil = nil) #

List all available streams


[View source]
def purge(stream : String, subject : String) : Int64 #

[View source]
def update(config : StreamConfig) : Stream #

Update an existing stream with the given config. The server replaces the stream's entire config, so any settings not in config are reset to their defaults. To change only some settings, use the update(name, **changes) overload instead.


[View source]
def update(name : String, **changes) : Stream #

Change the given settings on the stream with the given name, leaving all of its other settings as they are. The setting names are the same as the properties on NATS::JetStream::StreamConfig.

jetstream.stream.update "orders", max_age: 7.days

[View source]