Renaming ProcessorMailbox SourceSink

Author: fulmicoton-dd

quickwit/quickwit-indexing/src/source/file_source.rs

@@ -25,7 +25,7 @@ use quickwit_proto::types::SourceId;
 use super::doc_file_reader::ObjectUriBatchReader;
 #[cfg(feature = "queue-sources")]
 use super::queue_sources::coordinator::QueueCoordinator;
-use crate::source::{ProcessorMailbox, Source, SourceContext, SourceRuntime, TypedSourceFactory};
+use crate::source::{Source, SourceContext, SourceRuntime, SourceSink, TypedSourceFactory};
 
 enum FileSourceState {
     #[cfg(feature = "queue-sources")]
@@ -54,13 +54,13 @@ impl Source for FileSource {
     #[allow(unused_variables)]
     async fn initialize(
         &mut self,
-        processor_mailbox: &ProcessorMailbox,
+        source_sink: &SourceSink,
         ctx: &SourceContext,
     ) -> Result<(), ActorExitStatus> {
         match &mut self.state {
             #[cfg(feature = "queue-sources")]
             FileSourceState::Notification(coordinator) => {
-                coordinator.initialize(processor_mailbox, ctx).await
+                coordinator.initialize(source_sink, ctx).await
             }
             FileSourceState::Filepath { .. } => Ok(()),
         }
@@ -69,13 +69,13 @@ impl Source for FileSource {
     #[allow(unused_variables)]
     async fn emit_batches(
         &mut self,
-        processor_mailbox: &ProcessorMailbox,
+        source_sink: &SourceSink,
         ctx: &SourceContext,
     ) -> Result<Duration, ActorExitStatus> {
         match &mut self.state {
             #[cfg(feature = "queue-sources")]
             FileSourceState::Notification(coordinator) => {
-                coordinator.emit_batches(processor_mailbox, ctx).await?;
+                coordinator.emit_batches(source_sink, ctx).await?;
             }
             FileSourceState::Filepath {
                 batch_reader,
@@ -87,11 +87,11 @@ impl Source for FileSource {
                     .await?;
                 *num_bytes_processed += batch_builder.num_bytes;
                 *num_lines_processed += batch_builder.docs.len() as u64;
-                processor_mailbox
+                source_sink
                     .send_raw_doc_batch(batch_builder.build(), ctx)
                     .await?;
                 if batch_reader.is_eof() {
-                    processor_mailbox.send_exit_with_success(ctx).await?;
+                    source_sink.send_exit_with_success(ctx).await?;
                     return Err(ActorExitStatus::Success);
                 }
             }

quickwit/quickwit-indexing/src/source/gcp_pubsub_source.rs

@@ -34,7 +34,7 @@ use tracing::{debug, info, warn};
 
 use super::{BATCH_NUM_BYTES_LIMIT, EMIT_BATCHES_TIMEOUT};
 use crate::source::{
-    BatchBuilder, ProcessorMailbox, Source, SourceContext, SourceRuntime, TypedSourceFactory,
+    BatchBuilder, Source, SourceContext, SourceRuntime, SourceSink, TypedSourceFactory,
 };
 
 const DEFAULT_MAX_MESSAGES_PER_PULL: i32 = 1_000;
@@ -157,7 +157,7 @@ impl GcpPubSubSource {
 impl Source for GcpPubSubSource {
     async fn emit_batches(
         &mut self,
-        processor_mailbox: &ProcessorMailbox,
+        source_sink: &SourceSink,
         ctx: &SourceContext,
     ) -> Result<Duration, ActorExitStatus> {
         let now = Instant::now();
@@ -192,7 +192,7 @@ impl Source for GcpPubSubSource {
         // TODO: need to wait for all the id to be ack for at_least_once
         if self.should_exit() {
             info!(subscription=%self.subscription_name, "reached end of subscription");
-            processor_mailbox.send_exit_with_success(ctx).await?;
+            source_sink.send_exit_with_success(ctx).await?;
             return Err(ActorExitStatus::Success);
         }
         if !batch_builder.checkpoint_delta.is_empty() {
@@ -202,7 +202,7 @@ impl Source for GcpPubSubSource {
                 num_millis=%now.elapsed().as_millis(),
                 "Sending doc batch to indexer.");
             let message = batch_builder.build();
-            processor_mailbox.send_raw_doc_batch(message, ctx).await?;
+            source_sink.send_raw_doc_batch(message, ctx).await?;
         }
         Ok(Duration::default())
     }

quickwit/quickwit-indexing/src/source/ingest/mod.rs

@@ -46,8 +46,8 @@ use tracing::{debug, error, info, warn};
 use ulid::Ulid;
 
 use super::{
-    BATCH_NUM_BYTES_LIMIT, BatchBuilder, EMIT_BATCHES_TIMEOUT, ProcessorMailbox, Source,
-    SourceContext, SourceRuntime, TypedSourceFactory,
+    BATCH_NUM_BYTES_LIMIT, BatchBuilder, EMIT_BATCHES_TIMEOUT, Source, SourceContext, SourceRuntime,
+    SourceSink, TypedSourceFactory,
 };
 use crate::models::{LocalShardPositionsUpdate, NewPublishLock, NewPublishToken, PublishLock};
 
@@ -393,7 +393,7 @@ impl IngestSource {
     async fn reset_if_needed(
         &mut self,
         new_assigned_shard_ids: &BTreeSet<ShardId>,
-        doc_processor_mailbox: &ProcessorMailbox,
+        source_sink: &SourceSink,
         ctx: &SourceContext,
     ) -> anyhow::Result<()> {
         // No need to do anything if the list of shards before and after are empty.
@@ -440,10 +440,10 @@ impl IngestSource {
         self.publish_lock.kill().await;
         self.publish_lock = PublishLock::default();
         self.publish_token = self.client_id.new_publish_token();
-        doc_processor_mailbox
+        source_sink
             .send_publish_lock(NewPublishLock(self.publish_lock.clone()), ctx)
             .await?;
-        doc_processor_mailbox
+        source_sink
             .send_publish_token(NewPublishToken(self.publish_token.clone()), ctx)
             .await?;
         Ok(())
@@ -454,7 +454,7 @@ impl IngestSource {
 impl Source for IngestSource {
     async fn emit_batches(
         &mut self,
-        processor_mailbox: &ProcessorMailbox,
+        source_sink: &SourceSink,
         ctx: &SourceContext,
     ) -> Result<Duration, ActorExitStatus> {
         let mut batch_builder = BatchBuilder::new(SourceType::IngestV2);
@@ -497,18 +497,18 @@ impl Source for IngestSource {
                 "Sending doc batch to indexer."
             );
             let message = batch_builder.build();
-            processor_mailbox.send_raw_doc_batch(message, ctx).await?;
+            source_sink.send_raw_doc_batch(message, ctx).await?;
         }
         Ok(Duration::default())
     }
 
     async fn assign_shards(
         &mut self,
         new_assigned_shard_ids: BTreeSet<ShardId>,
-        processor_mailbox: &ProcessorMailbox,
+        source_sink: &SourceSink,
         ctx: &SourceContext,
     ) -> anyhow::Result<()> {
-        self.reset_if_needed(&new_assigned_shard_ids, processor_mailbox, ctx)
+        self.reset_if_needed(&new_assigned_shard_ids, source_sink, ctx)
             .await?;
 
         // As enforced by `reset_if_needed`, at this point, all currently assigned shards should be
@@ -952,7 +952,7 @@ mod tests {
         let (source_mailbox, _source_inbox) = universe.create_test_mailbox::<SourceActor>();
         let (doc_processor_mailbox, doc_processor_inbox) =
             universe.create_test_mailbox::<DocProcessor>();
-        let processor_mailbox = ProcessorMailbox::new(doc_processor_mailbox.clone());
+        let source_sink = SourceSink::new(doc_processor_mailbox.clone());
         let (observable_state_tx, _observable_state_rx) = watch::channel(serde_json::Value::Null);
         let ctx: SourceContext =
             ActorContext::for_test(&universe, source_mailbox, observable_state_tx);
@@ -962,7 +962,7 @@ mod tests {
         let shard_ids: BTreeSet<ShardId> = once(0).map(ShardId::from).collect();
         let publish_lock = source.publish_lock.clone();
         source
-            .assign_shards(shard_ids, &processor_mailbox, &ctx)
+            .assign_shards(shard_ids, &source_sink, &ctx)
             .await
             .unwrap();
         assert_eq!(sequence_rx.recv().await.unwrap(), 1);
@@ -976,7 +976,7 @@ mod tests {
         let shard_ids: BTreeSet<ShardId> = (0..2).map(ShardId::from).collect();
         let publish_lock = source.publish_lock.clone();
         source
-            .assign_shards(shard_ids, &processor_mailbox, &ctx)
+            .assign_shards(shard_ids, &source_sink, &ctx)
             .await
             .unwrap();
         assert_eq!(sequence_rx.recv().await.unwrap(), 2);
@@ -989,7 +989,7 @@ mod tests {
         let shard_ids: BTreeSet<ShardId> = (1..3).map(ShardId::from).collect();
         let publish_lock = source.publish_lock.clone();
         source
-            .assign_shards(shard_ids, &processor_mailbox, &ctx)
+            .assign_shards(shard_ids, &source_sink, &ctx)
             .await
             .unwrap();
 
@@ -1158,7 +1158,7 @@ mod tests {
         let (source_mailbox, _source_inbox) = universe.create_test_mailbox::<SourceActor>();
         let (doc_processor_mailbox, _doc_processor_inbox) =
             universe.create_test_mailbox::<DocProcessor>();
-        let processor_mailbox = ProcessorMailbox::new(doc_processor_mailbox.clone());
+        let source_sink = SourceSink::new(doc_processor_mailbox.clone());
         let (observable_state_tx, _observable_state_rx) = watch::channel(serde_json::Value::Null);
         let ctx: SourceContext =
             ActorContext::for_test(&universe, source_mailbox, observable_state_tx);
@@ -1168,7 +1168,7 @@ mod tests {
             BTreeSet::from_iter([ShardId::from(1), ShardId::from(2)]);
 
         source
-            .assign_shards(shard_ids, &processor_mailbox, &ctx)
+            .assign_shards(shard_ids, &source_sink, &ctx)
             .await
             .unwrap();
 
@@ -1325,7 +1325,7 @@ mod tests {
         let (source_mailbox, _source_inbox) = universe.create_test_mailbox::<SourceActor>();
         let (doc_processor_mailbox, _doc_processor_inbox) =
             universe.create_test_mailbox::<DocProcessor>();
-        let processor_mailbox = ProcessorMailbox::new(doc_processor_mailbox.clone());
+        let source_sink = SourceSink::new(doc_processor_mailbox.clone());
         let (observable_state_tx, _observable_state_rx) = watch::channel(serde_json::Value::Null);
         let ctx: SourceContext =
             ActorContext::for_test(&universe, source_mailbox, observable_state_tx);
@@ -1339,7 +1339,7 @@ mod tests {
 
         // In this scenario, the indexer will only be able to acquire shard 1.
         source
-            .assign_shards(shard_ids, &processor_mailbox, &ctx)
+            .assign_shards(shard_ids, &source_sink, &ctx)
             .await
             .unwrap();
 
@@ -1392,7 +1392,7 @@ mod tests {
         let (source_mailbox, _source_inbox) = universe.create_test_mailbox::<SourceActor>();
         let (doc_processor_mailbox, doc_processor_inbox) =
             universe.create_test_mailbox::<DocProcessor>();
-        let processor_mailbox = ProcessorMailbox::new(doc_processor_mailbox.clone());
+        let source_sink = SourceSink::new(doc_processor_mailbox.clone());
         let (observable_state_tx, _observable_state_rx) = watch::channel(serde_json::Value::Null);
         let ctx: SourceContext =
             ActorContext::for_test(&universe, source_mailbox, observable_state_tx);
@@ -1472,7 +1472,7 @@ mod tests {
         );
         fetch_message_tx.send(Ok(in_flight_value)).await.unwrap();
 
-        source.emit_batches(&processor_mailbox, &ctx).await.unwrap();
+        source.emit_batches(&source_sink, &ctx).await.unwrap();
         let doc_batch = doc_processor_inbox
             .recv_typed_message::<RawDocBatch>()
             .await
@@ -1498,7 +1498,7 @@ mod tests {
         assert_eq!(partition_deltas[1].1.from, Position::offset(22u64));
         assert_eq!(partition_deltas[1].1.to, Position::eof(23u64));
 
-        source.emit_batches(&processor_mailbox, &ctx).await.unwrap();
+        source.emit_batches(&source_sink, &ctx).await.unwrap();
         let shard = source.assigned_shards.get(&ShardId::from(2)).unwrap();
         assert_eq!(shard.status, IndexingStatus::ReachedEof);
 
@@ -1512,7 +1512,7 @@ mod tests {
             .await
             .unwrap();
 
-        source.emit_batches(&processor_mailbox, &ctx).await.unwrap();
+        source.emit_batches(&source_sink, &ctx).await.unwrap();
         let shard = source.assigned_shards.get(&ShardId::from(1)).unwrap();
         assert_eq!(shard.status, IndexingStatus::Error);
 
@@ -1533,7 +1533,7 @@ mod tests {
         );
         fetch_message_tx.send(Ok(in_flight_value)).await.unwrap();
 
-        source.emit_batches(&processor_mailbox, &ctx).await.unwrap();
+        source.emit_batches(&source_sink, &ctx).await.unwrap();
         let shard = source.assigned_shards.get(&ShardId::from(1)).unwrap();
         assert_eq!(shard.status, IndexingStatus::Active);
     }
@@ -1616,19 +1616,19 @@ mod tests {
         let (source_mailbox, _source_inbox) = universe.create_test_mailbox::<SourceActor>();
         let (doc_processor_mailbox, doc_processor_inbox) =
             universe.create_test_mailbox::<DocProcessor>();
-        let processor_mailbox = ProcessorMailbox::new(doc_processor_mailbox.clone());
+        let source_sink = SourceSink::new(doc_processor_mailbox.clone());
         let (observable_state_tx, _observable_state_rx) = watch::channel(serde_json::Value::Null);
         let ctx: SourceContext =
             ActorContext::for_test(&universe, source_mailbox, observable_state_tx);
 
         let shard_ids: BTreeSet<ShardId> = BTreeSet::from_iter([ShardId::from(1)]);
 
         source
-            .assign_shards(shard_ids, &processor_mailbox, &ctx)
+            .assign_shards(shard_ids, &source_sink, &ctx)
             .await
             .unwrap();
 
-        source.emit_batches(&processor_mailbox, &ctx).await.unwrap();
+        source.emit_batches(&source_sink, &ctx).await.unwrap();
 
         let shard = source.assigned_shards.get(&ShardId::from(1)).unwrap();
         assert_eq!(shard.status, IndexingStatus::NotFound);
@@ -1904,7 +1904,7 @@ mod tests {
         let (source_mailbox, _source_inbox) = universe.create_test_mailbox::<SourceActor>();
         let (doc_processor_mailbox, _doc_processor_inbox) =
             universe.create_test_mailbox::<DocProcessor>();
-        let processor_mailbox = ProcessorMailbox::new(doc_processor_mailbox.clone());
+        let source_sink = SourceSink::new(doc_processor_mailbox.clone());
         let (observable_state_tx, _observable_state_rx) = watch::channel(serde_json::Value::Null);
         let ctx: SourceContext =
             ActorContext::for_test(&universe, source_mailbox, observable_state_tx);
@@ -1919,7 +1919,7 @@ mod tests {
         });
 
         source
-            .assign_shards(shard_ids, &processor_mailbox, &ctx)
+            .assign_shards(shard_ids, &source_sink, &ctx)
             .await
             .unwrap();
 

quickwit/quickwit-indexing/src/source/ingest_api_source.rs

@@ -29,7 +29,7 @@ use serde::Serialize;
 use serde_json::Value as JsonValue;
 use tracing::{error, info};
 
-use super::{BatchBuilder, ProcessorMailbox, Source, SourceContext, TypedSourceFactory};
+use super::{BatchBuilder, Source, SourceContext, SourceSink, TypedSourceFactory};
 use crate::source::SourceRuntime;
 
 /// Wait time for SourceActor before pooling for new documents.
@@ -138,7 +138,7 @@ impl IngestApiSource {
 impl Source for IngestApiSource {
     async fn initialize(
         &mut self,
-        _: &ProcessorMailbox,
+        _: &SourceSink,
         ctx: &SourceContext,
     ) -> Result<(), ActorExitStatus> {
         if let Some(position) = self.counters.previous_offset {
@@ -150,7 +150,7 @@ impl Source for IngestApiSource {
 
     async fn emit_batches(
         &mut self,
-        batch_sink: &ProcessorMailbox,
+        batch_sink: &SourceSink,
         ctx: &SourceContext,
     ) -> Result<Duration, ActorExitStatus> {
         let fetch_req = FetchRequest {