Struct Conn 

pub struct Conn { /* private fields */ }

A live, namespace-bound connection. Cheap to clone, clones share one HTTP/2 channel, which is what makes multi-namespace fan-out cheap.

Implementations

impl Conn

pub async fn connect(profile: &ProfileRef) -> Result<Self, ConnectError>

Examples found in repository

examples/spike.rs (line 33)

async fn main() -> anyhow::Result<()> {
    let profile = ProfileRef {
        name: std::env::args().nth(1),
        config_file: None,
    };

    let conn = Conn::connect(&profile).await?;
    println!(
        "connected  profile={}  address={}  namespace={}",
        conn.profile(),
        conn.address(),
        conn.namespace()
    );

    let mut wf = conn.wf();
    let ns = wf
        .list_namespaces(Request::new(ListNamespacesRequest {
            page_size: 50,
            ..Default::default()
        }))
        .await?
        .into_inner();
    println!("\nnamespaces ({}):", ns.namespaces.len());
    for n in &ns.namespaces {
        if let Some(info) = &n.namespace_info {
            println!("  - {}", info.name);
        }
    }

    let count = wf
        .count_workflow_executions(Request::new(CountWorkflowExecutionsRequest {
            namespace: conn.namespace().to_string(),
            query: String::new(),
        }))
        .await?
        .into_inner();
    println!(
        "\ntotal workflows in `{}`: {}",
        conn.namespace(),
        count.count
    );

    let list = wf
        .list_workflow_executions(Request::new(ListWorkflowExecutionsRequest {
            namespace: conn.namespace().to_string(),
            page_size: 10,
            query: String::new(),
            ..Default::default()
        }))
        .await?
        .into_inner();

    println!("\nworkflows ({} shown):", list.executions.len());
    let mut first: Option<WorkflowExecution> = None;
    for e in &list.executions {
        let exec = e.execution.clone().unwrap_or_default();
        println!(
            "  {:<10} {:<28} {}",
            format!("{:?}", e.status()),
            e.r#type.as_ref().map(|t| t.name.as_str()).unwrap_or("?"),
            exec.workflow_id
        );
        first.get_or_insert(exec);
    }
    println!("  next_page_token: {} bytes", list.next_page_token.len());

    if let Some(exec) = first {
        let hist = wf
            .get_workflow_execution_history(Request::new(GetWorkflowExecutionHistoryRequest {
                namespace: conn.namespace().to_string(),
                execution: Some(exec.clone()),
                maximum_page_size: 100,
                // `wait_new_event: true` is what turns this into `tail -f`. Left false
                // here so the spike terminates.
                wait_new_event: false,
                history_event_filter_type: HistoryEventFilterType::AllEvent as i32,
                ..Default::default()
            }))
            .await?
            .into_inner();

        let events = hist.history.map(|h| h.events).unwrap_or_default();
        println!(
            "\nhistory of {} ({} events):",
            exec.workflow_id,
            events.len()
        );
        for ev in events.iter().take(15) {
            println!("  {:>4}  {:?}", ev.event_id, ev.event_type());
        }
    }

    Ok(())
}

pub fn namespace(&self) -> &str

Examples found in repository

examples/spike.rs (line 38)

async fn main() -> anyhow::Result<()> {
    let profile = ProfileRef {
        name: std::env::args().nth(1),
        config_file: None,
    };

    let conn = Conn::connect(&profile).await?;
    println!(
        "connected  profile={}  address={}  namespace={}",
        conn.profile(),
        conn.address(),
        conn.namespace()
    );

    let mut wf = conn.wf();
    let ns = wf
        .list_namespaces(Request::new(ListNamespacesRequest {
            page_size: 50,
            ..Default::default()
        }))
        .await?
        .into_inner();
    println!("\nnamespaces ({}):", ns.namespaces.len());
    for n in &ns.namespaces {
        if let Some(info) = &n.namespace_info {
            println!("  - {}", info.name);
        }
    }

    let count = wf
        .count_workflow_executions(Request::new(CountWorkflowExecutionsRequest {
            namespace: conn.namespace().to_string(),
            query: String::new(),
        }))
        .await?
        .into_inner();
    println!(
        "\ntotal workflows in `{}`: {}",
        conn.namespace(),
        count.count
    );

    let list = wf
        .list_workflow_executions(Request::new(ListWorkflowExecutionsRequest {
            namespace: conn.namespace().to_string(),
            page_size: 10,
            query: String::new(),
            ..Default::default()
        }))
        .await?
        .into_inner();

    println!("\nworkflows ({} shown):", list.executions.len());
    let mut first: Option<WorkflowExecution> = None;
    for e in &list.executions {
        let exec = e.execution.clone().unwrap_or_default();
        println!(
            "  {:<10} {:<28} {}",
            format!("{:?}", e.status()),
            e.r#type.as_ref().map(|t| t.name.as_str()).unwrap_or("?"),
            exec.workflow_id
        );
        first.get_or_insert(exec);
    }
    println!("  next_page_token: {} bytes", list.next_page_token.len());

    if let Some(exec) = first {
        let hist = wf
            .get_workflow_execution_history(Request::new(GetWorkflowExecutionHistoryRequest {
                namespace: conn.namespace().to_string(),
                execution: Some(exec.clone()),
                maximum_page_size: 100,
                // `wait_new_event: true` is what turns this into `tail -f`. Left false
                // here so the spike terminates.
                wait_new_event: false,
                history_event_filter_type: HistoryEventFilterType::AllEvent as i32,
                ..Default::default()
            }))
            .await?
            .into_inner();

        let events = hist.history.map(|h| h.events).unwrap_or_default();
        println!(
            "\nhistory of {} ({} events):",
            exec.workflow_id,
            events.len()
        );
        for ev in events.iter().take(15) {
            println!("  {:>4}  {:?}", ev.event_id, ev.event_type());
        }
    }

    Ok(())
}

