update code format

Author: luoluoyuyu

src/runtime/input/mod.rs

@@ -11,11 +11,11 @@
 // limitations under the License.
 
 mod input_protocol;
-mod interface;
 mod input_provider;
 mod input_runner;
+mod interface;
 pub mod protocol;
 
-pub use interface::{Input, InputState};
 pub use input_provider::InputProvider;
 pub use input_runner::InputRunner;
+pub use interface::{Input, InputState};

src/runtime/output/mod.rs

@@ -10,15 +10,15 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-mod output_protocol;
 mod interface;
+mod output_protocol;
 mod output_provider;
 mod output_runner;
 mod protocol;
 
+pub use interface::Output;
 #[allow(unused_imports)]
 pub use output_protocol::OutputProtocol;
-pub use interface::Output;
 pub use output_provider::OutputProvider;
 #[allow(unused_imports)]
 pub use output_runner::OutputRunner;

src/runtime/output/output_runner.rs

@@ -306,7 +306,11 @@ impl<P: OutputProtocol> Output for OutputRunner<P> {
         let output_id = self.output_id;
 
         let thread_handle = thread::Builder::new()
-            .name(format!("output-runner-{}-{}", protocol.name(), self.output_id))
+            .name(format!(
+                "output-runner-{}-{}",
+                protocol.name(),
+                self.output_id
+            ))
             .spawn(move || {
                 Self::worker_loop(protocol, d_r, c_r, c_s, state, mail_box, output_id);
             })

src/runtime/processor/wasm/wasm_host.rs

@@ -410,12 +410,9 @@ impl HostIterator for HostState {
 impl functionstream::core::collector::Host for HostState {
     fn emit(&mut self, target_id: u32, data: Vec<u8>) {
         let output_count = self.outputs.len();
-        let out = self
-            .outputs
-            .get_mut(target_id as usize)
-            .unwrap_or_else(|| {
-                panic!("Invalid target_id: {target_id}, available outputs: {output_count}");
-            });
+        let out = self.outputs.get_mut(target_id as usize).unwrap_or_else(|| {
+            panic!("Invalid target_id: {target_id}, available outputs: {output_count}");
+        });
 
         let buffer_or_event =
             BufferOrEvent::new_buffer(data, Some(format!("target_{}", target_id)), false, false);
@@ -427,12 +424,9 @@ impl functionstream::core::collector::Host for HostState {
 
     fn emit_watermark(&mut self, target_id: u32, ts: u64) {
         let output_count = self.outputs.len();
-        let out = self
-            .outputs
-            .get_mut(target_id as usize)
-            .unwrap_or_else(|| {
-                panic!("Invalid target_id: {target_id}, available outputs: {output_count}");
-            });
+        let out = self.outputs.get_mut(target_id as usize).unwrap_or_else(|| {
+            panic!("Invalid target_id: {target_id}, available outputs: {output_count}");
+        });
 
         let mut watermark_data = Vec::with_capacity(12);
         watermark_data.extend_from_slice(&target_id.to_le_bytes());

src/runtime/processor/wasm/wasm_processor.rs

@@ -464,31 +464,26 @@ impl WasmProcessor for WasmProcessorImpl {
                 .first()
                 .map(|(_, b)| b.as_slice())
                 .unwrap_or(&[]);
-            create_wasm_host(
-                first_bytes,
-                outputs,
-                init_context,
-                task_name,
-                create_time,
-            )
-            .map_err(|e| -> Box<dyn Error + Send> {
-                let error_msg = format!("Failed to create WasmHost: {}", e);
-                log::error!("{}", error_msg);
-                let mut full_error = error_msg.clone();
-                let mut source = e.source();
-                let mut depth = 0;
-                while let Some(err) = source {
-                    depth += 1;
-                    full_error.push_str(&format!("\n  Caused by ({}): {}", depth, err));
-                    source = err.source();
-                    if depth > 10 {
-                        full_error.push_str("\n  ... (error chain too long, truncated)");
-                        break;
+            create_wasm_host(first_bytes, outputs, init_context, task_name, create_time).map_err(
+                |e| -> Box<dyn Error + Send> {
+                    let error_msg = format!("Failed to create WasmHost: {}", e);
+                    log::error!("{}", error_msg);
+                    let mut full_error = error_msg.clone();
+                    let mut source = e.source();
+                    let mut depth = 0;
+                    while let Some(err) = source {
+                        depth += 1;
+                        full_error.push_str(&format!("\n  Caused by ({}): {}", depth, err));
+                        source = err.source();
+                        if depth > 10 {
+                            full_error.push_str("\n  ... (error chain too long, truncated)");
+                            break;
+                        }
                     }
-                }
-                log::error!("Full error chain:\n{}", full_error);
-                Box::new(WasmProcessorError::InitError(full_error))
-            })?
+                    log::error!("Full error chain:\n{}", full_error);
+                    Box::new(WasmProcessorError::InitError(full_error))
+                },
+            )?
         };
 
         *self.processor.borrow_mut() = Some(processor);
@@ -616,7 +611,10 @@ impl WasmProcessor for WasmProcessorImpl {
         Ok(())
     }
 
-    fn finish_checkpoint_outputs(&mut self, checkpoint_id: u64) -> Result<(), Box<dyn Error + Send>> {
+    fn finish_checkpoint_outputs(
+        &mut self,
+        checkpoint_id: u64,
+    ) -> Result<(), Box<dyn Error + Send>> {
         let mut store_ref = self.store.borrow_mut();
         let store = store_ref.as_mut().ok_or_else(|| -> Box<dyn Error + Send> {
             Box::new(WasmProcessorError::InitError(