docbox-core 0.13.2

Docbox core business logic and functionality
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

//! # Notifications
//!
//! Notifications queue system handling notifications for the app

mod mpsc;
mod noop;
pub mod process;
mod sqs;

use crate::aws::SqsClient;
pub use mpsc::MpscNotificationQueueSender;

use serde::Deserialize;

// Pretty common utility function
pub use sqs::parse_bucket_message;

#[derive(Debug, Clone, Deserialize)]
#[serde(tag = "provider", rename_all = "snake_case")]
pub enum NotificationConfig {
    Sqs { queue_url: String },
    Noop,
    Mpsc,
}

impl NotificationConfig {
    pub fn from_env() -> Self {
        match (
            std::env::var("DOCBOX_MPSC_QUEUE"),
            std::env::var("DOCBOX_SQS_URL"),
        ) {
            (Ok(_), _) => NotificationConfig::Mpsc,
            (_, Ok(queue_url)) => NotificationConfig::Sqs { queue_url },
            _ => NotificationConfig::Noop,
        }
    }
}

pub enum AppNotificationQueue {
    Sqs(sqs::SqsNotificationQueue),
    Noop(noop::NoopNotificationQueue),
    Mpsc(mpsc::MpscNotificationQueue),
}

impl AppNotificationQueue {
    pub fn from_config(sqs_client: SqsClient, config: NotificationConfig) -> Self {
        match config {
            NotificationConfig::Sqs { queue_url } => {
                tracing::debug!(%queue_url, "using SQS notification queue");
                AppNotificationQueue::Sqs(sqs::SqsNotificationQueue::create(sqs_client, queue_url))
            }
            NotificationConfig::Noop => {
                tracing::warn!("queue not specified, falling back to no-op queue");
                AppNotificationQueue::Noop(noop::NoopNotificationQueue)
            }
            NotificationConfig::Mpsc => {
                tracing::debug!("DOCBOX_MPSC_QUEUE is set using local webhook notification queue");
                AppNotificationQueue::Mpsc(mpsc::MpscNotificationQueue::create())
            }
        }
    }

    pub async fn next_message(&mut self) -> Option<NotificationQueueMessage> {
        match self {
            AppNotificationQueue::Sqs(queue) => queue.next_message().await,
            AppNotificationQueue::Noop(queue) => queue.next_message().await,
            AppNotificationQueue::Mpsc(queue) => queue.next_message().await,
        }
    }
}

/// Type of message from the notification queue
pub enum NotificationQueueMessage {
    FileCreated {
        bucket_name: String,
        object_key: String,
    },
}

pub(crate) trait NotificationQueue: Send + Sync + 'static {
    /// Request the next message from the notification queue
    async fn next_message(&mut self) -> Option<NotificationQueueMessage>;
}