pubnub 0.7.0

PubNub SDK for Rust
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

use async_channel::Sender;
use spin::rwlock::RwLock;
use uuid::Uuid;

use crate::{
    core::event_engine::EffectHandler,
    dx::subscribe::event_engine::{
        effects::{EmitMessagesEffectExecutor, EmitStatusEffectExecutor, SubscribeEffectExecutor},
        SubscribeEffect, SubscribeEffectInvocation,
    },
    lib::{
        alloc::{string::String, sync::Arc},
        core::fmt::{Debug, Formatter, Result},
    },
};

/// Subscription effect handler.
///
/// Handler responsible for effects implementation and creation in response on
/// effect invocation.
pub(crate) struct SubscribeEffectHandler {
    /// Subscribe call function pointer.
    subscribe_call: Arc<SubscribeEffectExecutor>,

    /// Emit status function pointer.
    emit_status: Arc<EmitStatusEffectExecutor>,

    /// Emit messages function pointer.
    emit_messages: Arc<EmitMessagesEffectExecutor>,

    /// Cancellation channel.
    cancellation_channel: Sender<String>,
}

impl SubscribeEffectHandler {
    /// Create subscribe effect handler.
    pub fn new(
        subscribe_call: Arc<SubscribeEffectExecutor>,
        emit_status: Arc<EmitStatusEffectExecutor>,
        emit_messages: Arc<EmitMessagesEffectExecutor>,
        cancellation_channel: Sender<String>,
    ) -> Self {
        Self {
            subscribe_call,
            emit_status,
            emit_messages,
            cancellation_channel,
        }
    }
}

impl EffectHandler<SubscribeEffectInvocation, SubscribeEffect> for SubscribeEffectHandler {
    fn create(&self, invocation: &SubscribeEffectInvocation) -> Option<SubscribeEffect> {
        match invocation {
            SubscribeEffectInvocation::Handshake { input, cursor } => {
                Some(SubscribeEffect::Handshake {
                    id: Uuid::new_v4().to_string(),
                    cancelled: RwLock::new(false),
                    input: input.clone(),
                    cursor: cursor.clone(),
                    executor: self.subscribe_call.clone(),
                    cancellation_channel: self.cancellation_channel.clone(),
                })
            }
            SubscribeEffectInvocation::Receive { input, cursor } => {
                Some(SubscribeEffect::Receive {
                    id: Uuid::new_v4().to_string(),
                    cancelled: RwLock::new(false),
                    input: input.clone(),
                    cursor: cursor.clone(),
                    executor: self.subscribe_call.clone(),
                    cancellation_channel: self.cancellation_channel.clone(),
                })
            }
            SubscribeEffectInvocation::EmitStatus(status) => Some(SubscribeEffect::EmitStatus {
                id: Uuid::new_v4().to_string(),
                status: status.clone(),
                executor: self.emit_status.clone(),
            }),
            SubscribeEffectInvocation::EmitMessages(messages, cursor) => {
                Some(SubscribeEffect::EmitMessages {
                    id: Uuid::new_v4().to_string(),
                    next_cursor: cursor.clone(),
                    updates: messages.clone(),
                    executor: self.emit_messages.clone(),
                })
            }
            _ => None,
        }
    }
}

impl Debug for SubscribeEffectHandler {
    fn fmt(&self, f: &mut Formatter<'_>) -> Result {
        write!(f, "SubscribeEffectHandler {{}}")
    }
}