update input and output

Author: luoluoyuyu

src/runtime/input/input.rs

@@ -0,0 +1,49 @@
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::runtime::buffer_and_event::BufferOrEvent;
+use crate::runtime::taskexecutor::InitContext;
+
+pub use crate::runtime::common::ComponentState as InputState;
+
+pub trait Input: Send + Sync {
+    fn init_with_context(
+        &mut self,
+        init_context: &InitContext,
+    ) -> Result<(), Box<dyn std::error::Error + Send>>;
+
+    fn start(&mut self) -> Result<(), Box<dyn std::error::Error + Send>>;
+
+    fn stop(&mut self) -> Result<(), Box<dyn std::error::Error + Send>>;
+
+    fn close(&mut self) -> Result<(), Box<dyn std::error::Error + Send>>;
+
+    fn get_next(&mut self) -> Result<Option<BufferOrEvent>, Box<dyn std::error::Error + Send>>;
+
+    fn poll_next(&mut self) -> Result<Option<BufferOrEvent>, Box<dyn std::error::Error + Send>> {
+        self.get_next()
+    }
+
+    fn take_checkpoint(
+        &mut self,
+        checkpoint_id: u64,
+    ) -> Result<(), Box<dyn std::error::Error + Send>>;
+
+    fn finish_checkpoint(
+        &mut self,
+        checkpoint_id: u64,
+    ) -> Result<(), Box<dyn std::error::Error + Send>>;
+
+    fn get_group_id(&self) -> usize;
+
+    fn set_error_state(&self) -> Result<(), Box<dyn std::error::Error + Send>>;
+}

src/runtime/input/input_protocol.rs

@@ -0,0 +1,38 @@
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::runtime::buffer_and_event::BufferOrEvent;
+use std::time::Duration;
+
+pub trait InputProtocol: Send + Sync + 'static {
+    fn init(&self) -> Result<(), Box<dyn std::error::Error + Send>>;
+    fn poll(
+        &self,
+        timeout: Duration,
+    ) -> Result<Option<BufferOrEvent>, Box<dyn std::error::Error + Send>>;
+    fn name(&self) -> String;
+    fn on_start(&self) -> Result<(), Box<dyn std::error::Error + Send>> {
+        Ok(())
+    }
+    fn on_stop(&self) -> Result<(), Box<dyn std::error::Error + Send>> {
+        Ok(())
+    }
+    fn on_close(&self) -> Result<(), Box<dyn std::error::Error + Send>> {
+        Ok(())
+    }
+    fn on_checkpoint(&self, _id: u64) -> Result<(), Box<dyn std::error::Error + Send>> {
+        Ok(())
+    }
+    fn on_checkpoint_finish(&self, _id: u64) -> Result<(), Box<dyn std::error::Error + Send>> {
+        Ok(())
+    }
+}

…c/runtime/input/input_source_provider.rs src/runtime/input/input_provider.rssrc/runtime/input/input_source_provider.rs renamed to src/runtime/input/input_provider.rs src/runtime/input/input_source_provider.rs renamed to src/runtime/input/input_provider.rs

@@ -10,32 +10,16 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-// InputSourceProvider - Input source provider
-//
-// Creates InputSource instances from configuration objects
-
-use crate::runtime::input::InputSource;
+use crate::runtime::input::Input;
 use crate::runtime::task::InputConfig;
 
-/// InputSourceProvider - Input source provider
-///
-/// Creates InputSource instances from configuration objects
-pub struct InputSourceProvider;
+pub struct InputProvider;
 
-impl InputSourceProvider {
-    /// Create multiple InputSource from InputConfig list
-    ///
-    /// # Arguments
-    /// - `input_configs`: InputConfig list
-    /// - `group_idx`: Input group index (used to identify which group the input source belongs to)
-    ///
-    /// # Returns
-    /// - `Ok(Vec<Box<dyn InputSource>>)`: Successfully created input source list
-    /// - `Err(...)`: Configuration parsing or creation failed
+impl InputProvider {
     pub fn from_input_configs(
         input_configs: &[InputConfig],
         group_idx: usize,
-    ) -> Result<Vec<Box<dyn InputSource>>, Box<dyn std::error::Error + Send>> {
+    ) -> Result<Vec<Box<dyn Input>>, Box<dyn std::error::Error + Send>> {
         if input_configs.is_empty() {
             return Err(Box::new(std::io::Error::new(
                 std::io::ErrorKind::InvalidData,
@@ -46,7 +30,6 @@ impl InputSourceProvider {
             )) as Box<dyn std::error::Error + Send>);
         }
 
-        // Check input source count limit (maximum 64)
         const MAX_INPUTS: usize = 64;
         if input_configs.len() > MAX_INPUTS {
             return Err(Box::new(std::io::Error::new(
@@ -60,7 +43,6 @@ impl InputSourceProvider {
             )) as Box<dyn std::error::Error + Send>);
         }
 
-        // Create InputSource for each InputConfig
         let mut inputs = Vec::new();
         for (input_idx, input_config) in input_configs.iter().enumerate() {
             let input = Self::from_input_config(input_config, group_idx, input_idx)?;
@@ -70,21 +52,11 @@ impl InputSourceProvider {
         Ok(inputs)
     }
 
-    /// Create InputSource from single InputConfig
-    ///
-    /// # Arguments
-    /// - `input_config`: Input source configuration
-    /// - `group_idx`: Input group index (used to identify which group the input source belongs to)
-    /// - `input_idx`: Input source index within group (used to identify different input sources within the same group)
-    ///
-    /// # Returns
-    /// - `Ok(Box<dyn InputSource>)`: Successfully created input source
-    /// - `Err(...)`: Parsing failed
     fn from_input_config(
         input_config: &InputConfig,
         group_idx: usize,
         input_idx: usize,
-    ) -> Result<Box<dyn InputSource>, Box<dyn std::error::Error + Send>> {
+    ) -> Result<Box<dyn Input>, Box<dyn std::error::Error + Send>> {
         match input_config {
             InputConfig::Kafka {
                 bootstrap_servers,
@@ -93,9 +65,9 @@ impl InputSourceProvider {
                 group_id,
                 extra,
             } => {
-                use crate::runtime::input::protocol::kafka::{KafkaConfig, KafkaInputSource};
+                use crate::runtime::input::InputRunner;
+                use crate::runtime::input::protocol::kafka::{KafkaConfig, KafkaProtocol};
 
-                // Convert bootstrap_servers string to Vec<String>
                 let servers: Vec<String> = bootstrap_servers
                     .split(',')
                     .map(|s| s.trim().to_string())
@@ -114,13 +86,8 @@ impl InputSourceProvider {
                     )) as Box<dyn std::error::Error + Send>);
                 }
 
-                // Convert partition from Option<u32> to Option<i32>
                 let partition_i32 = partition.map(|p| p as i32);
-
-                // Merge extra configuration into properties
                 let properties = extra.clone();
-
-                // Create KafkaConfig
                 let kafka_config = KafkaConfig::new(
                     servers,
                     topic.clone(),
@@ -129,9 +96,8 @@ impl InputSourceProvider {
                     properties,
                 );
 
-                // Create KafkaInputSource, pass in group_idx and input_idx
-                Ok(Box::new(KafkaInputSource::from_config(
-                    kafka_config,
+                Ok(Box::new(InputRunner::new(
+                    KafkaProtocol::new(kafka_config),
                     group_idx,
                     input_idx,
                 )))