Enum Subscribe

pub enum Subscribe<'a> {
    Initial {
        client: &'a Ax,
        request: SubscribeRequest,
    },
    Pending(BoxFuture<'a, Result<BoxStream<'static, SubscribeResponse>>>),
    Void,
}

Request builder for subscriptions.

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

Variants

Initial

Fields

client: &'a Ax

request: SubscribeRequest

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

Void

Implementations

impl<'a> Subscribe<'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("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 Subscribe<'a>

fn is_terminated(&self) -> bool

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

impl<'a> Future for Subscribe<'a>

type Output = Result<Pin<Box<dyn Stream<Item = SubscribeResponse> + 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.