1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
use log::{info};
use crate::fs_event_emitter::FsEventEmitter;
use crate::fs_completion_handler::{FsCompletionHandler, FsCompletionHandlerActor};
use crate::fs_notify_consumer::{FsNotifyConsumerHandler, FsNotifyConsumerHandlerActor};
use crate::event_processor::{EventProcessorActor, EventProcessor};
use crate::fs_event_retriever::FsRetriever;
use crate::event_handler::{EventHandler, OutputEvent};
use crate::completion_event_serializer::CompletionEventSerializer;
use crate::event_decoder::PayloadDecoder;
use crate::event_handler;
use std::fmt::Debug;
pub async fn local_service<
Err,
EventEncoderT,
CompletedEventT,
EventDecoderT,
EventT,
EventHandlerT,
>(
input_directory: impl AsRef<str>,
output_directory: impl AsRef<str>,
event_encoder: EventEncoderT,
event_decoder: EventDecoderT,
event_handler: EventHandlerT,
) -> Result<(), Box<dyn std::error::Error>>
where
Err: Debug + Clone + Send + Sync + 'static,
CompletedEventT: Clone + Send + Sync + 'static,
EventT: Clone + Send + Sync + 'static,
EventEncoderT: CompletionEventSerializer<CompletedEvent=CompletedEventT, Output=Vec<u8>, Error=crate::error::Error<Err>> + Clone + Send + Sync + 'static,
EventDecoderT: PayloadDecoder<EventT> + Clone + Send + Sync + 'static,
EventHandlerT: EventHandler<InputEvent=EventT, OutputEvent=CompletedEventT, Error=crate::error::Error<Err>> + Clone + Send + Sync + 'static
{
let input_directory = input_directory.as_ref();
let output_directory = output_directory.as_ref();
let fs_event_emitter = FsEventEmitter::new(output_directory);
let (fs_completion_handler, fs_completion_handle) = FsCompletionHandlerActor::new(
FsCompletionHandler::new(
event_encoder,
fs_event_emitter,
)
).await;
let (fs_notify_consumer, fs_notify_handle) = FsNotifyConsumerHandlerActor::new(
FsNotifyConsumerHandler::new(
input_directory,
fs_completion_handler.clone(),
)
).await;
let event_processors: Vec<_> = (0..40)
.into_iter()
.map(|_| {
EventProcessorActor::new(EventProcessor::new(
fs_notify_consumer.clone(),
fs_completion_handler.clone(),
event_handler.clone(),
FsRetriever::new(event_decoder.clone()),
))
})
.collect();
futures::future::join_all(event_processors.iter().map(|ep| ep.0.start_processing())).await;
info!("started processors");
futures::future::join_all(event_processors.into_iter().map(|ep| ep.1)).await;
fs_notify_handle.await;
fs_completion_handle.await;
info!("Complete");
Ok(())
}