fix(rust-client): async CA file read; typed ping/session-state/worker-info/drain-events

clients/rust/tests/client_behavior.rs

@@ -26,12 +26,12 @@ use zb_mom_ww_mxgateway_client::generated::mxaccess_gateway::v1::{
     AddItem2Reply, AddItemReply, AlarmConditionState, AlarmFeedMessage, AlarmTransitionKind,
     AuthenticateUserCommand, AuthenticateUserReply, BulkReadReply, BulkReadResult,
     BulkSubscribeReply, BulkWriteReply, BulkWriteResult, CloseSessionReply, CloseSessionRequest,
-    MxCommandKind, MxCommandReply, MxDataType, MxEvent, MxEventFamily, MxSparseArray,
-    MxSparseElement, MxStatusCategory, MxStatusProxy, MxStatusSource, MxValue,
+    DrainEventsReply, MxCommandKind, MxCommandReply, MxDataType, MxEvent, MxEventFamily,
+    MxSparseArray, MxSparseElement, MxStatusCategory, MxStatusProxy, MxStatusSource, MxValue,
     OnAlarmTransitionEvent, OpenSessionReply, OpenSessionRequest, ProtocolStatus,
     ProtocolStatusCode, QueryActiveAlarmsRequest, RegisterReply, ReplayGap, SessionState,
-    StreamAlarmsRequest, StreamEventsRequest, SubscribeResult, Write2BulkEntry, WriteBulkEntry,
-    WriteCommand, WriteSecured2BulkEntry, WriteSecuredBulkEntry,
+    SessionStateReply, StreamAlarmsRequest, StreamEventsRequest, SubscribeResult, WorkerInfoReply,
+    Write2BulkEntry, WriteBulkEntry, WriteCommand, WriteSecured2BulkEntry, WriteSecuredBulkEntry,
 };
 use zb_mom_ww_mxgateway_client::{
     next_correlation_id, ApiKey, ClientOptions, CommandError, Error, EventItem, GatewayClient,
@@ -550,6 +550,81 @@ async fn unary_invoke_maps_status_unavailable_to_error_unavailable() {
     );
 }
 
+#[tokio::test]
+async fn ping_round_trips_through_the_fake_gateway() {
+    let state = Arc::new(FakeState::default());
+    let endpoint = spawn_fake_gateway(state.clone()).await;
+    let client = GatewayClient::connect(ClientOptions::new(endpoint))
+        .await
+        .unwrap();
+    let session = client.session("session-fixture");
+
+    session.ping("hello").await.unwrap();
+
+    assert_eq!(
+        *state.last_command_kind.lock().await,
+        Some(MxCommandKind::Ping as i32)
+    );
+}
+
+#[tokio::test]
+async fn get_session_state_returns_typed_state() {
+    let state = Arc::new(FakeState::default());
+    let endpoint = spawn_fake_gateway(state.clone()).await;
+    let client = GatewayClient::connect(ClientOptions::new(endpoint))
+        .await
+        .unwrap();
+    let session = client.session("session-fixture");
+
+    let session_state = session.get_session_state().await.unwrap();
+
+    assert_eq!(session_state, SessionState::Ready);
+}
+
+#[tokio::test]
+async fn get_worker_info_returns_typed_reply() {
+    let state = Arc::new(FakeState::default());
+    let endpoint = spawn_fake_gateway(state.clone()).await;
+    let client = GatewayClient::connect(ClientOptions::new(endpoint))
+        .await
+        .unwrap();
+    let session = client.session("session-fixture");
+
+    let info = session.get_worker_info().await.unwrap();
+
+    assert_eq!(info.worker_process_id, 4242);
+    assert_eq!(info.worker_version, "1.2.3");
+}
+
+#[tokio::test]
+async fn drain_events_returns_events() {
+    let state = Arc::new(FakeState::default());
+    let endpoint = spawn_fake_gateway(state.clone()).await;
+    let client = GatewayClient::connect(ClientOptions::new(endpoint))
+        .await
+        .unwrap();
+    let session = client.session("session-fixture");
+
+    let events = session.drain_events(10).await.unwrap();
+
+    assert_eq!(events.len(), 1);
+}
+
+#[tokio::test]
+async fn get_session_state_malformed_without_payload() {
+    let state = Arc::new(FakeState::default());
+    let endpoint = spawn_fake_gateway(state.clone()).await;
+    let client = GatewayClient::connect(ClientOptions::new(endpoint))
+        .await
+        .unwrap();
+    let session = client.session("session-fixture");
+    *state.invoke_override.lock().await = Some(InvokeOverride::OkWithoutPayload);
+
+    let error = session.get_session_state().await.unwrap_err();
+
+    assert!(matches!(error, Error::MalformedReply { .. }));
+}
+
 #[tokio::test]
 async fn read_bulk_round_trips_through_the_fake_gateway() {
     let state = Arc::new(FakeState::default());
@@ -1337,6 +1412,57 @@ impl MxAccessGateway for FakeGateway {
             }));
         }
 
+        if kind == MxCommandKind::Ping as i32 {
+            return Ok(Response::new(MxCommandReply {
+                session_id: request.session_id,
+                correlation_id: "fake-correlation".to_owned(),
+                kind,
+                protocol_status: Some(ok_status("command ok")),
+                ..MxCommandReply::default()
+            }));
+        }
+
+        if kind == MxCommandKind::GetSessionState as i32 {
+            return Ok(Response::new(MxCommandReply {
+                session_id: request.session_id,
+                correlation_id: "fake-correlation".to_owned(),
+                kind,
+                protocol_status: Some(ok_status("command ok")),
+                payload: Some(mx_command_reply::Payload::SessionState(SessionStateReply {
+                    state: SessionState::Ready as i32,
+                })),
+                ..MxCommandReply::default()
+            }));
+        }
+
+        if kind == MxCommandKind::GetWorkerInfo as i32 {
+            return Ok(Response::new(MxCommandReply {
+                session_id: request.session_id,
+                correlation_id: "fake-correlation".to_owned(),
+                kind,
+                protocol_status: Some(ok_status("command ok")),
+                payload: Some(mx_command_reply::Payload::WorkerInfo(WorkerInfoReply {
+                    worker_process_id: 4242,
+                    worker_version: "1.2.3".to_owned(),
+                    ..WorkerInfoReply::default()
+                })),
+                ..MxCommandReply::default()
+            }));
+        }
+
+        if kind == MxCommandKind::DrainEvents as i32 {
+            return Ok(Response::new(MxCommandReply {
+                session_id: request.session_id,
+                correlation_id: "fake-correlation".to_owned(),
+                kind,
+                protocol_status: Some(ok_status("command ok")),
+                payload: Some(mx_command_reply::Payload::DrainEvents(DrainEventsReply {
+                    events: vec![MxEvent::default()],
+                })),
+                ..MxCommandReply::default()
+            }));
+        }
+
         if kind == MxCommandKind::ReadBulk as i32 {
             return Ok(Response::new(MxCommandReply {
                 session_id: request.session_id,