infrarust 1.6.1

A Rust universal Minecraft proxy
Documentation 
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
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137

use infrarust_config::{
    LogType,
    provider::{Provider, ProviderMessage},
};
use std::sync::Arc;
use tokio::sync::mpsc::{Receiver, Sender};
use tracing::{Instrument, Span, debug, debug_span, error, info, instrument, warn};

use super::service::ConfigurationService;

pub struct ConfigProvider {
    config_service: Arc<ConfigurationService>,
    provider_receiver: Receiver<ProviderMessage>,
    provider_sender: Sender<ProviderMessage>,
}

impl ConfigProvider {
    #[instrument(
        skip(config_service, provider_receiver, provider_sender),
        name = "create_config_provider"
    )]
    pub fn new(
        config_service: Arc<ConfigurationService>,
        provider_receiver: Receiver<ProviderMessage>,
        provider_sender: Sender<ProviderMessage>,
    ) -> Self {
        debug!(
            log_type = LogType::ConfigProvider.as_str(),
            "Creating new configuration provider"
        );
        Self {
            config_service,
            provider_receiver,
            provider_sender,
        }
    }

    #[instrument(skip(self))]
    pub async fn run(&mut self) {
        let span = debug_span!(
            "config_provider_run",
            log_type = LogType::ConfigProvider.as_str()
        );
        async {
            info!(
                log_type = LogType::ConfigProvider.as_str(),
                "Starting configuration provider"
            );
            while let Some(message) = self.provider_receiver.recv().await {
                self.handle_message(message).await;
            }
        }
        .instrument(span)
        .await
    }

    async fn handle_message(&mut self, message: ProviderMessage) {
        match message {
            ProviderMessage::Update {
                key,
                configuration,
                span: _span,
            } => {
                // Set span parent to the span that was passed in
                let new_span = debug_span!(
                    "config_provider: config_update",
                    key = %key,
                    has_config = configuration.is_some(),
                    log_type = LogType::ConfigProvider.as_str()
                );

                #[cfg(feature = "telemetry")]
                new_span.set_parent(_span.context());

                async {
                    debug!(
                        log_type = LogType::ConfigProvider.as_str(),
                        "Processing configuration update for: {}", key
                    );

                    if let Some(config) = configuration {
                        self.config_service
                            .update_configurations(vec![*config])
                            .instrument(debug_span!(
                                "config_provider: apply_config_update",
                                log_type = LogType::ConfigProvider.as_str()
                            ))
                            .await;
                    } else {
                        self.config_service.remove_configuration(&key).await;
                    }
                }
                .instrument(new_span)
                .await;
            }
            ProviderMessage::FirstInit(configs) => {
                debug!(
                    log_type = LogType::ConfigProvider.as_str(),
                    config_count = configs.len(),
                    "First initialization received"
                );
                let config_vec = configs.into_values().collect();
                self.config_service.update_configurations(config_vec).await;
            }
            ProviderMessage::Error(err) => {
                error!(log_type = LogType::ConfigProvider.as_str(), error = %err, "Provider error received");
            }
            ProviderMessage::Shutdown => {
                info!(
                    log_type = LogType::ConfigProvider.as_str(),
                    "Shutdown message received"
                );
            }
        }
    }

    #[instrument(skip(self, provider), fields(provider_name = %provider.get_name(), name = "register_provider"))]
    pub fn register_provider(&mut self, mut provider: Box<dyn Provider>) {
        let current_span = Span::current();
        let sender = self.provider_sender.clone();
        tokio::spawn(
            async move {
                provider.run().instrument(debug_span!("provider_run", log_type = LogType::ConfigProvider.as_str())).await;
                warn!(log_type = LogType::ConfigProvider.as_str(), "Provider stopped unexpectedly");
                if let Err(e) = sender
                    .send(ProviderMessage::Error(
                        "Unexpected provider termination".into(),
                    ))
                    .await
                {
                    error!(log_type = LogType::ConfigProvider.as_str(), error = %e, "Failed to send provider error");
                }
            }
            .instrument(current_span),
        );
    }
}