Enum SubscribeMonotonic

pub enum SubscribeMonotonic<'a> {
    Initial {
        client: &'a Ax,
        request: SubscribeMonotonicRequest,
    },
    Pending(BoxFuture<'a, Result<BoxStream<'static, SubscribeMonotonicResponse>>>),
    Void,
}

Request builder for monotonic subscriptions.

Monotonic subscriptions keep track of the highest sort order (LamportTimestamp and StreamId) seen so far, ending the stream with a SubscribeMonotonicResponse::TimeTravel message if the next event would be out of order.

Warning: SubscribeMonotonic implements the Future trait, as such it can be polled. Calling any SubscribeMonotonic function after polling will result in a panic!

Variants

Initial

Fields

client: &'a Ax

request: SubscribeMonotonicRequest

Pending(BoxFuture<'a, Result<BoxStream<'static, SubscribeMonotonicResponse>>>)

Void

Implementations

impl<'a> SubscribeMonotonic<'a>

pub fn with_lower_bound(self, lower_bound: OffsetMap) -> Self

Add a (exclusive) lower bound to the subscription query.

For more information on offsets, as well as lower and upper bounds refer to the offsets and partitions documentation page.

The lower bound limits the start of the query events. As an example, consider the following (example) events:

{ "offset": 1, "event": { "temperature": 10 } }
{ "offset": 3, "event": { "temperature": 12 } }
{ "offset": 14, "event": { "temperature": 9 } }

If you set the lower bound to 10, the first event to be returned would be the last of the example.

Panics

Calling this function after polling Subscribe will result in a panic.

Example

use ax_sdk::{Ax, AxOpts};
use futures::stream::StreamExt;
async fn lower_bound_example() {
    let service = Ax::new(AxOpts::default()).await.unwrap();
    let present_offsets = service.offsets().await.unwrap().present;
    let mut response = service.subscribe("FROM allEvents")
        .with_lower_bound(present_offsets)
        .await
        .unwrap();
    while let Some(event) = response.next().await {
        println!("{:?}", event);
    }
}

Generating an OffsetMap out of thin air is usually not possible because they require stream IDs — which require knowledge of the streams and so on. Hence, a more involved and useful example requires you to perform a query to get an offset map when the query finishes streaming all results.

use ax_sdk::{types::{tags, Offset, service::QueryResponse}, Ax, AxOpts};
use futures::stream::StreamExt;
async fn lower_bound_example() {
    let service = Ax::new(AxOpts::default()).await.unwrap();
    // We're publishing events for a completely functional example
    let publish_response = service
        .publish()
        .event(
            tags!("temperature", "sensor:temp-sensor1"),
            &serde_json::json!({ "temperature": 10 }),
        ).unwrap()
        .event(
            tags!("temperature", "sensor:temp-sensor2"),
            &serde_json::json!({ "temperature": 21 }),
        ).unwrap()
        .event(
            tags!("temperature", "sensor:temp-sensor3"),
            &serde_json::json!({ "temperature": 40 }),
        ).unwrap()
        .await.unwrap();
    // Query for the "halfway" event
    let mut query_response = service
        .query("FROM 'sensor:temp-sensor2'")
        .await
        .unwrap();
    // This loop is a dirty hack for demonstration purposes
    // in real world usage you will most likely be using the events
    // and keeping the offset map in the end.
    let offsets = loop {
        let result = query_response.next().await.unwrap();
        if let QueryResponse::Offsets(offsets) = result {
            break offsets.offsets;
        }
    };
    // Subcribe for all 'temperature' events with the previous query `OffsetMap`
    // as a lower bound. We're expecting to only see events after the "halfway"
    // event — {"temperature"}
    let mut subscribe_response = service
        .subscribe_monotonic("FROM 'temperature'")
        .with_lower_bound(offsets.clone())
        .await.unwrap();
    while let Some(response) = query_response.next().await {
        println!("{:?}", response);
    }
}

Trait Implementations

impl<'a> FusedFuture for SubscribeMonotonic<'a>

fn is_terminated(&self) -> bool

Returns true if the underlying future should no longer be polled.

impl<'a> Future for SubscribeMonotonic<'a>

type Output = Result<Pin<Box<dyn Stream<Item = SubscribeMonotonicResponse> + Send>>, Error>

The type of value produced on completion.

fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>

Attempts to resolve the future to a final value, registering the current task for wakeup if the value is not yet available.