mxaccessgw/clients/rust/src/session.rs

use crate::client::{EventStream, GatewayClient};
use crate::error::Error;
use crate::generated::mxaccess_gateway::v1::mx_command::Payload;
use crate::generated::mxaccess_gateway::v1::mx_command_reply;
use crate::generated::mxaccess_gateway::v1::{
    AddItem2Command, AddItemCommand, AdviseCommand, CloseSessionRequest, MxCommand, MxCommandKind,
    MxCommandReply, MxCommandRequest, MxValue as ProtoMxValue, OpenSessionRequest, RegisterCommand,
    StreamEventsRequest, Write2Command, WriteCommand,
};
use crate::value::MxValue;

/// Session identifier returned by the gateway.
#[derive(Clone)]
pub struct Session {
    id: String,
    client: GatewayClient,
}

impl Session {
    pub(crate) fn new(id: impl Into<String>, client: GatewayClient) -> Self {
        Self {
            id: id.into(),
            client,
        }
    }

    pub fn id(&self) -> &str {
        &self.id
    }

    pub async fn open(client: GatewayClient, client_session_name: &str) -> Result<Self, Error> {
        client
            .open_session(OpenSessionRequest {
                client_session_name: client_session_name.to_owned(),
                ..OpenSessionRequest::default()
            })
            .await
    }

    pub async fn close(&self) -> Result<(), Error> {
        self.client
            .close_session_raw(CloseSessionRequest {
                session_id: self.id.clone(),
                client_correlation_id: "rust-client-close-session".to_owned(),
            })
            .await?;
        Ok(())
    }

    pub async fn register(&self, client_name: &str) -> Result<i32, Error> {
        let reply = self
            .invoke(
                MxCommandKind::Register,
                Payload::Register(RegisterCommand {
                    client_name: client_name.to_owned(),
                }),
            )
            .await?;

        Ok(register_server_handle(&reply))
    }

    pub async fn add_item(&self, server_handle: i32, item_definition: &str) -> Result<i32, Error> {
        let reply = self
            .invoke(
                MxCommandKind::AddItem,
                Payload::AddItem(AddItemCommand {
                    server_handle,
                    item_definition: item_definition.to_owned(),
                }),
            )
            .await?;

        Ok(add_item_handle(&reply))
    }

    pub async fn add_item2(
        &self,
        server_handle: i32,
        item_definition: &str,
        item_context: &str,
    ) -> Result<i32, Error> {
        let reply = self
            .invoke(
                MxCommandKind::AddItem2,
                Payload::AddItem2(AddItem2Command {
                    server_handle,
                    item_definition: item_definition.to_owned(),
                    item_context: item_context.to_owned(),
                }),
            )
            .await?;

        Ok(add_item2_handle(&reply))
    }

    pub async fn advise(&self, server_handle: i32, item_handle: i32) -> Result<(), Error> {
        self.invoke(
            MxCommandKind::Advise,
            Payload::Advise(AdviseCommand {
                server_handle,
                item_handle,
            }),
        )
        .await?;
        Ok(())
    }

    pub async fn write(
        &self,
        server_handle: i32,
        item_handle: i32,
        value: MxValue,
        user_id: i32,
    ) -> Result<(), Error> {
        self.invoke(
            MxCommandKind::Write,
            Payload::Write(WriteCommand {
                server_handle,
                item_handle,
                value: Some(value.into_proto()),
                user_id,
            }),
        )
        .await?;
        Ok(())
    }

    pub async fn write2(
        &self,
        server_handle: i32,
        item_handle: i32,
        value: MxValue,
        timestamp_value: MxValue,
        user_id: i32,
    ) -> Result<(), Error> {
        self.invoke(
            MxCommandKind::Write2,
            Payload::Write2(Write2Command {
                server_handle,
                item_handle,
                value: Some(value.into_proto()),
                timestamp_value: Some(timestamp_value.into_proto()),
                user_id,
            }),
        )
        .await?;
        Ok(())
    }

    pub async fn events(&self) -> Result<EventStream, Error> {
        self.events_after(0).await
    }

    pub async fn events_after(&self, after_worker_sequence: u64) -> Result<EventStream, Error> {
        self.client
            .stream_events(StreamEventsRequest {
                session_id: self.id.clone(),
                after_worker_sequence,
            })
            .await
    }

    pub async fn invoke_raw(
        &self,
        kind: MxCommandKind,
        payload: Payload,
    ) -> Result<MxCommandReply, Error> {
        self.client
            .invoke_raw(self.command_request(kind, payload))
            .await
    }

    pub async fn invoke(
        &self,
        kind: MxCommandKind,
        payload: Payload,
    ) -> Result<MxCommandReply, Error> {
        self.client
            .invoke(self.command_request(kind, payload))
            .await
    }

    fn command_request(&self, kind: MxCommandKind, payload: Payload) -> MxCommandRequest {
        MxCommandRequest {
            session_id: self.id.clone(),
            client_correlation_id: format!("rust-client-{}", kind.as_str_name()),
            command: Some(MxCommand {
                kind: kind as i32,
                payload: Some(payload),
            }),
        }
    }
}

fn register_server_handle(reply: &MxCommandReply) -> i32 {
    match reply.payload.as_ref() {
        Some(mx_command_reply::Payload::Register(register)) => register.server_handle,
        _ => reply
            .return_value
            .as_ref()
            .and_then(int32_reply_value)
            .unwrap_or_default(),
    }
}

fn add_item_handle(reply: &MxCommandReply) -> i32 {
    match reply.payload.as_ref() {
        Some(mx_command_reply::Payload::AddItem(add_item)) => add_item.item_handle,
        _ => reply
            .return_value
            .as_ref()
            .and_then(int32_reply_value)
            .unwrap_or_default(),
    }
}

fn add_item2_handle(reply: &MxCommandReply) -> i32 {
    match reply.payload.as_ref() {
        Some(mx_command_reply::Payload::AddItem2(add_item)) => add_item.item_handle,
        _ => reply
            .return_value
            .as_ref()
            .and_then(int32_reply_value)
            .unwrap_or_default(),
    }
}

fn int32_reply_value(value: &ProtoMxValue) -> Option<i32> {
    match value.kind.as_ref()? {
        crate::generated::mxaccess_gateway::v1::mx_value::Kind::Int32Value(value) => Some(*value),
        _ => None,
    }
}