pub fn profile(&self) -> &str

Examples found in repository

examples/spike.rs (line 36)

async fn main() -> anyhow::Result<()> {
    let profile = ProfileRef {
        name: std::env::args().nth(1),
        config_file: None,
    };

    let conn = Conn::connect(&profile).await?;
    println!(
        "connected  profile={}  address={}  namespace={}",
        conn.profile(),
        conn.address(),
        conn.namespace()
    );

    let mut wf = conn.wf();
    let ns = wf
        .list_namespaces(Request::new(ListNamespacesRequest {
            page_size: 50,
            ..Default::default()
        }))
        .await?
        .into_inner();
    println!("\nnamespaces ({}):", ns.namespaces.len());
    for n in &ns.namespaces {
        if let Some(info) = &n.namespace_info {
            println!("  - {}", info.name);
        }
    }

    let count = wf
        .count_workflow_executions(Request::new(CountWorkflowExecutionsRequest {
            namespace: conn.namespace().to_string(),
            query: String::new(),
        }))
        .await?
        .into_inner();
    println!(
        "\ntotal workflows in `{}`: {}",
        conn.namespace(),
        count.count
    );

    let list = wf
        .list_workflow_executions(Request::new(ListWorkflowExecutionsRequest {
            namespace: conn.namespace().to_string(),
            page_size: 10,
            query: String::new(),
            ..Default::default()
        }))
        .await?
        .into_inner();

    println!("\nworkflows ({} shown):", list.executions.len());
    let mut first: Option<WorkflowExecution> = None;
    for e in &list.executions {
        let exec = e.execution.clone().unwrap_or_default();
        println!(
            "  {:<10} {:<28} {}",
            format!("{:?}", e.status()),
            e.r#type.as_ref().map(|t| t.name.as_str()).unwrap_or("?"),
            exec.workflow_id
        );
        first.get_or_insert(exec);
    }
    println!("  next_page_token: {} bytes", list.next_page_token.len());

    if let Some(exec) = first {
        let hist = wf
            .get_workflow_execution_history(Request::new(GetWorkflowExecutionHistoryRequest {
                namespace: conn.namespace().to_string(),
                execution: Some(exec.clone()),
                maximum_page_size: 100,
                // `wait_new_event: true` is what turns this into `tail -f`. Left false
                // here so the spike terminates.
                wait_new_event: false,
                history_event_filter_type: HistoryEventFilterType::AllEvent as i32,
                ..Default::default()
            }))
            .await?
            .into_inner();

        let events = hist.history.map(|h| h.events).unwrap_or_default();
        println!(
            "\nhistory of {} ({} events):",
            exec.workflow_id,
            events.len()
        );
        for ev in events.iter().take(15) {
            println!("  {:>4}  {:?}", ev.event_id, ev.event_type());
        }
    }

    Ok(())
}

pub fn address(&self) -> &str

The frontend this is connected to, as a URL. Never carries credentials: an API key lives in the connection options, not the target.

Examples found in repository

examples/spike.rs (line 37)

