HTTP middleware

Signed-off-by: itowlson <ivan.towlson@fermyon.com>

Author: itowlson

Cargo.lock

@@ -29,11 +29,17 @@ pub use spin_capabilities::InheritConfiguration;
 /// dependent component. Finally, the composer will export all exports from the
 /// dependent component to its dependents. The composer will then encode the
 /// composition graph into a byte array and return it.
-pub async fn compose<L: ComponentSourceLoader>(
+pub async fn compose<
+    L: ComponentSourceLoader,
+    Fut: std::future::Future<Output = Result<Vec<u8>, ComposeError>>,
+>(
     loader: &L,
     component: &L::Component,
+    apply_trigger_deps: impl Fn(Vec<u8>) -> Fut,
 ) -> Result<Vec<u8>, ComposeError> {
-    Composer::new(loader).compose(component).await
+    Composer::new(loader)
+        .compose(component, apply_trigger_deps)
+        .await
 }
 
 /// A Spin component dependency. This abstracts over the metadata associated with the
@@ -90,8 +96,10 @@ impl DependencyLike for spin_app::locked::LockedComponentDependency {
 pub trait ComponentSourceLoader {
     type Component: ComponentLike<Dependency = Self::Dependency>;
     type Dependency: DependencyLike;
+    type Source;
     async fn load_component_source(&self, source: &Self::Component) -> anyhow::Result<Vec<u8>>;
     async fn load_dependency_source(&self, source: &Self::Dependency) -> anyhow::Result<Vec<u8>>;
+    async fn load_source(&self, source: &Self::Source) -> anyhow::Result<Vec<u8>>;
 }
 
 /// A ComponentSourceLoader that loads component sources from the filesystem.
@@ -101,6 +109,7 @@ pub struct ComponentSourceLoaderFs;
 impl ComponentSourceLoader for ComponentSourceLoaderFs {
     type Component = spin_app::locked::LockedComponent;
     type Dependency = spin_app::locked::LockedComponentDependency;
+    type Source = spin_app::locked::LockedComponentSource;
 
     async fn load_component_source(&self, source: &Self::Component) -> anyhow::Result<Vec<u8>> {
         Self::load_from_locked_source(&source.source).await
@@ -109,6 +118,10 @@ impl ComponentSourceLoader for ComponentSourceLoaderFs {
     async fn load_dependency_source(&self, source: &Self::Dependency) -> anyhow::Result<Vec<u8>> {
         Self::load_from_locked_source(&source.source).await
     }
+
+    async fn load_source(&self, source: &Self::Source) -> anyhow::Result<Vec<u8>> {
+        Self::load_from_locked_source(source).await
+    }
 }
 
 impl ComponentSourceLoaderFs {
@@ -195,39 +208,47 @@ struct Composer<'a, L> {
 }
 
 impl<'a, L: ComponentSourceLoader> Composer<'a, L> {
-    async fn compose(mut self, component: &L::Component) -> Result<Vec<u8>, ComposeError> {
+    async fn compose<Fut: std::future::Future<Output = Result<Vec<u8>, ComposeError>>>(
+        mut self,
+        component: &L::Component,
+        apply_trigger_deps: impl Fn(Vec<u8>) -> Fut,
+    ) -> Result<Vec<u8>, ComposeError> {
         let source = self
             .loader
             .load_component_source(component)
             .await
             .map_err(ComposeError::PrepareError)?;
 
-        if component.dependencies().len() == 0 {
-            return Ok(source);
-        }
+        let fulfilled_source = if component.dependencies().len() == 0 {
+            source
+        } else {
+            let (world_id, instantiation_id) = self
+                .register_package(component.id(), None, source)
+                .map_err(ComposeError::PrepareError)?;
 
-        let (world_id, instantiation_id) = self
-            .register_package(component.id(), None, source)
-            .map_err(ComposeError::PrepareError)?;
+            let prepared = self.prepare_dependencies(world_id, component).await?;
 
-        let prepared = self.prepare_dependencies(world_id, component).await?;
+            let arguments = self
+                .build_instantiation_arguments(world_id, prepared)
+                .await?;
 
-        let arguments = self
-            .build_instantiation_arguments(world_id, prepared)
-            .await?;
+            for (argument_name, argument) in arguments {
+                self.graph
+                    .set_instantiation_argument(instantiation_id, &argument_name, argument)
+                    .map_err(|e| ComposeError::PrepareError(e.into()))?;
+            }
+
+            self.export_dependents_exports(world_id, instantiation_id)
+                .map_err(ComposeError::PrepareError)?;
 
-        for (argument_name, argument) in arguments {
             self.graph
-                .set_instantiation_argument(instantiation_id, &argument_name, argument)
-                .map_err(|e| ComposeError::PrepareError(e.into()))?;
-        }
+                .encode(Default::default())
+                .map_err(|e| ComposeError::EncodeError(e.into()))?
+        };
 
-        self.export_dependents_exports(world_id, instantiation_id)
-            .map_err(ComposeError::PrepareError)?;
+        let with_extras = apply_trigger_deps(fulfilled_source).await?;
 
-        self.graph
-            .encode(Default::default())
-            .map_err(|e| ComposeError::EncodeError(e.into()))
+        Ok(with_extras)
     }
 
     fn new(loader: &'a L) -> Self {

crates/compose/src/lib.rs

@@ -147,7 +147,7 @@ impl ApplicationToValidate {
 
         let loader = ComponentSourceLoader::new(&self.wasm_loader);
 
-        let wasm = spin_compose::compose(&loader, &component).await.with_context(|| format!("Spin needed to compose dependencies for {} as part of target checking, but composition failed", component.id))?;
+        let wasm = spin_compose::compose(&loader, &component, async |data| Ok(data)).await.with_context(|| format!("Spin needed to compose dependencies for {} as part of target checking, but composition failed", component.id))?;
 
         let host_requirements = if component.requires_service_chaining {
             vec!["local_service_chaining".to_string()]
@@ -185,6 +185,7 @@ impl<'a> ComponentSourceLoader<'a> {
 impl<'a> spin_compose::ComponentSourceLoader for ComponentSourceLoader<'a> {
     type Component = ComponentSource<'a>;
     type Dependency = WrappedComponentDependency;
+    type Source = spin_manifest::schema::v2::ComponentSource;
     async fn load_component_source(&self, source: &Self::Component) -> anyhow::Result<Vec<u8>> {
         let path = self
             .wasm_loader
@@ -210,6 +211,19 @@ impl<'a> spin_compose::ComponentSourceLoader for ComponentSourceLoader<'a> {
             .with_context(|| format!("componentizing {}", quoted_path(&path)))?;
         Ok(component.into())
     }
+
+    async fn load_source(&self, source: &Self::Source) -> anyhow::Result<Vec<u8>> {
+        let path = self
+            .wasm_loader
+            .load_component_source("in-memory-component", source)
+            .await?;
+        let bytes = tokio::fs::read(&path)
+            .await
+            .with_context(|| format!("reading {}", quoted_path(&path)))?;
+        let component = spin_componentize::componentize_if_necessary(&bytes)
+            .with_context(|| format!("componentizing {}", quoted_path(&path)))?;
+        Ok(component.into())
+    }
 }
 
 // This exists only to thwart the orphan rule

crates/environments/src/loader.rs

@@ -55,6 +55,7 @@ impl<T: RuntimeFactors, U: Send + 'static> FactorsExecutor<T, U> {
         runtime_config: T::RuntimeConfig,
         component_loader: &impl ComponentLoader<T, U>,
         trigger_type: Option<&str>,
+        trigger_dependencies_composer: impl TriggerDependenciesComposer,
     ) -> anyhow::Result<FactorsExecutorApp<T, U>> {
         let configured_app = self
             .factors
@@ -77,7 +78,11 @@ impl<T: RuntimeFactors, U: Send + 'static> FactorsExecutor<T, U> {
 
         for component in components {
             let instance_pre = component_loader
-                .load_instance_pre(&self.core_engine, &component)
+                .load_instance_pre(
+                    &self.core_engine,
+                    &component,
+                    &trigger_dependencies_composer,
+                )
                 .await?;
             component_instance_pres.insert(component.id().to_string(), instance_pre);
         }
@@ -116,19 +121,57 @@ pub trait ComponentLoader<T: RuntimeFactors, U>: Sync {
         &self,
         engine: &spin_core::wasmtime::Engine,
         component: &AppComponent,
+        trigger_dependencies_composer: &impl TriggerDependenciesComposer,
     ) -> anyhow::Result<Component>;
 
     /// Loads [`InstancePre`] for the given [`AppComponent`].
     async fn load_instance_pre(
         &self,
         engine: &spin_core::Engine<InstanceState<T::InstanceState, U>>,
         component: &AppComponent,
+        trigger_dependencies_composer: &impl TriggerDependenciesComposer,
     ) -> anyhow::Result<spin_core::InstancePre<InstanceState<T::InstanceState, U>>> {
-        let component = self.load_component(engine.as_ref(), component).await?;
+        let component = self
+            .load_component(engine.as_ref(), component, trigger_dependencies_composer)
+            .await?;
         engine.instantiate_pre(&component)
     }
 }
 
+#[async_trait]
+pub trait TriggerDependenciesComposer: Send + Sync {
+    async fn compose_trigger_dependencies(
+        &self,
+        trigger_dependencies: &HashMap<String, Vec<TriggerDependency>>,
+        component: Vec<u8>,
+    ) -> anyhow::Result<Vec<u8>>;
+}
+
+#[async_trait]
+impl TriggerDependenciesComposer for () {
+    async fn compose_trigger_dependencies(
+        &self,
+        trigger_dependencies: &HashMap<String, Vec<TriggerDependency>>,
+        component: Vec<u8>,
+    ) -> anyhow::Result<Vec<u8>> {
+        if trigger_dependencies.is_empty() {
+            Ok(component)
+        } else {
+            Err(anyhow::anyhow!("this trigger should not have dependencies"))
+        }
+    }
+}
+
+pub struct TriggerDependency {
+    pub data: TriggerDependencyData,
+    pub dependency: spin_app::locked::LockedComponentDependency,
+}
+
+pub enum TriggerDependencyData {
+    InMemory(Vec<u8>),
+    OnDisk(std::path::PathBuf),
+}
+
 type InstancePre<T, U> =
     spin_core::InstancePre<InstanceState<<T as RuntimeFactors>::InstanceState, U>>;
 
@@ -437,7 +480,7 @@ mod tests {
         let executor = Arc::new(FactorsExecutor::new(engine_builder, env.factors)?);
 
         let factors_app = executor
-            .load_app(app, Default::default(), &DummyComponentLoader, None)
+            .load_app(app, Default::default(), &DummyComponentLoader, None, ())
             .await?;
 
         let mut instance_builder = factors_app.prepare("empty")?;
@@ -463,6 +506,7 @@ mod tests {
             &self,
             engine: &spin_core::wasmtime::Engine,
             _component: &AppComponent,
+            _trigger_dependencies_composer: &impl TriggerDependenciesComposer,
         ) -> anyhow::Result<Component> {
             Ok(Component::new(engine, "(component)")?)
         }