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
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
use crate::config::{ClientConfig, ClientOpts};
use crate::shell::SubProcess;
use anyhow::{bail, Context, Result};
use async_trait::async_trait;
use cloudpub_common::fair_channel::FairSender;
use cloudpub_common::protocol::message::Message;
use cloudpub_common::protocol::{Break, ErrorInfo, ErrorKind, ServerEndpoint};
use cloudpub_common::utils::is_tcp_port_available;
use parking_lot::RwLock;
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tokio::time::{self, Duration, Instant};
use tracing::{debug, error};
/// ClientEndpoint intentionally has no PartialEq (see common/build.rs);
/// its deterministic prost encoding stands in for config equality.
fn encode_client_config(endpoint: &ServerEndpoint) -> Vec<u8> {
use cloudpub_common::prost::Message as ProstMessage;
endpoint
.client
.as_ref()
.map(|c| c.encode_to_vec())
.unwrap_or_default()
}
#[async_trait]
pub trait Plugin: Send + Sync {
/// Name of the plugin
fn name(&self) -> &'static str;
/// Setup the plugin environment
async fn setup(
&self,
config: &Arc<RwLock<ClientConfig>>,
opts: &ClientOpts,
command_rx: &mut mpsc::Receiver<Message>,
result_tx: &mpsc::Sender<Message>,
) -> Result<()>;
/// Publish a service using this plugin
async fn publish(
&self,
endpoint: &ServerEndpoint,
config: &Arc<RwLock<ClientConfig>>,
opts: &ClientOpts,
result_tx: &mpsc::Sender<Message>,
) -> Result<SubProcess>;
}
pub struct PluginHandle {
guid: String,
/// prost-encoded ClientEndpoint this handle was spawned for, used to
/// recognize duplicate EndpointStart deliveries for the same config
client_config: Vec<u8>,
server: Arc<RwLock<Option<SubProcess>>>,
cancel_tx: mpsc::Sender<Message>,
task: JoinHandle<()>,
}
impl PluginHandle {
pub fn spawn(
plugin: Arc<dyn Plugin>,
mut endpoint: ServerEndpoint,
config: Arc<RwLock<ClientConfig>>,
opts: ClientOpts,
to_server_tx: FairSender<Message>,
) -> Self {
let guid = endpoint.guid.clone();
let client_config = encode_client_config(&endpoint);
let guid_for_task = guid.clone();
let server = Arc::new(RwLock::new(None));
let server_task = server.clone();
// Per-process cancel and event channels
let (cancel_tx, mut cancel_rx) = mpsc::channel::<Message>(1);
let (proc_event_tx, mut proc_event_rx) = mpsc::channel::<Message>(1024);
let task = tokio::spawn(async move {
// Forward per-process events to the main client event channel
tokio::spawn({
let mut endpoint = endpoint.clone();
let to_server_tx = to_server_tx.clone();
let guid_for_task = guid_for_task.clone();
async move {
use tokio::time::{Duration, Instant};
let mut last_progress_time = Instant::now() - Duration::from_secs(1); // Allow first progress immediately
while let Some(mut msg) = proc_event_rx.recv().await {
let should_send = match &mut msg {
Message::Progress(progress_info) => {
progress_info.guid = guid_for_task.clone();
let now = Instant::now();
// Always send 0% and 100% progress messages, or throttle to 1 per second
if progress_info.current == 0
|| progress_info.current >= progress_info.total
|| now.duration_since(last_progress_time)
>= Duration::from_secs(1)
{
last_progress_time = now;
true
} else {
false
}
}
_ => true, // Always send non-progress messages
};
if should_send {
to_server_tx.send(msg.clone()).await.ok();
}
}
endpoint.status = Some("offline".to_string());
to_server_tx
.send(Message::EndpointStatus(endpoint.clone()))
.await
.ok();
}
});
// Initial lifecycle status: waiting
endpoint.status = Some("waiting".to_string());
let _ = to_server_tx
.send(Message::EndpointStatus(endpoint.clone()))
.await;
let res: Result<()> = async {
// Long running setup: cancellable via cancel_rx
plugin
.setup(&config, &opts, &mut cancel_rx, &proc_event_tx)
.await
.context("Failed to setup plugin")?;
let server_process = plugin
.publish(&endpoint, &config, &opts, &proc_event_tx)
.await
.context("Failed to publish plugin service")?;
// Wait until the port is bound
let now = Instant::now();
while is_tcp_port_available("127.0.0.1", server_process.port)
.await
.context("Check port availability")?
{
if server_process.result.read().is_err() {
return Ok(()); // Error already reported by subprocess
}
if now.elapsed() > Duration::from_secs(60) {
bail!("{}", crate::t!("error-start-server"));
}
debug!(
"Waiting for server to start on port {}",
server_process.port
);
time::sleep(Duration::from_secs(1)).await;
}
*server_task.write() = Some(server_process);
Ok(())
}
.await;
match res {
Ok(()) => {
endpoint.status = Some("online".into());
let _ = to_server_tx.send(Message::EndpointStatus(endpoint)).await;
}
Err(e) => {
error!("Error handling endpoint {}: {:#}", &guid_for_task, e);
let _ = to_server_tx
.send(Message::Error(ErrorInfo {
kind: ErrorKind::PublishFailed.into(),
message: e.to_string(),
guid: guid_for_task,
}))
.await;
}
}
});
Self {
guid,
client_config,
server,
cancel_tx,
task,
}
}
/// True when this handle already serves exactly this endpoint
/// configuration: a duplicate EndpointStart (the server re-sends them
/// after every EndpointStartAll and on repeated registrations) must
/// not bounce a running publication.
pub fn same_config(&self, endpoint: &ServerEndpoint) -> bool {
self.client_config == encode_client_config(endpoint)
}
/// The publication is being set up or its subprocess is alive. A
/// finished task with no healthy subprocess means setup failed or the
/// server died - a fresh spawn is warranted then.
pub fn is_healthy(&self) -> bool {
if !self.task.is_finished() {
return true;
}
match self.server.read().as_ref() {
Some(process) => process.result.read().is_ok(),
None => false,
}
}
pub fn port(&self) -> Option<u16> {
self.server.read().as_ref().map(|s| s.port)
}
pub fn guid(&self) -> &str {
&self.guid
}
pub fn send_break(&self) {
let _ = self.cancel_tx.try_send(Message::Break(Break {
guid: self.guid.clone(),
}));
}
}
impl Drop for PluginHandle {
fn drop(&mut self) {
// Abort main task and drop cancel_tx to notify setup to stop
self.task.abort();
// cancel_tx dropped here; receiver sees end-of-stream and should terminate
let _ = self.cancel_tx.try_send(Message::Break(Break {
guid: self.guid.clone(),
}));
}
}