async fn main() -> anyhow::Result<()> {
    let profile = ProfileRef {
        name: std::env::args().nth(1),
        config_file: None,
    };

    let conn = Conn::connect(&profile).await?;
    println!(
        "connected  profile={}  address={}  namespace={}",
        conn.profile(),
        conn.address(),
        conn.namespace()
    );

    let mut wf = conn.wf();
    let ns = wf
        .list_namespaces(Request::new(ListNamespacesRequest {
            page_size: 50,
            ..Default::default()
        }))
        .await?
        .into_inner();
    println!("\nnamespaces ({}):", ns.namespaces.len());
    for n in &ns.namespaces {
        if let Some(info) = &n.namespace_info {
            println!("  - {}", info.name);
        }
    }

    let count = wf
        .count_workflow_executions(Request::new(CountWorkflowExecutionsRequest {
            namespace: conn.namespace().to_string(),
            query: String::new(),
        }))
        .await?
        .into_inner();
    println!(
        "\ntotal workflows in `{}`: {}",
        conn.namespace(),
        count.count
    );

    let list = wf
        .list_workflow_executions(Request::new(ListWorkflowExecutionsRequest {
            namespace: conn.namespace().to_string(),
            page_size: 10,
            query: String::new(),
            ..Default::default()
        }))
        .await?
        .into_inner();

    println!("\nworkflows ({} shown):", list.executions.len());
    let mut first: Option<WorkflowExecution> = None;
    for e in &list.executions {
        let exec = e.execution.clone().unwrap_or_default();
        println!(
            "  {:<10} {:<28} {}",
            format!("{:?}", e.status()),
            e.r#type.as_ref().map(|t| t.name.as_str()).unwrap_or("?"),
            exec.workflow_id
        );
        first.get_or_insert(exec);
    }
    println!("  next_page_token: {} bytes", list.next_page_token.len());

    if let Some(exec) = first {
        let hist = wf
            .get_workflow_execution_history(Request::new(GetWorkflowExecutionHistoryRequest {
                namespace: conn.namespace().to_string(),
                execution: Some(exec.clone()),
                maximum_page_size: 100,
                // `wait_new_event: true` is what turns this into `tail -f`. Left false
                // here so the spike terminates.
                wait_new_event: false,
                history_event_filter_type: HistoryEventFilterType::AllEvent as i32,
                ..Default::default()
            }))
            .await?
            .into_inner();

        let events = hist.history.map(|h| h.events).unwrap_or_default();
        println!(
            "\nhistory of {} ({} events):",
            exec.workflow_id,
            events.len()
        );
        for ev in events.iter().take(15) {
            println!("  {:>4}  {:?}", ev.event_id, ev.event_type());
        }
    }

    Ok(())
}

pub fn wf(&self) -> Box<dyn WorkflowService>

Raw WorkflowService. Requests take tonic::Request<T> and the connection’s retry policy is already applied underneath.

Examples found in repository

examples/spike.rs (line 41)

async fn main() -> anyhow::Result<()> {
    let profile = ProfileRef {
        name: std::env::args().nth(1),
        config_file: None,
    };

    let conn = Conn::connect(&profile).await?;
    println!(
        "connected  profile={}  address={}  namespace={}",
        conn.profile(),
        conn.address(),
        conn.namespace()
    );

    let mut wf = conn.wf();
    let ns = wf
        .list_namespaces(Request::new(ListNamespacesRequest {
            page_size: 50,
            ..Default::default()
        }))
        .await?
        .into_inner();
    println!("\nnamespaces ({}):", ns.namespaces.len());
    for n in &ns.namespaces {
        if let Some(info) = &n.namespace_info {
            println!("  - {}", info.name);
        }
    }

    let count = wf
        .count_workflow_executions(Request::new(CountWorkflowExecutionsRequest {
            namespace: conn.namespace().to_string(),
            query: String::new(),
        }))
        .await?
        .into_inner();
    println!(
        "\ntotal workflows in `{}`: {}",
        conn.namespace(),
        count.count
    );

    let list = wf
        .list_workflow_executions(Request::new(ListWorkflowExecutionsRequest {
            namespace: conn.namespace().to_string(),
            page_size: 10,
            query: String::new(),
            ..Default::default()
        }))
        .await?
        .into_inner();

    println!("\nworkflows ({} shown):", list.executions.len());
    let mut first: Option<WorkflowExecution> = None;
    for e in &list.executions {
        let exec = e.execution.clone().unwrap_or_default();
        println!(
            "  {:<10} {:<28} {}",
            format!("{:?}", e.status()),
            e.r#type.as_ref().map(|t| t.name.as_str()).unwrap_or("?"),
            exec.workflow_id
        );
        first.get_or_insert(exec);
    }
    println!("  next_page_token: {} bytes", list.next_page_token.len());

    if let Some(exec) = first {
        let hist = wf
            .get_workflow_execution_history(Request::new(GetWorkflowExecutionHistoryRequest {
                namespace: conn.namespace().to_string(),
                execution: Some(exec.clone()),
                maximum_page_size: 100,
                // `wait_new_event: true` is what turns this into `tail -f`. Left false
                // here so the spike terminates.
                wait_new_event: false,
                history_event_filter_type: HistoryEventFilterType::AllEvent as i32,
                ..Default::default()
            }))
            .await?
            .into_inner();

        let events = hist.history.map(|h| h.events).unwrap_or_default();
        println!(
            "\nhistory of {} ({} events):",
            exec.workflow_id,
            events.len()
        );
        for ev in events.iter().take(15) {
            println!("  {:>4}  {:?}", ev.event_id, ev.event_type());
        }
    }

    Ok(())
}

