Struct async_nats::jetstream::context::Context

source · [−]

pub struct Context { /* private fields */ }

Expand description

A context which can perform jetstream scoped requests.

Implementations

source

impl Context

source

pub async fn publish(
 &self,
 subject: String,
 payload: Bytes
) -> Result<PublishAck, Error>

Publish a message to a given subject associated with a stream and returns an acknowledgment from the server that the message has been successfully delivered.

If the stream does not exist, no responders error will be returned.

Examples

let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let ack = jetstream.publish("events".to_string(), "data".into()).await?;

source

pub async fn publish_with_headers(
 &self,
 subject: String,
 headers: HeaderMap,
 payload: Bytes
) -> Result<PublishAck, Error>

Publish a message with headers to a given subject associated with a stream and returns an acknowledgment from the server that the message has been successfully delivered.

If the stream does not exist, no responders error will be returned.

Examples

let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let mut headers = async_nats::HeaderMap::new();
headers.append("X-key", b"Value".as_ref().try_into()?);
let ack = jetstream.publish_with_headers("events".to_string(), headers, "data".into()).await?;

source

pub async fn query_account(&self) -> Result<Account, Error>

Query the server for account information

source

pub async fn create_stream<S>(&self, stream_config: S) -> Result<Stream, Error> where
Config: From<S>,

Create a JetStream Stream with given config and return a handle to it. That handle can be used to manage and use Consumer.

Examples

use async_nats::jetstream::stream::Config;
use async_nats::jetstream::stream::DiscardPolicy;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let stream = jetstream.create_stream(Config {
    name: "events".to_string(),
    max_messages: 100_000,
    discard: DiscardPolicy::Old,
    ..Default::default()
}).await?;

source

pub async fn get_stream<T: AsRef<str>>(&self, stream: T) -> Result<Stream, Error>

Checks for Stream existence on the server and returns handle to it. That handle can be used to manage and use Consumer.

Examples

let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let stream = jetstream.get_stream("events").await?;

source

pub async fn get_or_create_stream<S>(
 &self,
 stream_config: S
) -> Result<Stream, Error> where
 S: Into<Config>,

Create a stream with the given configuration on the server if it is not present. Returns a handle to the stream on the server.

Note: This does not validate if the Stream on the server is compatible with the configuration passed in.

Examples

use async_nats::jetstream::stream::Config;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let stream = jetstream.get_or_create_stream(Config {
    name: "events".to_string(),
    max_messages: 10_000,
    ..Default::default()
}).await?;

source

pub async fn delete_stream<T: AsRef<str>>(
&self,
stream: T
) -> Result<DeleteStatus, Error>

Deletes a Stream with a given name.

Examples

use async_nats::jetstream::stream::Config;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let stream = jetstream.delete_stream("events").await?;

source

pub async fn update_stream<S>(&self, config: S) -> Result<Info, Error> where
S: Borrow<Config>,

Updates a Stream with a given config. If specific field cannot be updated, error is returned.

Examples

use async_nats::jetstream::stream::Config;
use async_nats::jetstream::stream::DiscardPolicy;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let stream = jetstream.update_stream(&Config {
    name: "events".to_string(),
    discard: DiscardPolicy::New,
    max_messages: 50_000,
    ..Default::default()
}).await?;

source

pub async fn request<T, V>(
 &self,
 subject: String,
 payload: &T
) -> Result<Response<V>, Error> where
 T: ?Sized + Serialize,
 V: DeserializeOwned,

Send a request to the jetstream JSON API.

This is a low level API used mostly internally, that should be used only in specific cases when this crate API on Consumer or Stream does not provide needed functionality.

Examples

let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let response: Response<Info> = jetstream
.request("STREAM.INFO.events".to_string(), &()).await?;