Wiring development

Cargo.toml

@@ -17,7 +17,7 @@ pin-utils = "0.1.0-alpha.4"
 slab = "0.4.2"
 streaming-iterator = "0.1.4"
 objekt-clonable = "0.2.2"
-multiqueue = "0.3.2"
+multiqueue2 = "0.1.6"
 crossbeam-channel = "0.3.9"
 bus = "2.2.2"
 

examples/single_directed.rs

@@ -0,0 +1,23 @@
+use bastion_streams::stream::flow::map::Map;
+use bastion_streams::stream::sink::ignore::Ignore;
+use bastion_streams::stream::source::single::Single;
+
+use bastion_streams::stream::topology::architect::Architect;
+use bastion_streams::stream::topology::container::Container;
+
+fn main() {
+    let single = Single::<i32>::new(999);
+    let mapper = Map::<i32, i32>::new(Box::new(|x: i32| x + 1));
+    let sink = Ignore::<i32>::new();
+
+    let stages = vec![
+        Container::Source(Box::new(single)),
+        Container::Transform(Box::new(mapper)),
+        Container::Sink(Box::new(sink)),
+    ];
+
+    let mut architect = Architect::graph(stages);
+    architect.check_bounds();
+    architect.visit_stages();
+    architect.run();
+}

src/lib.rs

@@ -1,4 +1,4 @@
-#![feature(clone_closures)]
+//#![feature(clone_closures)]
 
 #[macro_use]
 pub mod stream;

src/stream/flow/map.rs

@@ -5,20 +5,44 @@ use objekt_clonable::clonable;
 
 #[clonable]
 pub trait MapClosure<I, O>: Fn(I) -> O + Clone + Send + Sync + 'static {}
+impl<I, O, T> MapClosure<I, O> for T where T: Fn(I) -> O + Clone + Send + Sync + 'static {}
+
 type MapFn<I, O> = Box<dyn MapClosure<I, O>>;
 
 pub struct Map<I, O> {
-    pub shape: FlowShape<'static, I, O>,
-    pub stage_id: usize,
+    pub shape: Option<FlowShape<'static, I, O>>,
+    pub stage_id: Option<usize>,
 
     pub map_fn: MapFn<I, O>,
 
-    pub demand_rx: BroadcastReceiver<Demand>,
-    pub demand_tx: BroadcastSender<Demand>,
+    pub demand_rx: Option<BroadcastReceiver<Demand>>,
+    pub demand_tx: Option<BroadcastSender<Demand>>,
 
-    pub in_handler: Box<dyn InHandler>,
-    pub out_handler: Box<dyn OutHandler>,
-    pub logic: GraphStageLogic,
+    pub in_handler: Option<Box<dyn InHandler>>,
+    pub out_handler: Option<Box<dyn OutHandler>>,
+    pub logic: Option<GraphStageLogic>,
+}
+
+impl<I, O> Map<I, O>
+where
+    I: Clone,
+    O: Clone,
+{
+    pub fn new(map_fn: MapFn<I, O>) -> Self {
+        Self {
+            shape: None,
+            stage_id: None,
+
+            map_fn,
+
+            demand_rx: None,
+            demand_tx: None,
+
+            in_handler: None,
+            out_handler: None,
+            logic: None,
+        }
+    }
 }
 
 ///// Map handler
@@ -39,16 +63,16 @@ struct MapHandler<I, O> {
 }
 
 impl<I, O> OutHandler for MapHandler<I, O>
-    where
-        I: Clone + Send + Sync + 'static,
-        O: Clone + Send + Sync + 'static,
+where
+    I: Clone + Send + Sync + 'static,
+    O: Clone + Send + Sync + 'static,
 {
     fn name(&self) -> String {
         String::from("map-flow-out")
     }
 
     fn on_pull(&self) {
-        unimplemented!()
+        if let Ok(_elem) = self.out_rx.as_ref().unwrap().try_recv() {}
     }
 
     fn on_downstream_finish(&self) {
@@ -61,9 +85,9 @@ impl<I, O> OutHandler for MapHandler<I, O>
 }
 
 impl<I, O> InHandler for MapHandler<I, O>
-    where
-        I: Clone + Send + Sync,
-        O: Clone + Send + Sync,
+where
+    I: Clone + Send + Sync,
+    O: Clone + Send + Sync,
 {
     fn name(&self) -> String {
         String::from("map-flow-in")
@@ -78,7 +102,7 @@ impl<I, O> InHandler for MapHandler<I, O>
             // todo: on_pull make demand from the upper
             let demand = Demand {
                 stage_id: self.stage_id,
-                style: DemandStyle::DemandFull(100)
+                style: DemandStyle::DemandFull(100),
             };
             self.demand_tx.as_ref().unwrap().try_send(demand).unwrap();
         }
@@ -111,7 +135,7 @@ impl<I, O> Default for MapHandler<I, O> {
     }
 }
 
-impl<'a, I, O> GraphStage<'a> for Map<I, O>
+impl<I, O> GraphStage for Map<I, O>
 where
     I: Clone + Send + Sync + 'static,
     O: Clone + Send + Sync + 'static,
@@ -120,47 +144,62 @@ where
         let map_flow_inlet = Inlet::<I>::new(0, "Map.in");
         let map_flow_outlet = Outlet::<O>::new(0, "Map.out");
 
-        self.shape = FlowShape {
+        self.shape = Some(FlowShape {
             inlet: map_flow_inlet,
             outlet: map_flow_outlet,
-        };
+        });
     }
 
