New caching methods

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> {