add ExampleASyncSource
@@ -11,7 +11,7 @@ mod source_example {
use common_datablocks::DataBlock;
use crate::pipelines::new::processors::port::OutputPort;
use crate::pipelines::new::processors::processor::ProcessorPtr;
- use crate::pipelines::new::processors::sources::{SyncSource, SyncSourceProcessorWrap};
+ use crate::pipelines::new::processors::sources::{AsyncSource, ASyncSourceProcessorWrap, SyncSource, SyncSourceProcessorWrap};
struct ExampleSyncSource {
pos: usize,
@@ -36,5 +36,30 @@ mod source_example {
}
}
}
+
+ struct ExampleAsyncSource {
+ pos: usize,
+ data_blocks: Vec<DataBlock>,
+ }
+
+ impl ExampleAsyncSource {
+ pub fn create(data_blocks: Vec<DataBlock>, outputs: Vec<OutputPort>) -> Result<ProcessorPtr> {
+ ASyncSourceProcessorWrap::create(outputs, ExampleAsyncSource {
+ pos: 0,
+ data_blocks,
+ })
+ }
+ }
+
+ #[async_trait::async_trait]
+ impl AsyncSource for ExampleAsyncSource {
+ async fn generate(&mut self) -> Result<Option<DataBlock>> {
+ self.pos += 1;
+ match self.data_blocks.len() >= self.pos {
+ true => Ok(Some(self.data_blocks[self.pos - 1].clone())),
+ false => Ok(None),
+ }
+ }
+ }
}
