Struct AsyncUmaDbClient 

pub struct AsyncUmaDbClient { /* private fields */ }

Implementations

impl AsyncUmaDbClient

pub async fn connect( url: String, ca_path: Option<String>, batch_size: Option<u32>, api_key: Option<String>, ) -> DcbResult<Self>

pub async fn subscribe( &self, query: Option<DcbQuery>, after: Option<u64>, ) -> DcbResult<Box<dyn DcbSubscriptionAsync + Send + 'static>>

Examples found in repository

examples/client_async_insecure.rs (line 104)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Connect to the gRPC server
    let url = "http://localhost:50051".to_string();
    let client = UmaDbClient::new(url).connect_async().await?;

    // Define a consistency boundary
    let boundary = DcbQuery::new().item(
        DcbQueryItem::new()
            .types(["example"])
            .tags(["tag1", "tag2"]),
    );

    // Read events for a decision model
    let mut read_response = client
        .read(Some(boundary.clone()), None, false, None)
        .await?;

    // Build decision model
    while let Some(result) = read_response.next().await {
        match result {
            Ok(event) => {
                println!(
                    "Got event at position {}: {:?}",
                    event.position, event.event
                );
            }
            Err(status) => panic!("gRPC stream error: {}", status),
        }
    }

    // Remember the last-known position
    let last_known_position = read_response.head().await?;
    println!("Last known position is: {:?}", last_known_position);

    // Produce new event, attaching some metadata (e.g. provenance) that is
    // stored alongside the event and returned when it is read back.
    let event = DcbEvent::default()
        .event_type("example")
        .tags(["tag1", "tag2"])
        .data(b"Hello, world!")
        .uuid(Uuid::new_v4())
        .metadata_entry("source", "client_async_insecure")
        .metadata_entry("correlation_id", Uuid::new_v4().to_string());

    // Append event in consistency boundary
    let append_condition = DcbAppendCondition {
        fail_if_events_match: boundary.clone(),
        after: last_known_position,
    };
    let position1 = client
        .append(vec![event.clone()], Some(append_condition.clone()), None)
        .await?;

    println!("Appended event at position: {}", position1);

    // Append conflicting event - expect an error
    let conflicting_event = DcbEvent::default()
        .event_type("example")
        .tags(["tag1", "tag2"])
        .data(b"Hello, world!")
        .uuid(Uuid::new_v4()); // different UUID

    let conflicting_result = client
        .append(
            vec![conflicting_event.clone()],
            Some(append_condition.clone()),
            None,
        )
        .await;

    // Expect an integrity error
    match conflicting_result {
        Err(DcbError::IntegrityError(integrity_error)) => {
            println!("Conflicting event was rejected: {:?}", integrity_error);
        }
        other => panic!("Expected IntegrityError, got {:?}", other),
    }

    // Appending with identical events IDs and append conditions is idempotent.
    println!(
        "Retrying to append event at position: {:?}",
        last_known_position
    );
    let position2 = client
        .append(vec![event.clone()], Some(append_condition.clone()), None)
        .await?;

    if position1 == position2 {
        println!("Append method returned same commit position: {}", position2);
    } else {
        panic!("Expected idempotent retry!")
    }

    // Subscribe to all events for a projection
    let mut subscription = client.subscribe(None, None).await?;

    // Build an up-to-date view
    while let Some(result) = subscription.next().await {
        match result {
            Ok(ev) => {
                println!("Processing event at {}: {:?}", ev.position, ev.event);
                if ev.position == position2 {
                    println!("Projection has processed new event!");
                    break;
                }
            }
            Err(status) => panic!("gRPC stream error: {}", status),
        }
    }

    // Track an upstream position
    let upstream_position = client.get_tracking_info("upstream").await?;
    let next_upstream_position = upstream_position.unwrap_or(0) + 1;
    println!("Next upstream position: {next_upstream_position}");
    client
        .append(
            vec![],
            None,
            Some(TrackingInfo {
                source: "upstream".to_string(),
                position: next_upstream_position,
            }),
        )
        .await?;
    assert_eq!(
        next_upstream_position,
        client.get_tracking_info("upstream").await?.unwrap()
    );
    println!("Upstream position tracked okay!");

    // Try recording the same upstream position
    let conflicting_result = client
        .append(
            vec![],
            None,
            Some(TrackingInfo {
                source: "upstream".to_string(),
                position: next_upstream_position,
            }),
        )
        .await;

    // Expect an integrity error
    match conflicting_result {
        Err(DcbError::IntegrityError(integrity_error)) => {
            println!(
                "Conflicting upstream position was rejected: {:?}",
                integrity_error
            );
        }
        other => panic!("Expected IntegrityError, got {:?}", other),
    }

    Ok(())
}

