1use crate::{
2 Error, ErrorKind, Result,
3 channel_receiver_state::{ChannelReceiverStates, DeliveryCause},
4 channel_recovery_context::ChannelRecoveryContext,
5 internal_rpc::InternalRPCHandle,
6 killswitch::KillSwitch,
7 notifier::Notifier,
8 topology::ChannelDefinition,
9 types::{ChannelId, Identifier, PayloadSize},
10};
11use std::{
12 fmt,
13 sync::{Arc, Mutex, MutexGuard},
14};
15use tracing::trace;
16
17#[derive(Clone)]
18pub struct ChannelStatus(Arc<Mutex<Inner>>);
19
20impl ChannelStatus {
21 pub(crate) fn new(id: ChannelId, internal_rpc: InternalRPCHandle) -> Self {
22 Self(Arc::new(Mutex::new(Inner::new(id, internal_rpc))))
23 }
24
25 pub fn initializing(&self) -> bool {
26 [ChannelState::Initial, ChannelState::Reconnecting].contains(&self.lock_inner().state)
27 }
28
29 pub fn closing(&self) -> bool {
30 [ChannelState::Closing, ChannelState::Reconnecting].contains(&self.lock_inner().state)
31 }
32
33 pub fn connected(&self) -> bool {
34 self.lock_inner().state == ChannelState::Connected
35 }
36
37 pub fn reconnecting(&self) -> bool {
38 self.lock_inner().state == ChannelState::Reconnecting
39 }
40
41 pub(crate) fn connected_or_recovering(&self) -> bool {
42 [ChannelState::Connected, ChannelState::Reconnecting].contains(&self.lock_inner().state)
43 }
44
45 pub(crate) fn update_recovery_context<R, F: Fn(&mut ChannelRecoveryContext) -> R>(
46 &self,
47 apply: F,
48 ) -> Option<R> {
49 Some(apply(self.lock_inner().recovery_context.as_mut()?))
50 }
51
52 pub(crate) fn finalize_connection(&self) {
53 self.lock_inner().finalize_connection();
54 }
55
56 pub(crate) fn can_receive_messages(&self) -> bool {
57 [
58 ChannelState::Closing,
59 ChannelState::Connected,
60 ChannelState::Reconnecting,
61 ]
62 .contains(&self.lock_inner().state)
63 }
64
65 pub fn confirm(&self) -> bool {
66 self.lock_inner().confirm
67 }
68
69 pub(crate) fn set_confirm(&self) {
70 let mut inner = self.lock_inner();
71 inner.confirm = true;
72 trace!("Publisher confirms activated");
73 inner.finalize_connection();
74 }
75
76 pub fn state(&self) -> ChannelState {
77 self.lock_inner().state
78 }
79
80 pub(crate) fn set_state(&self, state: ChannelState) {
81 self.lock_inner().state = state;
82 }
83
84 pub(crate) fn state_error(&self, context: &'static str) -> Error {
85 let inner = self.lock_inner();
86 let error = Error::from(ErrorKind::InvalidChannelState(inner.state, context));
87 if inner.state == ChannelState::Reconnecting {
88 return error.with_notifier(inner.notifier());
89 }
90 error
91 }
92
93 pub(crate) fn set_reconnecting(&self, error: Error, topology: ChannelDefinition) -> Error {
94 self.lock_inner().set_reconnecting(error, topology)
95 }
96
97 pub(crate) fn auto_close(&self, id: ChannelId) -> bool {
98 id != 0 && self.lock_inner().state == ChannelState::Connected
99 }
100
101 #[cfg(test)]
102 pub(crate) fn receiver_state(&self) -> crate::channel_receiver_state::ChannelReceiverState {
103 self.lock_inner().receiver_state.receiver_state()
104 }
105
106 pub(crate) fn set_will_receive(
107 &self,
108 class_id: Identifier,
109 delivery_cause: DeliveryCause,
110 ) -> KillSwitch {
111 let mut inner = self.lock_inner();
112 inner
113 .receiver_state
114 .set_will_receive(class_id, delivery_cause);
115 inner.killswitch.clone()
116 }
117
118 pub(crate) fn set_content_length<
119 Handler: FnOnce(&DeliveryCause, bool),
120 OnInvalidClass: FnOnce(String) -> Result<()>,
121 OnError: FnOnce(String) -> Result<()>,
122 >(
123 &self,
124 channel_id: ChannelId,
125 class_id: Identifier,
126 length: PayloadSize,
127 handler: Handler,
128 invalid_class_hanlder: OnInvalidClass,
129 error_handler: OnError,
130 ) -> Result<()> {
131 let mut inner = self.lock_inner();
132 let confirm_mode = inner.confirm;
133 inner.receiver_state.set_content_length(
134 channel_id,
135 class_id,
136 length,
137 handler,
138 invalid_class_hanlder,
139 error_handler,
140 confirm_mode,
141 )
142 }
143
144 pub(crate) fn receive<
145 Handler: FnOnce(&DeliveryCause, PayloadSize, bool),
146 OnError: FnOnce(String) -> Result<()>,
147 >(
148 &self,
149 channel_id: ChannelId,
150 length: PayloadSize,
151 handler: Handler,
152 error_handler: OnError,
153 ) -> Result<()> {
154 let mut inner = self.lock_inner();
155 let confirm_mode = inner.confirm;
156 inner
157 .receiver_state
158 .receive(channel_id, length, handler, error_handler, confirm_mode)
159 }
160
161 pub(crate) fn set_send_flow(&self, flow: bool) {
162 self.lock_inner().send_flow = flow;
163 }
164
165 pub(crate) fn flow(&self) -> bool {
166 self.lock_inner().send_flow
167 }
168
169 fn lock_inner(&self) -> MutexGuard<'_, Inner> {
170 self.0.lock().unwrap_or_else(|e| e.into_inner())
171 }
172}
173
174#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
175pub enum ChannelState {
176 #[default]
177 Initial,
178 Reconnecting,
179 Connected,
180 Closing,
181 Closed,
182 Error,
183}
184
185impl fmt::Debug for ChannelStatus {
186 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
187 let mut debug = f.debug_struct("ChannelStatus");
188 if let Ok(inner) = self.0.try_lock() {
189 debug
190 .field("state", &inner.state)
191 .field("receiver_state", &inner.receiver_state)
192 .field("confirm", &inner.confirm)
193 .field("send_flow", &inner.send_flow);
194 }
195 debug.finish()
196 }
197}
198
199struct Inner {
200 id: ChannelId,
201 confirm: bool,
202 send_flow: bool,
203 state: ChannelState,
204 receiver_state: ChannelReceiverStates,
205 recovery_context: Option<ChannelRecoveryContext>,
206 killswitch: KillSwitch,
207 internal_rpc: InternalRPCHandle,
208}
209
210impl Inner {
211 fn new(id: ChannelId, internal_rpc: InternalRPCHandle) -> Self {
212 let this = Self {
213 id,
214 confirm: false,
215 send_flow: true,
216 state: ChannelState::default(),
217 receiver_state: ChannelReceiverStates::default(),
218 recovery_context: None,
219 killswitch: KillSwitch::default(),
220 internal_rpc,
221 };
222 this.update_rpc_status();
223 this
224 }
225
226 fn update_rpc_status(&self) {
227 self.internal_rpc
228 .set_channel_status(self.id, self.killswitch.clone());
229 }
230
231 fn set_reconnecting(&mut self, error: Error, topology: ChannelDefinition) -> Error {
232 self.state = ChannelState::Reconnecting;
233 std::mem::take(&mut self.killswitch).kill();
234 self.receiver_state.reset();
235 let ctx = ChannelRecoveryContext::new(error, topology);
236 let error = ctx.cause();
237 self.recovery_context = Some(ctx);
238 error
239 }
240
241 pub(crate) fn finalize_connection(&mut self) {
242 self.state = ChannelState::Connected;
243 self.update_rpc_status();
244 if let Some(ctx) = self.recovery_context.take() {
245 ctx.finalize_recovery();
246 }
247 }
248
249 fn notifier(&self) -> Option<Notifier> {
250 Some(self.recovery_context.as_ref()?.notifier())
251 }
252}