reactive-messaging 0.4.1

Reactive client/server communications, focused on high performance
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
138
139
140
141
142
143
144
145
146
147
148
149

//! Models what the server API should be like
//! so to allow us full control of the [Uni] and [Channel]s being used.

use std::marker::PhantomData;

use crate::uni::{GenericUni, MessagingMutinyStream, channel::{ChannelZeroCopy, ChannelMove}, Uni};

// All possible Channels & Uni variations for our server
////////////////////////////////////////////////////////

/// A zero-cost abstraction type allowing to use `reactive-mutiny` Zero-Copy channels
/// (incoming & outgoing messages are guaranteed to be zero-copied)
pub type ZeroCopySocketServer<const CONFIG:                    usize,
                              const PROCESSOR_UNI_INSTRUMENTS: usize,
                              const PROCESSOR_BUFFER_SIZE:     usize,
                              const SENDER_BUFFER_SIZE:        usize,
                              RemoteMessages,
                              LocalMessages /* BUFFER SIZE, etc */>
    = SocketServer<CONFIG,
                   PROCESSOR_UNI_INSTRUMENTS,
                   PROCESSOR_BUFFER_SIZE,
                   RemoteMessages,
                   LocalMessages,
                   Uni<PROCESSOR_UNI_INSTRUMENTS,
                       PROCESSOR_BUFFER_SIZE,
                       RemoteMessages,
                       ChannelZeroCopy<PROCESSOR_BUFFER_SIZE, RemoteMessages>>,
                   ChannelZeroCopy<SENDER_BUFFER_SIZE, LocalMessages>>;

/// A zero-cost abstraction type ready to use `reactive-mutiny' Movable channels
/// (incoming & outgoing messages might be copied, if the compiler can't optimize those operations)
pub type MovableSocketServer<const CONFIG:                    usize,
                             const PROCESSOR_UNI_INSTRUMENTS: usize,
                             const PROCESSOR_BUFFER_SIZE:     usize,
                             const SENDER_BUFFER_SIZE:        usize,
                             RemoteMessages,
                             LocalMessages /* BUFFER SIZE, etc */>
    = SocketServer<CONFIG,
                   PROCESSOR_UNI_INSTRUMENTS,
                   PROCESSOR_BUFFER_SIZE,
                   RemoteMessages,
                   LocalMessages,
                   Uni<PROCESSOR_UNI_INSTRUMENTS,
                       PROCESSOR_BUFFER_SIZE,
                       RemoteMessages,
                       ChannelMove<PROCESSOR_BUFFER_SIZE, RemoteMessages>>,
                   ChannelMove<SENDER_BUFFER_SIZE, LocalMessages>>;

pub struct SocketServer<const CONFIG: usize,
                        const PROCESSOR_UNI_INSTRUMENTS: usize,
                        const PROCESSOR_BUFFER_SIZE: usize,
                        RemoteMessages,
                        LocalMessages,
                        ProcessorUniType: GenericUni<PROCESSOR_UNI_INSTRUMENTS, PROCESSOR_BUFFER_SIZE>,
                        SenderChannelType> {
    pub _phantom: PhantomData<(RemoteMessages,LocalMessages,ProcessorUniType,SenderChannelType)>
}
impl<const CONFIG: usize,
     const PROCESSOR_UNI_INSTRUMENTS: usize,
     const PROCESSOR_BUFFER_SIZE: usize,
     RemoteMessages,
     LocalMessages,
     ProcessorUniType: GenericUni<PROCESSOR_UNI_INSTRUMENTS, PROCESSOR_BUFFER_SIZE>,
     SenderChannelType>
SocketServer<CONFIG, PROCESSOR_UNI_INSTRUMENTS, PROCESSOR_BUFFER_SIZE, RemoteMessages, LocalMessages, ProcessorUniType, SenderChannelType> {

    /// Starts the server, returning an `Arc<Self>` so it may still be shutdown
    pub fn start<LocalMessagesIteratorType:  Iterator<Item=LocalMessages>>
                (self,
                 _connection_events_handler: impl FnMut(ConnectionEvents<SenderChannelType>),
                 _processor:                 impl FnOnce(MessagingMutinyStream<PROCESSOR_UNI_INSTRUMENTS,
                                                                               PROCESSOR_BUFFER_SIZE,
                                                                               ProcessorUniType>)
                                                        -> LocalMessagesIteratorType)
                -> impl SocketServerController {
        self
    }

}

pub trait SocketServerController {
    fn _shutdown(self);
}
impl<const CONFIG: usize,
     const PROCESSOR_UNI_INSTRUMENTS: usize,
     const PROCESSOR_BUFFER_SIZE: usize,
     RemoteMessages,
     LocalMessages,
     ProcessorUniType: GenericUni<PROCESSOR_UNI_INSTRUMENTS, PROCESSOR_BUFFER_SIZE>,
     SenderChannelType>
SocketServerController for
SocketServer<CONFIG, PROCESSOR_UNI_INSTRUMENTS, PROCESSOR_BUFFER_SIZE, RemoteMessages, LocalMessages, ProcessorUniType, SenderChannelType> {
    fn _shutdown(self) {}
}


/// Helps to infer some types:
/// ```nocompile
///     type THE_TYPE_I_WANT = <SocketServer<...> as GenericSocketServer>::THE_TYPE_YOU_WANT
pub trait GenericSocketServer<const PROCESSOR_UNI_INSTRUMENTS: usize,
                              const PROCESSOR_BUFFER_SIZE: usize> {
    const CONFIG: usize;
    type RemoteMessages;
    type LocalMessages;
    type ProcessorUniType: GenericUni<PROCESSOR_UNI_INSTRUMENTS, PROCESSOR_BUFFER_SIZE>;
    type SenderChannelType;
    type ConnectionEventType;
    type StreamItemType;
    type StreamType;
}
impl<const CONFIG: usize,
     const PROCESSOR_UNI_INSTRUMENTS: usize,
     const PROCESSOR_BUFFER_SIZE: usize,
     RemoteMessages,
     LocalMessages,
     ProcessorUniType: GenericUni<PROCESSOR_UNI_INSTRUMENTS, PROCESSOR_BUFFER_SIZE>,
     SenderChannelType>
GenericSocketServer<PROCESSOR_UNI_INSTRUMENTS,
                    PROCESSOR_BUFFER_SIZE> for
SocketServer<CONFIG,
             PROCESSOR_UNI_INSTRUMENTS,
             PROCESSOR_BUFFER_SIZE,
             RemoteMessages,
             LocalMessages,
             ProcessorUniType,
             SenderChannelType> {

    const CONFIG: usize                    = CONFIG;
    type RemoteMessages                    = RemoteMessages;
    type LocalMessages                     = LocalMessages;
    type ProcessorUniType                  = ProcessorUniType;
    type SenderChannelType                 = SenderChannelType;
    type ConnectionEventType               = SenderChannelType;
    type StreamItemType                    = ProcessorUniType::DerivedItemType;
    type StreamType                        = MessagingMutinyStream<PROCESSOR_UNI_INSTRUMENTS,
                                                                   PROCESSOR_BUFFER_SIZE,
                                                                   ProcessorUniType>;
}


/// The possible connection events
pub enum ConnectionEvents<SenderChannelType> {
    /// Happens when the client connects
    _Connected(SenderChannelType),
    /// Happens when the client disconnects
    _Disconnected(SenderChannelType),
    /// Happens when there was an error when reading/writing data
    _IOError(SenderChannelType),
}