-    fn build_demand(&'a mut self, tx: BroadcastSender<Demand>, rx: BroadcastReceiver<Demand>) {
-        self.demand_tx = tx;
-        self.demand_rx = rx;
+    fn build_demand(&mut self, tx: BroadcastSender<Demand>, rx: BroadcastReceiver<Demand>) {
+        self.demand_tx = Some(tx);
+        self.demand_rx = Some(rx);
     }
 
-    fn create_logic(&mut self, _attributes: Attributes) -> GraphStageLogic {
+    fn create_logic(&mut self, stage_id: usize, _attributes: Attributes) {
         self.build_shape();
 
         let (in_tx, in_rx) = unbounded::<I>();
         let (out_tx, out_rx) = unbounded::<O>();
 
-        let handler = Box::new(MapHandler {
+        let handler = Some(Box::new(MapHandler {
             map_fn: Some(self.map_fn.clone()),
             in_tx: Some(in_tx),
             in_rx: Some(in_rx),
             out_rx: Some(out_rx),
             out_tx: Some(out_tx),
-            demand_rx: Some(self.demand_rx.clone()),
-            demand_tx: Some(self.demand_tx.clone()),
-            stage_id: self.stage_id
-        });
+            demand_rx: Some(self.demand_rx.as_ref().unwrap().clone()),
+            demand_tx: Some(self.demand_tx.as_ref().unwrap().clone()),
+            stage_id,
+        }));
+
+        self.stage_id = Some(stage_id);
 
-        self.in_handler = handler.clone();
-        self.out_handler = handler.clone();
+        self.in_handler = Some(handler.as_ref().unwrap().clone());
+        self.out_handler = Some(handler.as_ref().unwrap().clone());
 
-        let shape = Box::new(self.shape.clone());
+        let shape = Box::new(self.shape.as_ref().unwrap().clone());
 
         let mut gsl = GraphStageLogic::from_shape::<I, O>(shape);
-        gsl.set_inlet_handler(self.shape.inlet.clone(), self.in_handler.clone());
-        gsl.set_outlet_handler(self.shape.outlet.clone(), self.out_handler.clone());
-        self.logic = gsl.clone();
-        gsl
+        gsl.set_inlet_handler(
+            self.shape.as_ref().unwrap().inlet.clone(),
+            self.in_handler.as_ref().unwrap().clone(),
+        );
+        gsl.set_outlet_handler(
+            self.shape.as_ref().unwrap().outlet.clone(),
+            self.out_handler.as_ref().unwrap().clone(),
+        );
+        self.logic = Some(gsl);
+    }
+
+    fn get_shape(&self) -> ShapeType {
+        self.shape.as_ref().unwrap().shape_type()
+    }
+
+    fn get_stage_id(&self) -> usize {
+        self.stage_id.unwrap()
     }
 
-    fn get_shape(&'a self) -> ShapeType {
-        self.shape.shape_type()
+    fn get_logic(&self) -> &GraphStageLogic {
+        self.logic.as_ref().unwrap()
     }
 }

src/stream/sink/ignore.rs

@@ -3,15 +3,34 @@ use crossbeam_channel::{unbounded, Receiver, Sender};
 use futures::io::Error;
 
 pub struct Ignore<I> {
-    pub shape: SinkShape<'static, I>,
-    pub stage_id: usize,
+    pub shape: Option<SinkShape<'static, I>>,
+    pub stage_id: Option<usize>,
 
-    pub demand_rx: BroadcastReceiver<Demand>,
-    pub demand_tx: BroadcastSender<Demand>,
+    pub demand_rx: Option<BroadcastReceiver<Demand>>,
+    pub demand_tx: Option<BroadcastSender<Demand>>,
 
-    pub in_handler: Box<dyn InHandler>,
-    pub out_handler: Box<dyn OutHandler>,
-    pub logic: GraphStageLogic,
+    pub in_handler: Option<Box<dyn InHandler>>,
+    pub out_handler: Option<Box<dyn OutHandler>>,
+    pub logic: Option<GraphStageLogic>,
+}
+
+impl<I> Ignore<I>
+where
+    I: Clone,
+{
+    pub fn new() -> Self {
+        Self {
+            shape: None,
+            stage_id: None,
+
+            demand_rx: None,
+            demand_tx: None,
+
+            in_handler: None,
+            out_handler: None,
+            logic: None,
+        }
+    }
 }
 
 #[derive(Clone)]
@@ -26,8 +45,8 @@ struct IgnoreHandler<I> {
 }
 
 impl<I> InHandler for IgnoreHandler<I>
-    where
-        I: Clone + 'static,
+where
+    I: Clone + 'static,
 {
     fn name(&self) -> String {
         String::from("ignore-sink-in")
@@ -37,11 +56,12 @@ impl<I> InHandler for IgnoreHandler<I>
         if let Ok(_elem) = self.in_rx.as_ref().unwrap().try_recv() {
             println!("Ignored");
         } else {
+            println!("Demanding");
             // todo: handle error case of try_recv
             // todo: on_pull make demand from the upper
             let demand = Demand {
                 stage_id: self.stage_id,
-                style: DemandStyle::DemandFull(100)
+                style: DemandStyle::DemandFull(100),
             };
             self.demand_tx.as_ref().unwrap().try_send(demand).unwrap();
         }
@@ -56,45 +76,57 @@ impl<I> InHandler for IgnoreHandler<I>
     }
 }
 
-impl<'a, I> GraphStage<'a> for Ignore<I>
-    where
-        I: Clone +  'static,
+impl<I> GraphStage for Ignore<I>
+where
+    I: Clone + 'static,
 {
     fn build_shape(&mut self) {
         let ignore_sink_inlet = Inlet::<I>::new(0, "Sink.out");
-        self.shape = SinkShape {
+        self.shape = Some(SinkShape {
             inlet: ignore_sink_inlet,
-        };
+        });
     }
 
-    fn build_demand(&'a mut self, tx: BroadcastSender<Demand>, rx: BroadcastReceiver<Demand>) {
-        self.demand_tx = tx;
-        self.demand_rx = rx;
+    fn build_demand(&mut self, tx: BroadcastSender<Demand>, rx: BroadcastReceiver<Demand>) {
+        self.demand_tx = Some(tx);
+        self.demand_rx = Some(rx);
     }
 
-    fn create_logic(&mut self, _attributes: Attributes) -> GraphStageLogic {
+    fn create_logic(&mut self, stage_id: usize, _attributes: Attributes) {
         self.build_shape();
 
         let (tx, rx) = unbounded::<I>();
 
-        self.in_handler = Box::new(IgnoreHandler {
+        self.in_handler = Some(Box::new(IgnoreHandler {
             in_tx: Some(tx),
             in_rx: Some(rx),
-            demand_rx: Some(self.demand_rx.clone()),
-            demand_tx: Some(self.demand_tx.clone()),
-            stage_id: self.stage_id
-        });
+            demand_rx: Some(self.demand_rx.as_ref().unwrap().clone()),
+            demand_tx: Some(self.demand_tx.as_ref().unwrap().clone()),
+            stage_id,
+        }));
+
+        self.stage_id = Some(stage_id);
 
-        let shape = Box::new(self.shape.clone());
+        let shape = Box::new(self.shape.as_ref().unwrap().clone());
 
         let mut gsl = GraphStageLogic::from_shape::<I, NotUsed>(shape);
-        gsl.set_inlet_handler(self.shape.inlet.clone(), self.in_handler.clone());
-        self.logic = gsl.clone();
-        gsl
+        gsl.set_inlet_handler(
+            self.shape.as_ref().unwrap().inlet.clone(),
+            self.in_handler.as_ref().unwrap().clone(),
+        );
+        self.logic = Some(gsl);
     }
 
-    fn get_shape(&'a self) -> ShapeType {
-        let shape: &dyn Shape<I, NotUsed> = &self.shape;
+    fn get_shape(&self) -> ShapeType {
+        let shape: &dyn Shape<I, NotUsed> = self.shape.as_ref().unwrap();
         shape.shape_type()
     }
+
+    fn get_stage_id(&self) -> usize {
+        self.stage_id.unwrap()
+    }
+
+    fn get_logic(&self) -> &GraphStageLogic {
+        self.logic.as_ref().unwrap()
+    }
 }