pub fn operator(&self) -> Box<dyn OperatorService>

pub fn cloud(&self) -> Box<dyn CloudService>

pub fn raw(&self) -> &Client

The high-level client, for the handful of operations where temporalio-client already does the assembly work for us (schedules, workflow handles).

impl Conn

pub async fn follow_history( &self, namespace: &str, workflow_id: &str, run_id: &str, next_page_token: Vec<u8>, ) -> Result<HistoryPage, OpError>

One long-poll step of follow mode.

This is Conn::get_history with wait_new_event: true, which makes the call block until the workflow does something, up to about a minute, then it returns empty-handed and you call again. Never reach for this outside a task dedicated to following; anything else it is on will simply stop.

The behaviour of the continuation token differs from paging, and follow mode is built on the difference. Measured against a dev server:

	wait_new_event: false	wait_new_event: true
running workflow, caught up	returns 0 events, empty token	blocks, then returns new events, token stays non-empty
closed workflow	empty token	empty token, terminal event last

So an empty token here means the workflow has closed and there is nothing further to follow: the loop’s termination condition, and it is authoritative in a way that inspecting the last event’s type is not.

Passing an empty token restarts from event 1, so a caller resuming a follow should hand back the last non-empty token it saw. The page that token sits in is replayed, which is why events are merged rather than appended, see tmprl_core::history::merge_events.

pub async fn get_history( &self, namespace: &str, workflow_id: &str, run_id: &str, page_size: i32, next_page_token: Vec<u8>, ) -> Result<HistoryPage, OpError>

One page of history, normalised.

wait_new_event is false here. Setting it true turns this into a long poll that does not return until something happens, which is correct for follow mode and a hang everywhere else, so follow mode gets Conn::follow_history rather than a flag on this one that is easy to pass by accident.

impl Conn

pub async fn mutate(&self, m: &Mutation) -> Result<(), OpError>

Carry out a confirmed mutation.

One entry point rather than four, so the reducer has exactly one place that writes and the audit log has exactly one thing to wrap.

impl Conn

pub async fn list_namespaces(&self) -> Result<Vec<NamespaceInfo>, OpError>

Every namespace on the cluster, paged to exhaustion.

Namespace counts are small (tens, not thousands), so this collects rather than streaming. Workflow listing will not be able to do that.

impl Conn

pub async fn list_schedules( &self, namespace: &str, page_size: i32, next_page_token: Vec<u8>, ) -> Result<SchedulePage, OpError>

impl Conn

pub async fn list_workflows( &self, namespace: &str, query: &str, page_size: i32, next_page_token: Vec<u8>, ) -> Result<WorkflowPage, OpError>

One page of executions matching query, newest first.

This deliberately does not page to exhaustion the way list_namespaces does. A namespace list is tens of rows; a workflow list is unbounded, and draining it would hang the interface on any real cluster.

pub async fn list_workflows_across( &self, namespaces: &[String], query: &str, page_size: i32, ) -> Result<(Vec<WorkflowRow>, Continuation), OpError>

The first page from several namespaces at once, merged newest-first.

Returned alongside the rows is a continuation token per namespace, because the namespaces exhaust at different points and one merged token cannot express that. A namespace that is already finished simply does not appear in the returned list. Feed that list to Conn::continue_workflows_across for the next page.

pub async fn continue_workflows_across( &self, tokens: &Continuation, query: &str, page_size: i32, ) -> Result<(Vec<WorkflowRow>, Continuation), OpError>

The next page, asking only the namespaces that still have one.

Taking the token list rather than the namespace list is the point. Re-deriving the namespaces from the original scope would hand an exhausted namespace an empty token, which the server reads as “start from the beginning”, so it would return page one again, along with a fresh token, and that namespace would never finish.

pub async fn count_workflows_by_status( &self, namespace: &str, query: &str, ) -> Result<StatusCounts, OpError>

Per-status counts for the list header.

One GROUP BY call rather than one call per status. The grouped counts are approximate by design on large clusters (Temporal says so) which is why the total is taken from the response rather than summed from the groups.

impl Conn

pub async fn count_workflows_across( &self, namespaces: &[String], query: &str, ) -> Result<StatusCounts, OpError>

Header counts summed over a fan-out.

A header that counted only the first of several namespaces would be quietly wrong, which is worse than having no header at all.

Trait Implementations

impl Clone for Conn

fn clone(&self) -> Conn

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.