examples/client_async_secure.rs (line 126)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Connect to the gRPC server
    let url = "https://localhost:50051".to_string();
    let client = UmaDbClient::new(url)
        .ca_path("server.pem".to_string()) // For self-signed server certificates.
        .api_key("umadb:example-api-key-4f7c2b1d9e5f4a038c1a".to_string())
        .connect_async()
        .await?;

    // Define a consistency boundary
    let cb = DcbQuery {
        items: vec![DcbQueryItem {
            types: vec!["example".to_string()],
            tags: vec!["tag1".to_string(), "tag2".to_string()],
        }],
    };

    // Read events for a decision model
    let mut read_response = client.read(Some(cb.clone()), None, false, None).await?;

    // Build decision model
    while let Some(result) = read_response.next().await {
        match result {
            Ok(event) => {
                println!(
                    "Got event at position {}: {:?}",
                    event.position, event.event
                );
            }
            Err(status) => panic!("gRPC stream error: {}", status),
        }
    }

    // Remember the last-known position
    let last_known_position = read_response.head().await?;
    println!("Last known position is: {:?}", last_known_position);

    // Produce new event, attaching some metadata (e.g. provenance) that is
    // stored alongside the event and returned when it is read back.
    let mut metadata = Vec::new();
    metadata.push(("source".to_string(), "client_async_secure".to_string()));
    metadata.push(("correlation_id".to_string(), Uuid::new_v4().to_string()));
    let event = DcbEvent {
        event_type: "example".to_string(),
        tags: vec!["tag1".to_string(), "tag2".to_string()],
        data: b"Hello, world!".to_vec(),
        uuid: Some(Uuid::new_v4()),
        metadata,
    };

    // Append event in consistency boundary
    let commit_position1 = client
        .append(
            vec![event.clone()],
            Some(DcbAppendCondition {
                fail_if_events_match: cb.clone(),
                after: last_known_position,
            }),
            None,
        )
        .await?;
    println!("Appended event at position: {}", commit_position1);

    // Append conflicting event - expect an error
    let conflicting_event = DcbEvent {
        event_type: "example".to_string(),
        tags: vec!["tag1".to_string(), "tag2".to_string()],
        data: b"Hello, world!".to_vec(),
        uuid: Some(Uuid::new_v4()), // different UUID
        metadata: Vec::new(),
    };
    let conflicting_result = client
        .append(
            vec![conflicting_event.clone()],
            Some(DcbAppendCondition {
                fail_if_events_match: cb.clone(),
                after: last_known_position,
            }),
            None,
        )
        .await;

    // Expect an integrity error
    match conflicting_result {
        Err(DcbError::IntegrityError(integrity_error)) => {
            println!("Conflicting event was rejected: {:?}", integrity_error);
        }
        other => panic!("Expected IntegrityError, got {:?}", other),
    }

    // Conditional appends with event UUIDs are idempotent.
    println!(
        "Retrying to append event at position: {:?}",
        last_known_position
    );
    let commit_position2 = client
        .append(
            vec![event.clone()],
            Some(DcbAppendCondition {
                fail_if_events_match: cb.clone(),
                after: last_known_position,
            }),
            None,
        )
        .await?;

    if commit_position1 == commit_position2 {
        println!(
            "Append method returned same commit position: {}",
            commit_position2
        );
    } else {
        panic!("Expected idempotent retry!")
    }

    // Subscribe to all events for a projection
    let mut subscription = client.subscribe(None, None).await?;

    // Build an up-to-date view
    while let Some(result) = subscription.next().await {
        match result {
            Ok(ev) => {
                println!("Processing event at {}: {:?}", ev.position, ev.event);
                if ev.position == commit_position2 {
                    println!("Projection has processed new event!");
                    break;
                }
            }
            Err(status) => panic!("gRPC stream error: {}", status),
        }
    }

    // Track an upstream position
    let upstream_position = client.get_tracking_info("upstream").await?;
    let next_upstream_position = upstream_position.unwrap_or(0) + 1;
    println!("Next upstream position: {next_upstream_position}");
    client
        .append(
            vec![],
            None,
            Some(TrackingInfo {
                source: "upstream".to_string(),
                position: next_upstream_position,
            }),
        )
        .await?;
    assert_eq!(
        next_upstream_position,
        client.get_tracking_info("upstream").await?.unwrap()
    );
    println!("Upstream position tracked okay!");

    // Try recording the same upstream position
    let conflicting_result = client
        .append(
            vec![],
            None,
            Some(TrackingInfo {
                source: "upstream".to_string(),
                position: next_upstream_position,
            }),
        )
        .await;

    // Expect an integrity error
    match conflicting_result {
        Err(DcbError::IntegrityError(integrity_error)) => {
            println!(
                "Conflicting upstream position was rejected: {:?}",
                integrity_error
            );
        }
        other => panic!("Expected IntegrityError, got {:?}", other),
    }

    Ok(())
}

