Some performance and code improvements

src/cluster_engine/engine.rs

@@ -39,6 +39,8 @@ impl Grouping {
 // - fix "Action".
 // - find a way to merge action, query and response in a type-safe manner...
 // - rename cluster_engine to model?
+// - move the different operations in modules and just share the state
+// - replace row_cache with the LRU crate I have open
 
 /// This signifies the action we're currently evaluating
 /// It is used for sending requests and receiving responses
@@ -213,18 +215,14 @@ impl Engine {
 
     // Send the last action over the wire to be calculated
     fn update(&mut self, payload: (Query, Action)) -> Result<()> {
-        Ok(self.link.input_sender.send((payload.0, payload.1))?)
+        Ok(self.link.request(&payload.0, payload.1)?)
     }
 
     /// Fetch the channels to see if there're any updates
     pub fn process(&mut self) -> Result<()> {
-        let response = match self.link.output_receiver.try_recv() {
+        let response = match self.link.receive()? {
-            // We received something
+            Some(n) => n,
-            Ok(Ok(response)) => response,
+            None => return Ok(()),
-            // We received nothing
-            Err(_) => return Ok(()),
-            // There was an error, we forward it
-            Ok(Err(e)) => return Err(e),
         };
 
         match response {
@@ -269,43 +267,36 @@ impl Engine {
             .collect()
     }
 
-    pub fn request_contents(&mut self, range: &Range<usize>) -> Result<()> {
+    /// Query the contents for the current filter settings.
-        // Mark the rows as being loaded
+    /// This call will return the available data and request additional data when it is missing.
-        for index in range.clone() {
+    /// The return value indicates whether a row is loaded or loading.
-            if self.row_cache.cache_get(&index).is_none() {
+    pub fn current_contents(&mut self, range: &Range<usize>) -> Result<Vec<Option<QueryRow>>> {
-                self.row_cache.cache_set(index, LoadingState::Loading);
-            }
-        }
-        let request = self.make_normal_query(range.clone());
-        self.link
-            .input_sender
-            .send((request.clone(), Action::Mails))?;
-        Ok(())
-    }
-
-    /// Query the contents for the current filter settings
-    /// This is a blocking call to simplify things a great deal
-    /// - returns the data, and an indicator that data is missing so that we can load more data
-    pub fn current_contents(
-        &mut self,
-        range: &Range<usize>,
-    ) -> Result<(Vec<Option<QueryRow>>, bool)> {
         // build an array with either empty values or values from our cache.
         let mut rows = Vec::new();
-        let mut data_missing = false;
+
+        let mut missing_data = false;
         for index in range.clone() {
             let entry = self.row_cache.cache_get(&index);
             let entry = match entry {
                 Some(LoadingState::Loaded(n)) => Some((*n).clone()),
                 Some(LoadingState::Loading) => None,
                 None => {
-                    data_missing = true;
+                    // for simplicity, we keep the "something is missing" state separate
+                    missing_data = true;
+
+                    // Mark the row as being loaded
+                    self.row_cache.cache_set(index, LoadingState::Loading);
                     None
                 }
             };
             rows.push(entry);
         }
-        Ok((rows, data_missing))
+        // Only if at least some data is missing do we perform the request
+        if missing_data && !range.is_empty() {
+            let request = self.make_normal_query(range.clone());
+            self.link.request(&request, Action::Mails)?;
+        }
+        Ok(rows)
     }
 
     pub fn is_busy(&self) -> bool {
@@ -320,7 +311,7 @@ impl Engine {
 
     /// If we're loading mails
     pub fn is_mail_busy(&self) -> bool {
-        !self.link.input_sender.is_empty()
+        self.link.is_processing()
     }
 
     fn make_group_query(&self) -> Result<Query> {

src/cluster_engine/link.rs

@@ -25,10 +25,6 @@ use super::partitions::{Partition, Partitions};
 // - instead of hard-coding subject/sender-domain, have a "Detail" trait
 // - consider a better logic for the cache (by row id and just fetch the smallest range that contains all missing numbers)
 
-pub trait Payload<O> {
-    fn map_response<T>(self, response: T) -> Self;
-}
-
 #[derive(Debug)]
 pub enum Response<Context: Send + 'static> {
     Grouped(Query, Context, Partitions),
@@ -43,6 +39,41 @@ pub struct Link<Context: Send + 'static> {
     pub input_sender: InputSender<Context>,
     pub output_receiver: OutputReciever<Context>,
     pub handle: Handle,
+    // We need to account for the brief moment where the processing channel is empty
+    // but we're applying the results. If there is a UI update in this window,
+    // the UI will not update again after the changes were applied because an empty
+    // channel indicates completed processing.
+    // There's also a delay between a request taken out of the input channel and being
+    // put into the output channel. In order to account for all of this, we emploty a
+    // request counter to know how many requests are currently in the pipeline
+    request_counter: usize,
+}
+
+impl<Context: Send + Sync + 'static> Link<Context> {
+    pub fn request(&mut self, query: &Query, context: Context) -> Result<()> {
+        self.request_counter += 1;
+        self.input_sender.send((query.clone(), context))?;
+        Ok(())
+    }
+
+    pub fn receive(&mut self) -> Result<Option<Response<Context>>> {
+        match self.output_receiver.try_recv() {
+            // We received something
+            Ok(Ok(response)) => {
+                // Only subtract if we successfuly received a value
+                self.request_counter -= 1;
+                Ok(Some(response))
+            }
+            // We received nothing
+            Err(_) => Ok(None),
+            // There was an error, we forward it
+            Ok(Err(e)) => Err(e),
+        }
+    }
+
+    pub fn is_processing(&self) -> bool {
+        self.request_counter > 0
+    }
 }
 
 pub fn run<Context: Send + Sync + 'static>(config: &Config) -> Result<Link<Context>> {
@@ -54,6 +85,7 @@ pub fn run<Context: Send + Sync + 'static>(config: &Config) -> Result<Link<Conte
         input_sender,
         output_receiver,
         handle,
+        request_counter: 0,
     })
 }
 
@@ -68,7 +100,7 @@ fn inner_loop<Context: Send + Sync + 'static>(
         let response = match query {
             Query::Grouped { .. } => {
                 let partitions = calculate_partitions(&result)?;
-                Response::Grouped(query, context, Partitions::new(partitions))
+                Response::Grouped(query, context, partitions)
             }
             Query::Normal { .. } => {
                 let converted = calculate_rows(&result)?;
@@ -79,14 +111,14 @@ fn inner_loop<Context: Send + Sync + 'static>(
     }
 }
 
-fn calculate_partitions(result: &[QueryResult]) -> Result<Vec<Partition>> {
+fn calculate_partitions(result: &[QueryResult]) -> Result<Partitions> {
     let mut partitions = Vec::new();
     for r in result.iter() {
         let partition = r.try_into()?;
         partitions.push(partition);
     }
 
-    Ok(partitions)
+    Ok(Partitions::new(partitions))
 }
 
 fn calculate_rows(result: &[QueryResult]) -> Result<Vec<QueryRow>> {

src/gui/mail_panel.rs

@@ -26,16 +26,18 @@ impl<'a> Widget for MailPanel<'a> {
                     &mut selected_row,
                     self.engine.current_element_count(),
                     |range| {
-                        let (rows, load_more) = match self.engine.current_contents(&range) {
+                        // we overshoot the range a bit, as otherwise somehow the bottom is always empty
-                            Ok((n, load_more)) => (n, load_more),
+                        let range = std::ops::Range {
+                            start: range.start,
+                            end: range.end + 6,
+                        };
+                        let rows = match self.engine.current_contents(&range) {
+                            Ok(n) => n,
                             Err(e) => {
                                 *self.error = Some(e);
-                                (empty_vec.clone(), false)
+                                empty_vec.clone()
                             }
                         };
-                        if load_more {
-                            *self.error = self.engine.request_contents(&range).err();
-                        }
                         rows
                     },
                 )