From a4736cab33f9f4e5d7d517571b6c113d0e70de14 Mon Sep 17 00:00:00 2001
From: imDema <luca.de.martini@protonmail.com>
Date: Tue, 16 Jul 2024 09:45:25 +0200
Subject: [PATCH] New caching methods

---
 src/operator/cache.rs | 30 ++++++++++++++++++++++++++++++
 1 file changed, 30 insertions(+)

diff --git a/src/operator/cache.rs b/src/operator/cache.rs
index 5dff436..c3e1e91 100644
--- a/src/operator/cache.rs
+++ b/src/operator/cache.rs
@@ -136,6 +136,21 @@ impl<I: Data + Send> StreamCache<I> {
         cache.data.clone()
     }
 
+    pub fn into_inner(self) -> HashMap<CoordUInt, Vec<I>> {
+        assert!(self.data.lock().is_complete(), "Reading cache before it was complete. execution from a cached stream must start after the previous StreamContext has completed!");
+        Arc::into_inner(self.data).expect("multiple Arc ref in StreamCache").into_inner().data
+    }
+
+    pub fn values_cloned(&self) -> impl Iterator<Item=I> {
+        let clone = self.inner_cloned();
+        clone.into_values().flatten()
+    }
+
+    pub fn into_values(self) -> impl Iterator<Item=I> {
+        let inner = self.into_inner();
+        inner.into_values().flatten()
+    }
+
     /// Consume the cache creating a new [Stream] in a [StreamContext].
     /// 
     /// The [StreamCache] will behave as a source with the same parallelism (and distribution of data)
@@ -155,6 +170,21 @@ impl<I: Data + Send> StreamCache<I> {
         let source = CacheSource::new(self.replication, self.data);
         ctx.stream(source)
     }
+
+    /// Consume the cache creating a new [Stream] in a [StreamContext] with the same [RuntimeConfig].
+    /// 
+    /// The [StreamCache] will behave as a source with the same parallelism (and distribution of data)
+    /// as the original [Stream] it was cached from.
+    /// 
+    /// **Returns**: a tuple containing the new [StreamContext] and the [Stream] with the [CacheSource].
+    /// 
+    /// + **ATTENTION** ⚠️: The cache can be resumed **only after** the execution of its origin
+    /// `StreamContext` has terminated.
+    pub fn stream(self) -> (StreamContext, Stream<CacheSource<I>>) {
+        let ctx = StreamContext::new(self.config.clone());
+        let stream = self.stream_in(&ctx);
+        (ctx, stream)
+    }
 }
 
 impl<I: Data + Send> Clone for StreamCache<I> {