Simplify pipeline API

maplibre-demo/src/main.rs

@@ -133,7 +133,7 @@ fn run_headless() {
             &mut pipeline_context,
         );
 
-        let mut processor = pipeline_context.teardown();
+        let mut processor = pipeline_context.take_processor();
 
         while let Some(v) = processor
             .as_any_mut()

maplibre/src/io/pipeline.rs

@@ -33,12 +33,27 @@ pub trait PipelineProcessor: Downcast {
 impl_downcast!(PipelineProcessor);
 
 pub struct PipelineContext {
-    pub processor: Box<dyn PipelineProcessor>,
+    processor: Box<dyn PipelineProcessor>,
 }
 
 impl PipelineContext {
-    pub fn teardown(self) -> Box<dyn PipelineProcessor> {
-        self.processor
+    pub fn new<P>(processor: P) -> Self
+    where
+        P: PipelineProcessor,
+    {
+        Self {
+            processor: Box::new(processor),
+        }
+    }
+
+    pub fn take_processor<P>(self) -> Option<Box<P>>
+    where
+        P: PipelineProcessor,
+    {
+        self.processor.into_any().downcast::<P>().ok()
+    }
+    pub fn processor_mut(&mut self) -> &mut dyn PipelineProcessor {
+        self.processor.as_mut()
     }
 }
 

maplibre/src/io/pipeline_steps.rs

@@ -39,7 +39,7 @@ impl Processable for IndexLayerStep {
         let index = IndexProcessor::new();
 
         context
-            .processor
+            .processor_mut()
             .finished_layer_indexing(&tile_request.coords, index.get_geometries());
         (tile_request, request_id, tile)
     }
@@ -70,7 +70,9 @@ impl Processable for TessellateLayerStep {
 
             let mut tessellator = ZeroTessellator::<IndexDataType>::default();
             if let Err(e) = layer.process(&mut tessellator) {
-                context.processor.unavailable_layer(coords, layer_name);
+                context
+                    .processor_mut()
+                    .unavailable_layer(coords, layer_name);
 
                 tracing::error!(
                     "layer {} at {} tesselation failed {:?}",
@@ -79,7 +81,7 @@ impl Processable for TessellateLayerStep {
                     e
                 );
             } else {
-                context.processor.finished_layer_tesselation(
+                context.processor_mut().finished_layer_tesselation(
                     coords,
                     tessellator.buffer.into(),
                     tessellator.feature_indices,
@@ -95,7 +97,9 @@ impl Processable for TessellateLayerStep {
             .collect::<HashSet<_>>();
 
         for missing_layer in tile_request.layers.difference(&available_layers) {
-            context.processor.unavailable_layer(coords, missing_layer);
+            context
+                .processor_mut()
+                .unavailable_layer(coords, missing_layer);
 
             tracing::info!(
                 "requested layer {} at {} not found in tile",
@@ -107,7 +111,7 @@ impl Processable for TessellateLayerStep {
         tracing::info!("tile tessellated at {} finished", &tile_request.coords);
 
         context
-            .processor
+            .processor_mut()
             .finished_tile_tesselation(request_id, &tile_request.coords);
 
         (tile_request, request_id, tile)
@@ -133,9 +137,7 @@ mod tests {
 
     #[test]
     fn test() {
-        let mut context = PipelineContext {
-            processor: Box::new(DummyPipelineProcessor),
-        };
+        let mut context = PipelineContext::new(DummyPipelineProcessor);
 
         let pipeline = build_vector_tile_pipeline();
         let output = pipeline.process(

maplibre/src/map_schedule.rs

@@ -101,7 +101,7 @@ where
         let http_source_client: HttpSourceClient<HC> = HttpSourceClient::new(http_client);
         register_stages(&mut schedule, http_source_client, Box::new(scheduler));
 
-        register_render_stages(&mut schedule, false);
+        register_render_stages(&mut schedule, false).unwrap();
 
         Self {
             map_window_config,
@@ -194,7 +194,7 @@ where
                 )
                 .await
                 .unwrap();
-                &self.map_context.make_full(renderer);
+                self.map_context.make_full(renderer);
                 true
             }
             EventuallyMapContext::_Uninitialized => false,
@@ -244,7 +244,7 @@ where
         let tile_repository = TileRepository::new();
         let mut schedule = Schedule::default();
 
-        register_render_stages(&mut schedule, true);
+        register_render_stages(&mut schedule, true).unwrap();
 
         Self {
             map_window_config,

maplibre/src/stages/message.rs

@@ -98,12 +98,9 @@ impl SharedThreadState {
     #[tracing::instrument(skip_all)]
     pub fn process_tile(&self, request_id: TileRequestID, data: Box<[u8]>) -> Result<(), Error> {
         if let Some(tile_request) = self.get_tile_request(request_id) {
-            let mut processor = HeadedPipelineProcessor {
+            let mut pipeline_context = PipelineContext::new(HeadedPipelineProcessor {
                 state: self.clone(),
-            };
-            let mut pipeline_context = PipelineContext {
-                processor: Box::new(processor),
-            };
+            });
             let pipeline = build_vector_tile_pipeline();
             pipeline.process((tile_request, request_id, data), &mut pipeline_context);
         }

maplibre/src/stages/mod.rs

@@ -74,12 +74,15 @@ impl PipelineProcessor for HeadedPipelineProcessor {
     }
 
     fn unavailable_layer(&mut self, coords: &WorldTileCoords, layer_name: &str) {
-        self.state.message_sender.send(TessellateMessage::Layer(
+        self.state
+            .message_sender
+            .send(TessellateMessage::Layer(
                 LayerTessellateMessage::UnavailableLayer {
                     coords: *coords,
                     layer_name: layer_name.to_owned(),
                 },
-        ));
+            ))
+            .unwrap();
     }
 
     fn finished_layer_tesselation(