pub fn close(&self)

Stops all streaming responses opened by this client.

This affects only streams opened via this client instance’s read() and subscribe() methods; it does not stop or otherwise affect the server, nor streams opened by other clients.

pub async fn connect_with_tls_options( url: String, tls_options: Option<ClientTlsOptions>, batch_size: Option<u32>, api_key: Option<String>, ) -> DcbResult<Self>

Trait Implementations

impl DcbEventStoreAsync for AsyncUmaDbClient

fn read<'a, 'async_trait>( &'a self, query: Option<DcbQuery>, start: Option<u64>, backwards: bool, limit: Option<u32>, ) -> Pin<Box<dyn Future<Output = DcbResult<Box<dyn DcbReadResponseAsync + Send + 'static>>> + Send + 'async_trait>>

where Self: 'async_trait, 'a: 'async_trait,

Reads events from the store based on the provided query and constraints

fn head<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = DcbResult<Option<u64>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Returns the current head position of the event store, or None if empty

fn get_tracking_info<'life0, 'life1, 'async_trait>( &'life0 self, source: &'life1 str, ) -> Pin<Box<dyn Future<Output = DcbResult<Option<u64>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Returns the greatest recorded upstream position for a tracking source, if any

fn append<'life0, 'async_trait>( &'life0 self, events: Vec<DcbEvent>, condition: Option<DcbAppendCondition>, tracking_info: Option<TrackingInfo>, ) -> Pin<Box<dyn Future<Output = DcbResult<u64>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Appends given events to the event store, unless the condition fails

fn read_with_head<'a, 'async_trait>( &'a self, query: Option<DcbQuery>, after: Option<u64>, backwards: bool, limit: Option<u32>, ) -> Pin<Box<dyn Future<Output = Result<(Vec<DcbSequencedEvent>, Option<u64>), DcbError>> + Send + 'async_trait>>

where 'a: 'async_trait, Self: 'async_trait,

Reads events from the store and returns them as a tuple of (Vec<DcbSequencedEvent>, Option<u64>)

impl Drop for AsyncUmaDbClient

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.