Add requirement markers (eclipse-uprotocol#19)

Add openfasttrace requirement implementation and test markers to the codebase.

Author: AnotherDaniel

README.md

@@ -14,10 +14,10 @@ This codebase is approaching beta state, mirroring progress in up-spec and up-ru
 - [x] create a usubscription-cli module, a simple command-line frontend for running up-subscription
 - [x] set up a devcontainer
 - [x] feed back learnings and clarifications into up-spec usubscription documentation
+- [x] implement subscription expiry
 - [ ] implement any changes necessary to align with update/re-write of uSubscription specification
-- [ ] create a little demo application for interacting with up-subscription
 - [ ] tests for subscriber persistency
-- [ ] implement subscription expiry
+- [ ] create a little demo application for interacting with up-subscription
 
 ## Getting Started
 

up-subscription/src/handlers/fetch_subscribers.rs

@@ -49,6 +49,7 @@ impl RequestHandler for FetchSubscribersRequestHandler {
         message_attributes: &UAttributes,
         request_payload: Option<UPayload>,
     ) -> Result<Option<UPayload>, ServiceInvocationError> {
+        // [impl->dsn~usubscription-fetch-subscribers-protobuf~1]
         let (fetch_subscribers_request, _source) =
             helpers::extract_inputs::<FetchSubscribersRequest>(
                 RESOURCE_ID_FETCH_SUBSCRIBERS,
@@ -114,6 +115,7 @@ mod tests {
 
     use crate::{helpers, tests::test_lib};
 
+    // [utest->dsn~usubscription-fetch-subscribers-protobuf~1]
     #[tokio::test]
     async fn test_subscribe_success() {
         helpers::init_once();

up-subscription/src/handlers/fetch_subscriptions.rs

@@ -51,6 +51,7 @@ impl RequestHandler for FetchSubscriptionsRequestHandler {
         message_attributes: &UAttributes,
         request_payload: Option<UPayload>,
     ) -> Result<Option<UPayload>, ServiceInvocationError> {
+        // [impl->dsn~usubscription-fetch-subscriptions-protobuf~1]
         let (fetch_subscriptions_request, _source) =
             helpers::extract_inputs::<FetchSubscriptionsRequest>(
                 RESOURCE_ID_FETCH_SUBSCRIPTIONS,
@@ -152,6 +153,7 @@ mod tests {
 
     use crate::{helpers, tests::test_lib};
 
+    // [utest->dsn~usubscription-fetch-subscriptions-protobuf~1]
     #[tokio::test]
     async fn test_fetch_subscriptions_success() {
         helpers::init_once();

up-subscription/src/handlers/register_for_notifications.rs

@@ -45,6 +45,7 @@ impl RequestHandler for RegisterNotificationsRequestHandler {
         message_attributes: &UAttributes,
         request_payload: Option<UPayload>,
     ) -> Result<Option<UPayload>, ServiceInvocationError> {
+        // [impl->dsn~usubscription-register-notifications-protobuf~1]
         let (register_for_notifications_request, source) =
             helpers::extract_inputs::<NotificationsRequest>(
                 RESOURCE_ID_REGISTER_FOR_NOTIFICATIONS,
@@ -87,6 +88,7 @@ mod tests {
 
     use crate::{helpers, tests::test_lib};
 
+    // [utest->dsn~usubscription-register-notifications-protobuf~1]
     #[tokio::test]
     async fn test_register_notification_success() {
         helpers::init_once();

up-subscription/src/handlers/subscribe.rs

@@ -122,6 +122,7 @@ mod tests {
 
     use crate::{helpers, tests::test_lib};
 
+    // [utest->dsn~usubscription-subscribe-protobuf~1]
     #[tokio::test]
     async fn test_subscribe_success() {
         helpers::init_once();

up-subscription/src/handlers/unregister_for_notifications.rs

@@ -45,6 +45,7 @@ impl RequestHandler for UnregisterNotificationsRequestHandler {
         message_attributes: &UAttributes,
         request_payload: Option<UPayload>,
     ) -> Result<Option<UPayload>, ServiceInvocationError> {
+        // [impl->dsn~usubscription-unregister-notifications-protobuf~1]
         let (_subscription_request, source) = helpers::extract_inputs::<NotificationsRequest>(
             RESOURCE_ID_UNREGISTER_FOR_NOTIFICATIONS,
             resource_id,
@@ -81,6 +82,7 @@ mod tests {
 
     use crate::{helpers, tests::test_lib};
 
+    // [utest->dsn~usubscription-unregister-notifications-protobuf~1]
     #[tokio::test]
     async fn test_unregister_notification_success() {
         helpers::init_once();

up-subscription/src/handlers/unsubscribe.rs

@@ -45,6 +45,7 @@ impl RequestHandler for UnubscribeRequestHandler {
         message_attributes: &UAttributes,
         request_payload: Option<UPayload>,
     ) -> Result<Option<UPayload>, ServiceInvocationError> {
+        // [impl->dsn~usubscription-unsubscribe-protobuf~1]
         let (unsubscribe_request, source) = helpers::extract_inputs::<UnsubscribeRequest>(
             RESOURCE_ID_UNSUBSCRIBE,
             resource_id,
@@ -95,6 +96,7 @@ mod tests {
 
     use crate::{helpers, tests::test_lib};
 
+    // [utest->dsn~usubscription-unsubscribe-protobuf~1]
     #[tokio::test]
     async fn test_unsubscribe_success() {
         helpers::init_once();

up-subscription/src/notification_manager.rs

@@ -118,6 +118,7 @@ pub(crate) async fn notification_engine(
                 status,
                 respond_to,
             } => {
+                // [impl->dsn~usubscription-change-notification-type~1]
                 let update = Update {
                     topic: Some(topic).into(),
                     subscriber: Some(SubscriberInfo {
@@ -131,6 +132,7 @@ pub(crate) async fn notification_engine(
 
                 // Send Update message to general notification channel
                 // as per usubscription.proto RegisterForNotifications(NotificationsRequest)
+                // [impl->dsn~usubscription-change-notification-topic~1]
                 match UMessageBuilder::publish(usubscription_uri(RESOURCE_ID_SUBSCRIPTION_CHANGE))
                     .with_message_id(UUID::build())
                     .build_with_protobuf_payload(&update)
@@ -139,6 +141,7 @@ pub(crate) async fn notification_engine(
                         error!("Error building global update notification message: {e}");
                     }
                     Ok(update_msg) => {
+                        // [impl->dsn~usubscription-change-notification-topic~1]
                         if let Err(e) = up_transport.send(update_msg).await {
                             error!(
                                 "Error sending global subscription-change update notification: {e}"
@@ -148,6 +151,7 @@ pub(crate) async fn notification_engine(
                 }
 
                 // Send Update message to any dedicated registered notification-subscribers
+                // [impl->req~usubscription-register-notifications~1]
                 if let Ok(topics) = notifications.get_topics() {
                     for topic_entry in topics {
                         debug!(

up-subscription/src/persistency.rs

@@ -112,10 +112,12 @@ impl SubscriptionsStore {
         let topic_string = &topic.to_uri(PERSIST_UP_SCHEMA);
 
         Ok(
+            // [impl->req~usubscription-subscribe-multiple~1]
             if let Some(mut subscriber_list) = self
                 .persistency
                 .get::<HashMap<SubscriberAsString, Option<ExpiryTimestamp>>>(topic_string)
             {
+                // [impl->req~usubscription-subscribe-expiration-extension~1]
                 subscriber_list.insert(subscriber_string.clone(), expiry);
                 self.persistency
                     .set(topic_string, &subscriber_list)
@@ -182,6 +184,8 @@ impl SubscriptionsStore {
     /// Return a list of all subscribers of given topic
     /// * returns `Vec<SubscriberUUri>` that contains all subscriber UUris registered for the topic
     /// * returns a `PersistencyError` in case of problems with serialization of data or manipulation of persist storage
+    // [impl->req~usubscription-fetch-subscribers-stable-sorting~1]
+    // [impl->req~usubscription-fetch-subscriptions-stable-sorting~1]
     pub(crate) fn get_topic_subscribers(
         &self,
         topic: &TopicUUri,
@@ -210,6 +214,7 @@ impl SubscriptionsStore {
     /// Return a list of all topics subscribed to by given subscriber
     /// * returns `Vec<TopicUUri>` that contains all topics subscribed to by subscriber
     /// * returns a `PersistencyError` in case of problems with serialization of data or manipulation of persist storage
+    // [impl->req~usubscription-fetch-subscriptions-stable-sorting~1]
     pub(crate) fn get_subscriber_topics(
         &self,
         subscriber: &SubscriberUUri,
@@ -270,6 +275,7 @@ impl SubscriptionsStore {
     /// This function does two things
     /// - remove any subscription relationships from persistency that have an expiration timestamp that lies in the past
     /// - return all remaining subscription relationships which have an expiration timestamp that has not yet expired
+    // [impl->req~usubscription-subscribe-no-expiration~1]
     pub(crate) fn get_and_prune_expiring_subscriptions(
         &mut self,
     ) -> Result<Vec<(SubscriberUUri, TopicUUri, ExpiryTimestamp)>, PersistencyError> {
@@ -446,6 +452,7 @@ impl RemoteTopicsStore {
                 ))
             })?
         } else {
+            // [impl->req~usubscription-subscribe-remote-pending~1]
             self.set_topic_state(topic, TopicState::SUBSCRIBE_PENDING)?
         })
     }

up-subscription/src/subscription_manager.rs

@@ -145,6 +145,8 @@ enum Event {
 
 // Core business logic of subscription management - includes container data types for tracking subscriptions and remote subscriptions.
 // Interfacing with this purely works via channels, so we do not have to deal with mutexes and similar concepts.
+// [impl->req~usubscription-unsubscribe-notifications~1]
+// [impl->dsn~usubscription-state-machine~1]
 pub(crate) async fn handle_message(
     configuration: Arc<USubscriptionConfiguration>,
     transport: Arc<dyn UTransport>,
@@ -165,6 +167,8 @@ pub(crate) async fn handle_message(
         mpsc::channel::<InternalSubscriptionEvent>(INTERNAL_COMMAND_BUFFER_SIZE);
 
     // At startup, set up timed unsubscribe for any persisted subscriptions that define an expiration timestamp
+    // [impl->req~usubscription-subscribe-expiration~1]
+    // [impl->req~usubscription-subscribe-no-expiration~1]
     match subscriptions.get_and_prune_expiring_subscriptions() {
         Ok(list) => {
             for (subscriber, topic, expiry_millis) in list {
@@ -210,6 +214,7 @@ pub(crate) async fn handle_message(
                     expiry,
                     respond_to,
                 } => {
+                    // [impl->req~usubscription-subscribe~1]
                     match add_subscription(
                         configuration.clone(),
                         transport.clone(),
@@ -220,6 +225,7 @@ pub(crate) async fn handle_message(
                         topic.clone(),
                         expiry,
                     ) {
+                        // [impl->req~usubscription-subscribe-notifications~1]
                         Ok(result) => {
                             // Send topic state change notification
                             notification_manager::notify(
@@ -244,6 +250,7 @@ pub(crate) async fn handle_message(
                     topic,
                     respond_to,
                 } => {
+                    // [impl->req~usubscription-unsubscribe~1]
                     match remove_subscription(
                         configuration.clone(),
                         transport.clone(),
@@ -277,6 +284,7 @@ pub(crate) async fn handle_message(
                     offset,
                     respond_to,
                 } => match fetch_subscribers(&subscriptions, topic, offset) {
+                    // [impl->req~usubscription-fetch-subscribers~1]
                     Ok(result) => {
                         if respond_to.send(result).is_err() {
                             error!("Problem with internal communication");
@@ -422,10 +430,13 @@ fn add_subscription(
     let _ = topic_subscribers.add_subscription(&subscriber, &topic, expiry)?;
 
     // For REMOTE topics, we explicitly track state due to _PENDING scenarios
+    // [impl->req~usubscription-subscribe-multiple~1]
     let state = if topic.is_remote_authority(&uri_provider.get_authority()) {
         let state = remote_topics.add_topic_or_get_state(&topic)?;
 
         // if this remote topic is not yet SUBSCRIBED, perform remote subscription
+        // [impl->req~usubscription-subscribe-remote~1]
+        // [impl->req~usubscription-subscribe-unsubscribe-pending~1]
         if state != TopicState::SUBSCRIBED {
             let topic_clone = topic.clone();
             let internal_cmd_sender_clone = internal_cmd_sender.clone();
@@ -447,6 +458,8 @@ fn add_subscription(
     };
 
     // Set up timed unsubscribe in case expiration timestamp is set
+    // [impl->req~usubscription-subscribe-expiration~1]
+    // [impl->req~usubscription-subscribe-no-expiration~1]
     if let Some(expiry_millis) = expiry {
         schedule_unsubscribe(
             expiry_millis,
@@ -473,6 +486,8 @@ fn remove_subscription(
     topic: TopicUUri,
 ) -> Result<SubscriptionStatus, persistency::PersistencyError> {
     // if this was the last subscriber to topic and topic is remote
+    // [impl->req~usubscription-unsubscribe-last-remote~1]
+    // [impl->req~usubscription-unsubscribe-subscribe-pending~1]
     if topic_subscribers.remove_subscription(&subscriber, &topic)?
         && topic.is_remote_authority(&uri_provider.get_authority())
     {
@@ -486,6 +501,8 @@ fn remove_subscription(
         });
     }
 
+    // [impl->req~usubscription-unsubscribe-multiple~1]
+    // [impl->req~usubscription-unsubscribe-remote-unsubscribed~1]
     Ok(SubscriptionStatus {
         // Whatever happens with the remote topic state - as far as the local client is concerned, it has now UNSUBSCRIBED from this topic
         state: TopicState::UNSUBSCRIBED.into(),
@@ -503,11 +520,13 @@ fn fetch_subscribers(
     // the remote topic is already fully SUBSCRIBED, of still SUSBCRIBED_PENDING
     let mut subscribers = topic_subscribers.get_topic_subscribers(&topic)?;
 
+    // [impl->req~usubscription-fetch-subscribers-offset~1]
     if let Some(offset) = offset {
         subscribers.drain(..offset as usize);
     }
 
     // split up result list, to make sense of has_more_records field
+    // [impl->req~usubscription-fetch-subscribers-has-more-records~1]
     let has_more = if subscribers.len() > UP_MAX_FETCH_SUBSCRIBERS_LEN {
         subscribers.truncate(UP_MAX_FETCH_SUBSCRIBERS_LEN);
         true
@@ -526,6 +545,7 @@ fn fetch_subscriptions(
     offset: Option<u32>,
 ) -> Result<SubscriptionsResponse, persistency::PersistencyError> {
     let mut results: Vec<SubscriptionEntry> = match request {
+        // [impl->req~usubscription-fetch-subscriptions-by-subscriber~1]
         RequestKind::Subscriber(subscriber) => topic_subscribers
             .get_subscriber_topics(&subscriber)?
             .iter()
@@ -543,6 +563,7 @@ fn fetch_subscriptions(
             })
             .collect(),
 
+        // [impl->req~usubscription-fetch-subscriptions-by-topic~1]
         RequestKind::Topic(topic) => topic_subscribers
             .get_topic_subscribers(&topic)?
             .iter()
@@ -561,11 +582,13 @@ fn fetch_subscriptions(
             .collect(),
     };
 
+    // [impl->req~usubscription-fetch-subscriptions-offset~1]
     if let Some(offset) = offset {
         results.drain(..offset as usize);
     }
 
     // split up result list, to make sense of has_more_records field
+    // [impl->req~usubscription-fetch-subscriptions-has-more-records~1]
     let mut has_more = false;
     if results.len() > UP_MAX_FETCH_SUBSCRIPTIONS_LEN {
         results.truncate(UP_MAX_FETCH_SUBSCRIPTIONS_LEN);
@@ -582,6 +605,7 @@ async fn remote_subscribe(
     transport: Arc<dyn UTransport>,
     internal_cmd_sender: Sender<InternalSubscriptionEvent>,
 ) -> Result<(), UStatus> {
+    // [impl->dsn~usubscription-subscribe-remote-subscriber-change~1]
     let rpc_client: Arc<dyn RpcClient> = Arc::new(
         InMemoryRpcClient::new(transport, uri_provider)
             .await
@@ -595,6 +619,7 @@ async fn remote_subscribe(
     };
 
     // send request
+    // [impl->req~usubscription-remote-max-timeout~1]
     let subscription_response: SubscriptionResponse = rpc_client
         .invoke_proto_method(
             make_remote_subscribe_uuri(&subscription_request.topic),
@@ -610,6 +635,7 @@ async fn remote_subscribe(
         })?;
 
     // deal with response
+    // [impl->req~usubscription-subscribe-remote-response~1]
     if subscription_response.is_state(TopicState::SUBSCRIBED) {
         debug!("Got remote subscription response, state SUBSCRIBED");
 
@@ -634,6 +660,7 @@ async fn remote_unsubscribe(
     internal_cmd_sender: Sender<InternalSubscriptionEvent>,
 ) -> Result<(), UStatus> {
     let rpc_client: Arc<dyn RpcClient> = Arc::new(
+        // [impl->dsn~usubscription-unsubscribe-remote-subscriber-change~1]
         InMemoryRpcClient::new(transport, uri_provider)
             .await
             .map_err(|e| UStatus::fail_with_code(UCode::INTERNAL, e.to_string()))?,
@@ -751,6 +778,7 @@ mod tests {
         mock_provider
     }
 
+    // [utest->req~usubscription-subscribe-expiration~1]
     #[tokio::test]
     async fn test_schedule_future_unsubscribe() {
         let (internal_cmd_sender, mut internal_cmd_receiver) =
@@ -786,6 +814,7 @@ mod tests {
         }
     }
 
+    // [utest->req~usubscription-subscribe-expiration~1]
     #[tokio::test]
     async fn test_schedule_past_unsubscribe() {
         let (internal_cmd_sender, mut internal_cmd_receiver) =