fix(composition): general cleanup

src/composition/run_composition.rs

@@ -143,9 +143,14 @@ mod tests {
             .read_file(mock_read_file)
             .build();
 
-        let subgraph_change_events: BoxStream<SubgraphEvent> =
-            once(async { SubgraphEvent::SubgraphChanged(SubgraphSchemaChanged::default()) })
-                .boxed();
+        let subgraph_change_events: BoxStream<SubgraphEvent> = once(async {
+            SubgraphEvent::SubgraphChanged(SubgraphSchemaChanged::new(
+                "some-name".to_string(),
+                "some sdl".to_string(),
+            ))
+        })
+        .boxed();
+
         let (mut composition_messages, composition_subtask) = Subtask::new(composition_handler);
         let abort_handle = composition_subtask.run(subgraph_change_events);
 

src/composition/runner.rs

@@ -213,14 +213,21 @@ pub enum SubgraphEvent {
 }
 /// An event denoting that the subgraph has changed, emitting its name and the SDL reflecting that
 /// change
-#[derive(derive_getters::Getters, Default)]
+#[derive(derive_getters::Getters)]
 pub struct SubgraphSchemaChanged {
     /// Subgraph name
     name: String,
     /// SDL with changes
     sdl: String,
 }
 
+#[cfg(test)]
+impl SubgraphSchemaChanged {
+    pub fn new(name: String, sdl: String) -> Self {
+        Self { name, sdl }
+    }
+}
+
 /// The subgraph is no longer watched
 #[derive(derive_getters::Getters, Default)]
 pub struct SubgraphSchemaRemoved {
@@ -309,17 +316,20 @@ impl SubtaskHandleStream for SubgraphWatchers {
                                 Subtask::<SubgraphWatcher, WatchedSdlChange>::new(subgraph_watcher);
 
                             let sender = sender.clone();
-                            let subgraph_name_c = subgraph_name.clone();
-                            let messages_abort_handle = tokio::spawn(async move {
-                                while let Some(change) = messages.next().await {
-                                    let _ = sender
-                                        .send(SubgraphEvent::SubgraphChanged(
-                                            SubgraphSchemaChanged {
-                                                name: subgraph_name_c.to_string(),
-                                                sdl: change.sdl().to_string(),
-                                            },
-                                        ))
-                                        .tap_err(|err| tracing::error!("{:?}", err));
+                            let messages_abort_handle = tokio::spawn({
+                                let subgraph_name = subgraph_name.clone();
+
+                                async move {
+                                    while let Some(change) = messages.next().await {
+                                        let _ = sender
+                                            .send(SubgraphEvent::SubgraphChanged(
+                                                SubgraphSchemaChanged {
+                                                    name: subgraph_name.to_string(),
+                                                    sdl: change.sdl().to_string(),
+                                                },
+                                            ))
+                                            .tap_err(|err| tracing::error!("{:?}", err));
+                                    }
                                 }
                             })
                             .abort_handle();