WIP Experiment implementation of FOP-1

Cargo.toml

@@ -25,7 +25,7 @@ license = "MPL-2.0"
 
 [workspace.dependencies]
 structpack = "1.0"
-gaia-stub = "1.0"
-gaia-ccsds-c2a = "1.0"
-gaia-tmtc = "1.0"
+gaia-stub = { path = "gaia-stub" }
+gaia-ccsds-c2a = { path = "gaia-ccsds-c2a" }
+gaia-tmtc = { path = "gaia-tmtc" }
 c2a-devtools-frontend = "1.0"

gaia-stub/proto/broker.proto

@@ -11,6 +11,12 @@ service Broker {
 
   rpc OpenCommandStream(stream CommandStreamRequest) returns (stream CommandStreamResponse);
   rpc PostTelemetry(PostTelemetryRequest) returns (PostTelemetryResponse);
+
+  rpc PostSetVR(PostSetVRRequest) returns (PostSetVRResponse);
+  rpc PostUnlock(PostUnlockRequest) returns (PostUnlockResponse);
+  rpc PostADCommand(PostADCommandRequest) returns (PostADCommandResponse);
+  rpc SubscribeFopFrameEvents(SubscribeFopFrameEventsRequest) returns (stream FopFrameEvent);
+  rpc GetFopState(GetFopStateRequest) returns (GetFopStateResponse);
 }
 
 message PostCommandRequest {
@@ -52,3 +58,50 @@ message GetLastReceivedTelemetryRequest {
 message GetLastReceivedTelemetryResponse {
   tco_tmiv.Tmiv tmiv = 1;
 }
+
+message PostSetVRRequest {
+  uint32 vr = 1;
+}
+
+message PostSetVRResponse {
+}
+
+message PostUnlockRequest { }
+
+message PostUnlockResponse { }
+
+message PostADCommandRequest {
+    tco_tmiv.Tco tco = 1;
+}
+
+message PostADCommandResponse {
+    bool success = 1;
+    uint64 frame_id = 2;
+}
+
+message SubscribeFopFrameEventsRequest {
+}
+
+message FopFrameEvent {
+    uint64 frame_id = 1;
+    enum EventType {
+        TRANSMIT = 0;
+        ACKNOWLEDGED = 1;
+        RETRANSMIT = 2;
+        CANCEL = 3;
+    };
+    EventType event_type = 2;
+}
+
+message GetFopStateRequest {}
+
+message GetFopStateResponse {
+  bool received_clcw = 1;
+  bool lockout_flag = 2;
+  bool wait_flag = 3;
+  bool retransmit_flag = 4;
+  uint32 next_expected_sequence_number = 5;
+
+  bool has_next_sequence_number = 11;
+  uint32 next_sequence_number = 12;
+}

gaia-tmtc/src/broker.rs

@@ -3,6 +3,7 @@ use std::{fmt::Debug, sync::Arc};
 use anyhow::Result;
 use futures::prelude::*;
 use gaia_stub::tco_tmiv::Tco;
+use std::pin::Pin;
 use tokio::sync::Mutex;
 use tokio_stream::wrappers::BroadcastStream;
 use tonic::{Request, Response, Status, Streaming};
@@ -11,36 +12,78 @@ use super::telemetry::{self, LastTmivStore};
 
 pub use gaia_stub::broker::*;
 
-pub struct BrokerService<C> {
+pub struct BrokerService<C, F> {
     cmd_handler: Mutex<C>,
+    fop_command_service: Mutex<F>,
     tlm_bus: telemetry::Bus,
     last_tmiv_store: Arc<LastTmivStore>,
 }
 
-impl<C> BrokerService<C> {
+impl<C, F> BrokerService<C, F> {
     pub fn new(
         cmd_service: C,
+        fop_command_service: F,
         tlm_bus: telemetry::Bus,
         last_tmiv_store: Arc<LastTmivStore>,
     ) -> Self {
         Self {
             cmd_handler: Mutex::new(cmd_service),
+            fop_command_service: Mutex::new(fop_command_service),
             tlm_bus,
             last_tmiv_store,
         }
     }
 }
 
+use async_trait::async_trait;
+pub enum FopFrameEvent {
+    Transmit(u64),
+    Acknowledged(u64),
+    Retransmit(u64),
+    Cancel(u64),
+}
+
+#[derive(Default)]
+pub struct ClcwInfo {
+    pub lockout: bool,
+    pub wait: bool,
+    pub retransmit: bool,
+    pub next_expected_fsn: u8,
+}
+
+pub struct FopState {
+    pub last_clcw: Option<ClcwInfo>,
+    pub next_fsn: Option<u8>,
+}
+
+#[async_trait]
+pub trait FopCommandService {
+    async fn send_set_vr(&self, value: u8);
+
+    async fn send_unlock(&self);
+
+    async fn send_ad_command(&self, tco: Tco) -> Result<u64>;
+
+    async fn subscribe_frame_events(
+        &self,
+    ) -> Result<Pin<Box<dyn Stream<Item = FopFrameEvent> + Send + Sync>>>;
+
+    async fn get_fop_state(&self) -> Result<FopState>;
+}
+
 #[tonic::async_trait]
-impl<C> broker_server::Broker for BrokerService<C>
+impl<C, F> broker_server::Broker for BrokerService<C, F>
 where
     C: super::Handle<Arc<Tco>> + Send + Sync + 'static,
     C::Response: Send + 'static,
+    F: FopCommandService + Send + Sync + 'static,
 {
     type OpenCommandStreamStream =
         stream::BoxStream<'static, Result<CommandStreamResponse, Status>>;
     type OpenTelemetryStreamStream =
         stream::BoxStream<'static, Result<TelemetryStreamResponse, Status>>;
+    type SubscribeFopFrameEventsStream =
+        stream::BoxStream<'static, Result<gaia_stub::broker::FopFrameEvent, Status>>;
 
     #[tracing::instrument(skip(self))]
     async fn post_command(
@@ -66,6 +109,59 @@ where
         Ok(Response::new(PostCommandResponse {}))
     }
 
+    #[tracing::instrument(skip(self))]
+    async fn post_set_vr(
+        &self,
+        request: Request<PostSetVrRequest>,
+    ) -> Result<Response<PostSetVrResponse>, tonic::Status> {
+        let message = request.into_inner();
+        let value = message.vr;
+        self.fop_command_service
+            .lock()
+            .await
+            .send_set_vr(value as _)
+            .await;
+        Ok(Response::new(PostSetVrResponse {}))
+    }
+
+    #[tracing::instrument(skip(self))]
+    async fn post_unlock(
+        &self,
+        _request: Request<PostUnlockRequest>,
+    ) -> Result<Response<PostUnlockResponse>, tonic::Status> {
+        self.fop_command_service.lock().await.send_unlock().await;
+        Ok(Response::new(PostUnlockResponse {}))
+    }
+
+    #[tracing::instrument(skip(self))]
+    async fn post_ad_command(
+        &self,
+        request: Request<PostAdCommandRequest>,
+    ) -> Result<Response<PostAdCommandResponse>, tonic::Status> {
+        let message = request.into_inner();
+
+        let tco = message
+            .tco
+            .ok_or_else(|| Status::invalid_argument("tco is required"))?;
+
+        fn internal_error<E: Debug>(e: E) -> Status {
+            Status::internal(format!("{:?}", e))
+        }
+        let id = self
+            .fop_command_service
+            .lock()
+            .await
+            .send_ad_command(tco)
+            .await
+            .map_err(internal_error)?;
+
+        tracing::info!("AD command sent");
+        Ok(Response::new(PostAdCommandResponse {
+            success: true,
+            frame_id: id,
+        }))
+    }
+
     #[tracing::instrument(skip(self))]
     async fn open_telemetry_stream(
         &self,
@@ -115,4 +211,62 @@ where
             Err(Status::not_found("not received yet"))
         }
     }
+
+    #[tracing::instrument(skip(self))]
+    async fn subscribe_fop_frame_events(
+        &self,
+        _request: tonic::Request<SubscribeFopFrameEventsRequest>,
+    ) -> Result<tonic::Response<Self::SubscribeFopFrameEventsStream>, tonic::Status> {
+        use futures::StreamExt;
+        let stream = self
+            .fop_command_service
+            .lock()
+            .await
+            .subscribe_frame_events()
+            .await
+            .map_err(|_| Status::internal("failed to subscribe frame events"))?;
+        use gaia_stub::broker::fop_frame_event::EventType;
+        let stream = stream.map(|e| {
+            let (frame_id, event_type) = match e {
+                FopFrameEvent::Transmit(id) => (id, EventType::Transmit),
+                FopFrameEvent::Acknowledged(id) => (id, EventType::Acknowledged),
+                FopFrameEvent::Retransmit(id) => (id, EventType::Retransmit),
+                FopFrameEvent::Cancel(id) => (id, EventType::Cancel),
+            };
+            Ok(gaia_stub::broker::FopFrameEvent {
+                frame_id,
+                event_type: event_type.into(),
+            })
+        });
+        Ok(Response::new(stream.boxed()))
+    }
+
+    #[tracing::instrument(skip(self))]
+    async fn get_fop_state(
+        &self,
+        _request: tonic::Request<GetFopStateRequest>,
+    ) -> Result<tonic::Response<GetFopStateResponse>, tonic::Status> {
+        let state = self
+            .fop_command_service
+            .lock()
+            .await
+            .get_fop_state()
+            .await
+            .map_err(|_| Status::internal("failed to get fop state"))?;
+        let received_clcw = state.last_clcw.is_some();
+        let clcw = state.last_clcw.unwrap_or_default();
+
+        let resp = GetFopStateResponse {
+            received_clcw,
+            lockout_flag: clcw.lockout,
+            wait_flag: clcw.wait,
+            retransmit_flag: clcw.retransmit,
+            next_expected_sequence_number: clcw.next_expected_fsn as _,
+
+            has_next_sequence_number: state.next_fsn.is_some(),
+            next_sequence_number: state.next_fsn.unwrap_or_default() as _,
+        };
+
+        Ok(Response::new(resp))
+    }
 }

tmtc-c2a/Cargo.toml

@@ -43,6 +43,7 @@ tokio-tungstenite = "0.20.1"
 itertools = "0.12.1"
 notalawyer = "0.1.0"
 notalawyer-clap = "0.1.0"
+tokio-stream = { version = "0.1.16", features = ["sync"] }
 
 [build-dependencies]
 tonic-build = "0.11"