Remote pagination and obsolete sorting requirements (eclipse-uprotocol#24)

- Pull in latest up-spec
- Remove pagination code, to follow up-spec/#283
- Remove obsolete sorting requirement links

Author: AnotherDaniel

up-spec

@@ -24,10 +24,7 @@ use up_rust::{
     UAttributes,
 };
 
-use crate::{
-    helpers,
-    subscription_manager::{SubscribersResponse, SubscriptionEvent},
-};
+use crate::{helpers, subscription_manager::SubscriptionEvent, usubscription::SubscriberUUri};
 
 pub(crate) struct FetchSubscribersRequestHandler {
     subscription_sender: Sender<SubscriptionEvent>,
@@ -57,7 +54,7 @@ impl RequestHandler for FetchSubscribersRequestHandler {
                 &request_payload,
                 message_attributes,
             )?;
-        let FetchSubscribersRequest { topic, offset, .. } = fetch_subscribers_request;
+        let FetchSubscribersRequest { topic, .. } = fetch_subscribers_request;
         let Some(topic) = topic.into_option() else {
             return Err(ServiceInvocationError::InvalidArgument(
                 "No topic defined in request".to_string(),
@@ -69,12 +66,8 @@ impl RequestHandler for FetchSubscribersRequestHandler {
         helpers::validate_uri(&topic)?;
 
         // Interact with subscription manager backend
-        let (respond_to, receive_from) = oneshot::channel::<SubscribersResponse>();
-        let se = SubscriptionEvent::FetchSubscribers {
-            topic,
-            offset,
-            respond_to,
-        };
+        let (respond_to, receive_from) = oneshot::channel::<Vec<SubscriberUUri>>();
+        let se = SubscriptionEvent::FetchSubscribers { topic, respond_to };
 
         if let Err(e) = self.subscription_sender.send(se).await {
             error!("Error communicating with subscription manager: {e}");
@@ -89,8 +82,7 @@ impl RequestHandler for FetchSubscribersRequestHandler {
         };
 
         // Build and return result
-        let (subscribers, has_more) = fetch_subscribers_response;
-        let subscriber_infos = subscribers
+        let subscriber_infos = fetch_subscribers_response
             .iter()
             .map(|subscriber| SubscriberInfo {
                 uri: Some(subscriber.clone()).into(),
@@ -99,7 +91,6 @@ impl RequestHandler for FetchSubscribersRequestHandler {
             .collect();
         let fetch_subscribers_response = FetchSubscribersResponse {
             subscribers: subscriber_infos,
-            has_more_records: Some(has_more),
             ..Default::default()
         };
 
@@ -165,7 +156,7 @@ mod tests {
                 topic, respond_to, ..
             } => {
                 assert_eq!(topic, test_lib::helpers::local_topic1_uri());
-                let _ = respond_to.send(SubscribersResponse::default());
+                let _ = respond_to.send(Vec::default());
             }
             _ => panic!("Wrong event type"),
         }

up-subscription/src/handlers/fetch_subscribers.rs

@@ -26,9 +26,7 @@ use up_rust::{
 
 use crate::{
     helpers,
-    subscription_manager::{
-        RequestKind, SubscriptionEntry, SubscriptionEvent, SubscriptionsResponse,
-    },
+    subscription_manager::{RequestKind, SubscriptionEntry, SubscriptionEvent},
 };
 
 pub(crate) struct FetchSubscriptionsRequestHandler {
@@ -60,9 +58,7 @@ impl RequestHandler for FetchSubscriptionsRequestHandler {
                 message_attributes,
             )?;
 
-        let FetchSubscriptionsRequest {
-            request, offset, ..
-        } = fetch_subscriptions_request;
+        let FetchSubscriptionsRequest { request, .. } = fetch_subscriptions_request;
         let request_kind = match request {
             Some(Request::Topic(topic)) => {
                 if !topic.is_empty() {
@@ -92,10 +88,9 @@ impl RequestHandler for FetchSubscriptionsRequestHandler {
         };
 
         // Interact with subscription manager backend
-        let (respond_to, receive_from) = oneshot::channel::<SubscriptionsResponse>();
+        let (respond_to, receive_from) = oneshot::channel::<Vec<SubscriptionEntry>>();
         let se = SubscriptionEvent::FetchSubscriptions {
             request: request_kind,
-            offset,
             respond_to,
         };
 
@@ -112,8 +107,7 @@ impl RequestHandler for FetchSubscriptionsRequestHandler {
         };
 
         // Build and return result
-        let (subscriptions, has_more) = fetch_subscriptions_response;
-        let subscription_list: Vec<Subscription> = subscriptions
+        let subscription_list: Vec<Subscription> = fetch_subscriptions_response
             .iter()
             .map(
                 |SubscriptionEntry {
@@ -135,7 +129,6 @@ impl RequestHandler for FetchSubscriptionsRequestHandler {
 
         let fetch_subscriptions_response = FetchSubscriptionsResponse {
             subscriptions: subscription_list,
-            has_more_records: Some(has_more),
             ..Default::default()
         };
 
@@ -202,7 +195,6 @@ mod tests {
         match subscription_event {
             SubscriptionEvent::FetchSubscriptions {
                 request,
-                offset,
                 respond_to,
             } => {
                 match request {
@@ -216,11 +208,8 @@ mod tests {
                     }
                     _ => panic!("Wrong request details"),
                 }
-                assert_eq!(offset.unwrap_or_default(), 42);
-
-                let _ = respond_to.send(SubscriptionsResponse::default());
+                let _ = respond_to.send(Vec::default());
             }
-
             _ => panic!("Wrong event type"),
         }
     }

up-subscription/src/handlers/fetch_subscriptions.rs

@@ -184,8 +184,6 @@ 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,
@@ -214,7 +212,6 @@ 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,

up-subscription/src/persistency.rs

@@ -43,11 +43,6 @@ use crate::{
 // via tokio mpsc channel. This design allows to forgo the use of any synhronization primitives on the subscription-tracking container
 // data types, as any access is coordinated/serialized via the Event selection loop.
 
-// Maximum number of `Subscriber` entries to be returned in a `FetchSusbcriptions´ operation.
-const UP_MAX_FETCH_SUBSCRIBERS_LEN: usize = 100;
-// Maximum number of `Subscriber` entries to be returned in a `FetchSusbcriptions´ operation.
-const UP_MAX_FETCH_SUBSCRIPTIONS_LEN: usize = 100;
-
 // Queue size of message channel for internal commands - like subscription status change messages or subscription expiration commands.
 const INTERNAL_COMMAND_BUFFER_SIZE: usize = 128;
 
@@ -68,9 +63,6 @@ pub(crate) struct SubscriptionEntry {
     pub(crate) status: SubscriptionStatus,
 }
 
-pub(crate) type SubscribersResponse = (Vec<SubscriberUUri>, bool); // List of subscribers, boolean flag stating if there exist more entries than contained in list
-pub(crate) type SubscriptionsResponse = (Vec<SubscriptionEntry>, bool); // List of subscriber entries, boolean flag stating if there exist more entries than contained in list
-
 // This is the 'outside API' of subscription manager, it includes some events that are only to be used in (and only enabled for) testing.
 #[derive(Debug)]
 pub(crate) enum SubscriptionEvent {
@@ -87,13 +79,11 @@ pub(crate) enum SubscriptionEvent {
     },
     FetchSubscribers {
         topic: TopicUUri,
-        offset: Option<u32>,
-        respond_to: oneshot::Sender<SubscribersResponse>, // return list of subscribers and flag indicating whether there are more
+        respond_to: oneshot::Sender<Vec<SubscriberUUri>>,
     },
     FetchSubscriptions {
         request: RequestKind,
-        offset: Option<u32>,
-        respond_to: oneshot::Sender<SubscriptionsResponse>, // return list of Subscriptions and flag indicating whether there are more
+        respond_to: oneshot::Sender<Vec<SubscriptionEntry>>,
     },
     // Purely for use during testing: get copy of current topic-subscriper ledger
     #[cfg(test)]
@@ -281,26 +271,23 @@ pub(crate) async fn handle_message(
                         }
                     }
                 }
-                SubscriptionEvent::FetchSubscribers {
-                    topic,
-                    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");
+                SubscriptionEvent::FetchSubscribers { topic, respond_to } => {
+                    match fetch_subscribers(&subscriptions, topic) {
+                        // [impl->req~usubscription-fetch-subscribers~1]
+                        Ok(result) => {
+                            if respond_to.send(result).is_err() {
+                                error!("Problem with internal communication");
+                            }
+                        }
+                        Err(e) => {
+                            panic!("Persistency failure {e}")
                         }
                     }
-                    Err(e) => {
-                        panic!("Persistency failure {e}")
-                    }
-                },
+                }
                 SubscriptionEvent::FetchSubscriptions {
                     request,
-                    offset,
                     respond_to,
-                } => match fetch_subscriptions(&subscriptions, &remote_topics, request, offset) {
+                } => match fetch_subscriptions(&subscriptions, &remote_topics, request) {
                     Ok(result) => {
                         if respond_to.send(result).is_err() {
                             error!("Problem with internal communication");
@@ -541,37 +528,19 @@ fn remove_subscription(
 fn fetch_subscribers(
     topic_subscribers: &persistency::SubscriptionsStore,
     topic: TopicUUri,
-    offset: Option<u32>,
-) -> Result<SubscribersResponse, persistency::PersistencyError> {
+) -> Result<Vec<SubscriberUUri>, persistency::PersistencyError> {
     // This will get *every* client that subscribed to `topic` - no matter whether (in the case of remote subscriptions)
     // 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
-    } else {
-        false
-    };
-
-    Ok((subscribers, has_more))
+    topic_subscribers.get_topic_subscribers(&topic)
 }
 
 // Fetch all subscriptions of a topic or subscribers
 fn fetch_subscriptions(
     topic_subscribers: &persistency::SubscriptionsStore,
     remote_topics: &persistency::RemoteTopicsStore,
     request: RequestKind,
-    offset: Option<u32>,
-) -> Result<SubscriptionsResponse, persistency::PersistencyError> {
-    let mut results: Vec<SubscriptionEntry> = match request {
+) -> Result<Vec<SubscriptionEntry>, persistency::PersistencyError> {
+    let results: Vec<SubscriptionEntry> = match request {
         // [impl->req~usubscription-fetch-subscriptions-by-subscriber~1]
         RequestKind::Subscriber(subscriber) => topic_subscribers
             .get_subscriber_topics(&subscriber)?
@@ -609,20 +578,7 @@ 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);
-        has_more = true;
-    }
-
-    Ok((results, has_more))
+    Ok(results)
 }
 
 // Perform remote topic subscription