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, RwLock, RwLockReadGuard, RwLockWriteGuard},
14};
15use tracing::trace;
16
17#[derive(Clone)]
18pub struct ChannelStatus(Arc<RwLock<Inner>>);
19
20impl ChannelStatus {
21 pub(crate) fn new(id: ChannelId, internal_rpc: InternalRPCHandle) -> Self {
22 Self(Arc::new(RwLock::new(Inner::new(id, internal_rpc))))
23 }
24
25 pub fn initializing(&self) -> bool {
26 [ChannelState::Initial, ChannelState::Reconnecting].contains(&self.read().state)
27 }
28
29 pub fn closing(&self) -> bool {
30 [ChannelState::Closing, ChannelState::Reconnecting].contains(&self.read().state)
31 }
32
33 pub fn connected(&self) -> bool {
34 self.read().state == ChannelState::Connected
35 }
36
37 pub fn reconnecting(&self) -> bool {
38 self.read().state == ChannelState::Reconnecting
39 }
40
41 pub(crate) fn connected_or_recovering(&self) -> bool {
42 [ChannelState::Connected, ChannelState::Reconnecting].contains(&self.read().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.write().recovery_context.as_mut()?))
50 }
51
52 pub(crate) fn finalize_connection(&self) {
53 self.write().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.read().state)
63 }
64
65 pub fn confirm(&self) -> bool {
66 self.read().confirm
67 }
68
69 pub(crate) fn set_confirm(&self) {
70 let mut inner = self.write();
71 inner.confirm = true;
72 trace!("Publisher confirms activated");
73 inner.finalize_connection();
74 }
75
76 pub(crate) fn set_state(&self, state: ChannelState) {
77 self.write().state = state;
78 }
79
80 pub(crate) fn state_error(&self, context: &'static str) -> Error {
81 let inner = self.read();
82 let error = Error::from(ErrorKind::InvalidChannelState(inner.state, context));
83 if inner.state == ChannelState::Reconnecting {
84 return error.with_notifier(inner.notifier());
85 }
86 error
87 }
88
89 pub(crate) fn set_reconnecting(&self, error: Error, topology: ChannelDefinition) -> Error {
90 self.write().set_reconnecting(error, topology)
91 }
92
93 pub(crate) fn auto_close(&self, id: ChannelId) -> bool {
94 id != 0 && self.connected()
95 }
96
97 #[cfg(test)]
98 pub(crate) fn receiver_state(&self) -> crate::channel_receiver_state::ChannelReceiverState {
99 self.write().receiver_state.receiver_state()
100 }
101
102 pub(crate) fn set_will_receive(
103 &self,
104 class_id: Identifier,
105 delivery_cause: DeliveryCause,
106 ) -> KillSwitch {
107 let mut inner = self.write();
108 inner
109 .receiver_state
110 .set_will_receive(class_id, delivery_cause);
111 inner.killswitch.clone()
112 }
113
114 pub(crate) fn set_content_length<
115 Handler: FnOnce(&DeliveryCause, bool),
116 OnInvalidClass: FnOnce(String) -> Result<()>,
117 OnError: FnOnce(String) -> Result<()>,
118 >(
119 &self,
120 channel_id: ChannelId,
121 class_id: Identifier,
122 length: PayloadSize,
123 handler: Handler,
124 invalid_class_hanlder: OnInvalidClass,
125 error_handler: OnError,
126 ) -> Result<()> {
127 let mut inner = self.write();
128 let confirm_mode = inner.confirm;
129 inner.receiver_state.set_content_length(
130 channel_id,
131 class_id,
132 length,
133 handler,
134 invalid_class_hanlder,
135 error_handler,
136 confirm_mode,
137 )
138 }
139
140 pub(crate) fn receive<
141 Handler: FnOnce(&DeliveryCause, PayloadSize, bool),
142 OnError: FnOnce(String) -> Result<()>,
143 >(
144 &self,
145 channel_id: ChannelId,
146 length: PayloadSize,
147 handler: Handler,
148 error_handler: OnError,
149 ) -> Result<()> {
150 let mut inner = self.write();
151 let confirm_mode = inner.confirm;
152 inner
153 .receiver_state
154 .receive(channel_id, length, handler, error_handler, confirm_mode)
155 }
156
157 pub(crate) fn set_send_flow(&self, flow: bool) {
158 self.write().send_flow = flow;
159 }
160
161 pub(crate) fn flow(&self) -> bool {
162 self.read().send_flow
163 }
164
165 fn read(&self) -> RwLockReadGuard<'_, Inner> {
166 self.0.read().unwrap_or_else(|e| e.into_inner())
167 }
168
169 fn write(&self) -> RwLockWriteGuard<'_, Inner> {
170 self.0.write().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_read() {
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 match self.recovery_context.as_ref() {
236 Some(ctx) => ctx.cause(),
237 None => {
238 let ctx = ChannelRecoveryContext::new(error, topology);
239 let error = ctx.cause();
240 self.recovery_context = Some(ctx);
241 error
242 }
243 }
244 }
245
246 pub(crate) fn finalize_connection(&mut self) {
247 self.state = ChannelState::Connected;
248 self.update_rpc_status();
249 if let Some(ctx) = self.recovery_context.take() {
250 ctx.finalize_recovery();
251 }
252 }
253
254 fn notifier(&self) -> Option<Notifier> {
255 Some(self.recovery_context.as_ref()?.notifier())
256 }
257}