refactor(hydroflow_plus): dedup signatures for Stream operators (#1525

)

hydroflow_plus/src/location/cluster.rs

@@ -54,15 +54,8 @@ impl<'a, C> Location<'a> for Cluster<'a, C> {
         &self.flow_state
     }
 
-    fn make_from(id: LocationId, flow_state: FlowState) -> Self {
-        match id {
-            LocationId::Cluster(id) => Cluster {
-                id,
-                flow_state,
-                _phantom: PhantomData,
-            },
-            _ => panic!(),
-        }
+    fn is_top_level() -> bool {
+        true
     }
 }
 

hydroflow_plus/src/location/external_process.rs

@@ -53,15 +53,8 @@ impl<'a, P> Location<'a> for ExternalProcess<'a, P> {
         &self.flow_state
     }
 
-    fn make_from(id: LocationId, flow_state: FlowState) -> Self {
-        match id {
-            LocationId::ExternalProcess(id) => ExternalProcess {
-                id,
-                flow_state,
-                _phantom: PhantomData,
-            },
-            _ => panic!(),
-        }
+    fn is_top_level() -> bool {
+        true
     }
 }
 

hydroflow_plus/src/location/mod.rs

@@ -55,7 +55,7 @@ pub trait Location<'a>: Clone {
 
     fn flow_state(&self) -> &FlowState;
 
-    fn make_from(id: LocationId, flow_state: FlowState) -> Self;
+    fn is_top_level() -> bool;
 
     fn nest(&self) -> Tick<Self>
     where
@@ -111,10 +111,12 @@ pub trait Location<'a>: Clone {
     fn source_iter<T, E: IntoIterator<Item = T>>(
         &self,
         e: impl Quoted<'a, E>,
-    ) -> Stream<T, Bounded, Self>
+    ) -> Stream<T, Unbounded, Self>
     where
         Self: Sized + NoTick,
     {
+        // TODO(shadaj): we mark this as unbounded because we do not yet have a representation
+        // for bounded top-level streams, and this is the only way to generate one
         let e = e.splice_untyped();
 
         Stream::new(
@@ -126,10 +128,13 @@ pub trait Location<'a>: Clone {
         )
     }
 
-    fn singleton<T: Clone>(&self, e: impl Quoted<'a, T>) -> Singleton<T, Bounded, Self>
+    fn singleton<T: Clone>(&self, e: impl Quoted<'a, T>) -> Singleton<T, Unbounded, Self>
     where
         Self: Sized + NoTick,
     {
+        // TODO(shadaj): we mark this as unbounded because we do not yet have a representation
+        // for bounded top-level singletons, and this is the only way to generate one
+
         let e_arr = q!([e]);
         let e = e_arr.splice_untyped();
 

hydroflow_plus/src/location/process.rs

@@ -28,14 +28,7 @@ impl<'a, P> Location<'a> for Process<'a, P> {
         &self.flow_state
     }
 
-    fn make_from(id: LocationId, flow_state: FlowState) -> Self {
-        match id {
-            LocationId::Process(id) => Process {
-                id,
-                flow_state,
-                _phantom: PhantomData,
-            },
-            _ => panic!(),
-        }
+    fn is_top_level() -> bool {
+        true
     }
 }

hydroflow_plus/src/location/tick.rs

@@ -26,9 +26,7 @@ impl<'a, L: Location<'a>> Location<'a> for Tick<L> {
         self.l.flow_state()
     }
 
-    fn make_from(id: LocationId, flow_state: FlowState) -> Self {
-        Tick {
-            l: L::make_from(id, flow_state),
-        }
+    fn is_top_level() -> bool {
+        false
     }
 }

hydroflow_plus/src/singleton.rs

@@ -515,6 +515,15 @@ impl<'a, T: Clone, W, N: Location<'a>> Clone for Optional<T, W, N> {
 }
 
 impl<'a, T, W, N: Location<'a>> Optional<T, W, N> {
+    // TODO(shadaj): this is technically incorrect; we should only return the first element of the stream
+    pub fn into_stream(self) -> Stream<T, W, N> {
+        if N::is_top_level() {
+            panic!("Converting an optional to a stream is not yet supported at the top level");
+        }
+
+        Stream::new(self.location, self.ir_node.into_inner())
+    }
+
     pub fn map<U, F: Fn(T) -> U + 'a>(self, f: impl IntoQuotedMut<'a, F>) -> Optional<U, W, N> {
         Optional::new(
             self.location,
@@ -563,11 +572,6 @@ impl<'a, T, W, N: Location<'a>> Optional<T, W, N> {
 }
 
 impl<'a, T, N: Location<'a>> Optional<T, Bounded, Tick<N>> {
-    // TODO(shadaj): this is technically incorrect; we should only return the first element of the stream
-    pub fn into_stream(self) -> Stream<T, Bounded, Tick<N>> {
-        Stream::new(self.location, self.ir_node.into_inner())
-    }
-
     pub fn cross_singleton<O>(
         self,
         other: impl Into<Optional<O, Bounded, Tick<N>>>,