Struct async_nats::jetstream::message::Message
source · Fields§
§message: Message§context: ContextImplementations§
source§impl Message
impl Message
sourcepub async fn ack(&self) -> Result<(), Error>
pub async fn ack(&self) -> Result<(), Error>
Acknowledges a message delivery by sending +ACK to the server.
If AckPolicy is set to All or Explicit, messages has to be acked.
Otherwise redeliveries will occur and Consumer will not be able to advance.
Examples
use futures::StreamExt;
use async_nats::jetstream::consumer::PullConsumer;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);
let consumer: PullConsumer = jetstream
.get_stream("events").await?
.get_consumer("pull").await?;
let mut messages = consumer.fetch().max_messages(100).messages().await?;
while let Some(message) = messages.next().await {
message?.ack().await?;
}sourcepub async fn ack_with(&self, kind: AckKind) -> Result<(), Error>
pub async fn ack_with(&self, kind: AckKind) -> Result<(), Error>
Acknowledges a message delivery by sending a chosen AckKind variant to the server.
Examples
use futures::StreamExt;
use async_nats::jetstream::AckKind;
use async_nats::jetstream::consumer::PullConsumer;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);
let consumer: PullConsumer = jetstream
.get_stream("events").await?
.get_consumer("pull").await?;
let mut messages = consumer.fetch().max_messages(100).messages().await?;
while let Some(message) = messages.next().await {
message?.ack_with(AckKind::Nak).await?;
}sourcepub async fn double_ack(&self) -> Result<(), Error>
pub async fn double_ack(&self) -> Result<(), Error>
Acknowledges a message delivery by sending +ACK to the server
and awaits for confirmation for the server that it received the message.
Useful if user wants to ensure exactly once semantics.
If AckPolicy is set to All or Explicit, messages has to be acked.
Otherwise redeliveries will occur and Consumer will not be able to advance.
Examples
use futures::StreamExt;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);
let consumer = jetstream
.get_stream("events").await?
.get_consumer("pull").await?;
let mut messages = consumer.fetch().max_messages(100).messages().await?;
while let Some(message) = messages.next().await {
message?.double_ack().await?;
}sourcepub fn info(&self) -> Result<Info<'_>, Error>
pub fn info(&self) -> Result<Info<'_>, Error>
Returns the JetStream message ID
if this is a JetStream message.
Examples found in repository?
src/jetstream/object_store/mod.rs (line 477)
462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596
fn poll_next(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> Poll<Option<Self::Item>> {
loop {
if self.done {
debug!("Object Store list done");
return Poll::Ready(None);
}
match self.subscription.poll_next_unpin(cx) {
Poll::Ready(message) => match message {
None => return Poll::Ready(None),
Some(message) => {
let message = message?;
let info = message.info()?;
trace!("num pending: {}", info.pending);
if info.pending == 0 {
self.done = true;
}
let response: ObjectInfo = serde_json::from_slice(&message.payload)?;
if response.deleted {
continue;
}
return Poll::Ready(Some(
serde_json::from_slice(&message.payload).map_err(|err| {
Box::from(std::io::Error::new(
ErrorKind::Other,
format!("failed to serialize object info: {err}"),
))
}),
));
}
},
Poll::Pending => return Poll::Pending,
}
}
}
}
/// Represents an object stored in a bucket.
pub struct Object<'a> {
pub info: ObjectInfo,
remaining_bytes: Vec<u8>,
has_pending_messages: bool,
digest: Option<ring::digest::Context>,
subscription: crate::jetstream::consumer::push::Ordered<'a>,
}
impl<'a> Object<'a> {
pub(crate) fn new(subscription: Ordered<'a>, info: ObjectInfo) -> Self {
Object {
subscription,
info,
remaining_bytes: Vec::new(),
has_pending_messages: true,
digest: Some(ring::digest::Context::new(&SHA256)),
}
}
/// Returns information about the object.
pub fn info(&self) -> &ObjectInfo {
&self.info
}
}
impl tokio::io::AsyncRead for Object<'_> {
fn poll_read(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> std::task::Poll<std::io::Result<()>> {
if !self.remaining_bytes.is_empty() {
let len = cmp::min(buf.remaining(), self.remaining_bytes.len());
buf.put_slice(&self.remaining_bytes[..len]);
self.remaining_bytes = self.remaining_bytes[len..].to_vec();
return Poll::Ready(Ok(()));
}
if self.has_pending_messages {
match self.subscription.poll_next_unpin(cx) {
Poll::Ready(message) => match message {
Some(message) => {
let message = message.map_err(|err| {
std::io::Error::new(
std::io::ErrorKind::Other,
format!("error from JetStream subscription: {err}"),
)
})?;
let len = cmp::min(buf.remaining(), message.payload.len());
buf.put_slice(&message.payload[..len]);
if let Some(context) = &mut self.digest {
context.update(&message.payload);
}
self.remaining_bytes
.extend_from_slice(&message.payload[len..]);
let info = message.info().map_err(|err| {
std::io::Error::new(
std::io::ErrorKind::Other,
format!("error from JetStream subscription: {err}"),
)
})?;
if info.pending == 0 {
let digest = self.digest.take().map(|context| context.finish());
if let Some(digest) = digest {
if format!("SHA-256={}", base64::encode_config(digest, URL_SAFE))
!= self.info.digest
{
return Poll::Ready(Err(io::Error::new(
ErrorKind::InvalidData,
"wrong digest",
)));
}
} else {
return Poll::Ready(Err(io::Error::new(
ErrorKind::InvalidData,
"digest should be Some",
)));
}
self.has_pending_messages = false;
}
Poll::Ready(Ok(()))
}
None => Poll::Ready(Err(std::io::Error::new(
std::io::ErrorKind::Other,
"subscription ended before reading whole object",
))),
},
Poll::Pending => Poll::Pending,
}
} else {
Poll::Ready(Ok(()))
}
}More examples
src/jetstream/kv/mod.rs (line 709)
700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813
fn poll_next(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<Option<Self::Item>> {
match self.subscription.poll_next_unpin(cx) {
Poll::Ready(message) => match message {
None => Poll::Ready(None),
Some(message) => {
let message = message?;
let info = message.info()?;
let operation = match message
.headers
.as_ref()
.and_then(|headers| headers.get(KV_OPERATION))
.unwrap_or(&HeaderValue::from(KV_OPERATION_PUT))
.iter()
.next()
.unwrap()
.as_str()
{
KV_OPERATION_DELETE => Operation::Delete,
KV_OPERATION_PURGE => Operation::Purge,
_ => Operation::Put,
};
let key = message
.subject
.strip_prefix(&self.prefix)
.map(|s| s.to_string())
.unwrap();
Poll::Ready(Some(Ok(Entry {
bucket: self.bucket.clone(),
key,
value: message.payload.to_vec(),
revision: info.stream_sequence,
created: info.published,
delta: info.pending,
operation,
})))
}
},
std::task::Poll::Pending => Poll::Pending,
}
}
fn size_hint(&self) -> (usize, Option<usize>) {
(0, None)
}
}
pub struct History<'a> {
subscription: super::consumer::push::Ordered<'a>,
done: bool,
prefix: String,
bucket: String,
}
impl<'a> futures::Stream for History<'a> {
type Item = Result<Entry, Error>;
fn poll_next(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<Option<Self::Item>> {
if self.done {
return Poll::Ready(None);
}
match self.subscription.poll_next_unpin(cx) {
Poll::Ready(message) => match message {
None => Poll::Ready(None),
Some(message) => {
let message = message?;
let info = message.info()?;
if info.pending == 0 {
self.done = true;
}
let operation = match message
.headers
.as_ref()
.and_then(|headers| headers.get(KV_OPERATION))
.unwrap_or(&HeaderValue::from(KV_OPERATION_PUT))
.iter()
.next()
.unwrap()
.as_str()
{
KV_OPERATION_DELETE => Operation::Delete,
KV_OPERATION_PURGE => Operation::Purge,
_ => Operation::Put,
};
let key = message
.subject
.strip_prefix(&self.prefix)
.map(|s| s.to_string())
.unwrap();
Poll::Ready(Some(Ok(Entry {
bucket: self.bucket.clone(),
key,
value: message.payload.to_vec(),
revision: info.stream_sequence,
created: info.published,
delta: info.pending,
operation,
})))
}
},
std::task::Poll::Pending => Poll::Pending,
}
}src/jetstream/consumer/push.rs (line 574)
474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611
fn poll_next(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
loop {
match self.shutdown.try_recv() {
Ok(err) => return Poll::Ready(Some(Err(err))),
Err(TryRecvError::Closed) => {
return Poll::Ready(Some(Err(Box::from(io::Error::new(
ErrorKind::Other,
"push consumer task closed",
)))))
}
Err(TryRecvError::Empty) => {}
}
if self.subscriber.is_none() {
match self.subscriber_future.as_mut() {
None => {
let context = self.context.clone();
let sequence = self.stream_sequence.clone();
let config = self.consumer.config.clone();
let stream_name = self.consumer.info.stream_name.clone();
self.subscriber_future = Some(Box::pin(async move {
recreate_consumer_and_subscription(
context,
config,
stream_name,
sequence.load(Ordering::Relaxed),
)
.await
}));
match self.subscriber_future.as_mut().unwrap().as_mut().poll(cx) {
Poll::Ready(subscriber) => {
self.subscriber_future = None;
self.subscriber = Some(subscriber?);
}
Poll::Pending => {
return Poll::Pending;
}
}
}
Some(subscriber) => match subscriber.as_mut().poll(cx) {
Poll::Ready(subscriber) => {
self.subscriber_future = None;
self.consumer_sequence.store(0, Ordering::Relaxed);
self.subscriber = Some(subscriber?);
}
Poll::Pending => {
return Poll::Pending;
}
},
}
}
if let Some(subscriber) = self.subscriber.as_mut() {
match subscriber.receiver.poll_recv(cx) {
Poll::Ready(maybe_message) => {
match maybe_message {
Some(message) => {
*self.last_seen.lock().unwrap() = Instant::now();
match message.status {
Some(StatusCode::IDLE_HEARTBEAT) => {
debug!("received idle heartbeats");
if let Some(headers) = message.headers.as_ref() {
if let Some(sequence) =
headers.get(crate::header::NATS_LAST_STREAM)
{
let sequence: u64 = sequence
.iter().next().unwrap()
.parse()
.map_err(|err|
Box::new(io::Error::new(
ErrorKind::Other,
format!("could not parse header into u64: {err}"))
))?;
if sequence
!= self.stream_sequence.load(Ordering::Relaxed)
{
self.subscriber = None;
}
}
}
if let Some(subject) = message.reply {
// TODO store pending_publish as a future and return errors from it
let client = self.context.client.clone();
tokio::task::spawn(async move {
client
.publish(subject, Bytes::from_static(b""))
.await
.unwrap();
});
}
continue;
}
Some(_) => {
continue;
}
None => {
let jetstream_message = jetstream::message::Message {
message,
context: self.context.clone(),
};
let info = jetstream_message.info()?;
trace!("consumer sequence: {:?}, stream sequence {:?}, consumer sequence in message: {:?} stream sequence in message: {:?}",
self.consumer_sequence,
self.stream_sequence,
info.consumer_sequence,
info.stream_sequence);
if info.consumer_sequence
!= self.consumer_sequence.load(Ordering::Relaxed) + 1
&& info.stream_sequence
!= self.stream_sequence.load(Ordering::Relaxed) + 1
{
debug!(
"ordered consumer mismatch. current {}, info: {}",
self.consumer_sequence.load(Ordering::Relaxed),
info.consumer_sequence
);
self.subscriber = None;
continue;
}
self.stream_sequence
.store(info.stream_sequence, Ordering::Relaxed);
self.consumer_sequence
.store(info.consumer_sequence, Ordering::Relaxed);
return Poll::Ready(Some(Ok(jetstream_message)));
}
}
}
None => {
debug!("received None from subscription");
return Poll::Ready(None);
}
}
}
Poll::Pending => return Poll::Pending,
}
}
}
}