1use std::{
2 collections::{HashMap, HashSet},
3 error::Error,
4 fmt::{Display, Formatter},
5 os::unix::fs::FileTypeExt,
6 path::{Path, PathBuf},
7 sync::{
8 Arc, Mutex,
9 atomic::{AtomicU64, Ordering},
10 },
11};
12
13use anyhow::{Context, Result, ensure};
14use async_trait::async_trait;
15use serde::{Deserialize, Serialize};
16use serde_json::Value;
17use tokio::{
18 io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
19 net::{UnixListener, unix::OwnedWriteHalf},
20 sync::{Mutex as AsyncMutex, oneshot},
21 task::JoinHandle,
22};
23use tokio_util::sync::CancellationToken;
24use tracing::{debug, info};
25
26use super::{ActorInvocationFailure, ActorKey};
27
28const ACTOR_EXECUTOR_PROTOCOL_VERSION: u32 = 12;
29pub(crate) const MAX_ACTOR_EXECUTOR_MESSAGE_BYTES: usize = 32 * 1024 * 1024;
30
31#[derive(Debug, Serialize)]
32pub struct ActorMethodInvocation {
33 pub request_id: String,
34 pub actor: ActorKey,
35 pub method: String,
36 pub args: Vec<Value>,
37 pub state: Option<Value>,
38 pub connections: Vec<ActorSocketConnection>,
39}
40
41#[derive(Debug, Serialize)]
42pub struct ActorMethodEviction {
43 pub actor: ActorKey,
44}
45
46#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
47pub struct ActorSocketConnection {
48 pub id: String,
49 pub metadata: Value,
50 pub tags: Vec<String>,
51}
52
53#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
54#[serde(tag = "type", rename_all = "snake_case")]
55pub enum ActorSocketMessage {
56 Text { data: String },
57 Binary { data: String },
58}
59
60#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
61#[serde(tag = "type", rename_all = "snake_case")]
62pub enum ActorSocketEvent {
63 Connect {
64 connection: ActorSocketConnection,
65 },
66 Message {
67 connection_id: String,
68 message: ActorSocketMessage,
69 },
70 Disconnect {
71 connection: ActorSocketConnection,
72 code: u16,
73 reason: String,
74 was_clean: bool,
75 },
76}
77
78#[derive(Debug, Serialize)]
79pub struct ActorSocketInvocation {
80 pub request_id: String,
81 pub actor: ActorKey,
82 pub event: ActorSocketEvent,
83 pub connections: Vec<ActorSocketConnection>,
84 pub state: Option<Value>,
85}
86
87#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
88#[serde(tag = "type", rename_all = "snake_case")]
89pub enum ActorSocketEffect {
90 Send {
91 connection_id: String,
92 message: ActorSocketMessage,
93 },
94 Broadcast {
95 message: ActorSocketMessage,
96 except_connection_ids: Vec<String>,
97 tags: Vec<String>,
98 },
99 Close {
100 connection_id: String,
101 code: u16,
102 reason: String,
103 },
104 Reject {
105 connection_id: String,
106 code: u16,
107 reason: String,
108 },
109 SetMetadata {
110 connection_id: String,
111 metadata: Value,
112 },
113 SetTags {
114 connection_id: String,
115 tags: Vec<String>,
116 },
117}
118
119#[derive(Debug, PartialEq)]
120pub enum ActorMethodOutcome {
121 Completed {
122 result: Value,
123 state: Value,
124 effects: Vec<ActorSocketEffect>,
125 },
126 Failed(ActorInvocationFailure),
127}
128
129#[derive(Debug, PartialEq)]
130pub enum ActorSocketOutcome {
131 Handled {
132 state: Value,
133 effects: Vec<ActorSocketEffect>,
134 },
135 Failed(ActorInvocationFailure),
136}
137
138#[async_trait]
139pub trait ActorExecutor: Send + Sync {
140 fn supports(&self, actor_type: &str) -> bool;
141
142 async fn invoke(&self, invocation: ActorMethodInvocation) -> Result<ActorMethodOutcome>;
143
144 async fn handle_socket(
145 &self,
146 _invocation: ActorSocketInvocation,
147 ) -> Result<ActorSocketOutcome> {
148 Ok(ActorSocketOutcome::Failed(ActorInvocationFailure {
149 code: "socket_not_supported".into(),
150 message: "actor executor does not support sockets".into(),
151 }))
152 }
153
154 async fn evict(&self, _eviction: ActorMethodEviction) -> Result<()> {
155 Ok(())
156 }
157}
158
159pub(crate) struct ActorExecutorListener {
160 listener: UnixListener,
161 socket_path: PathBuf,
162}
163
164impl ActorExecutorListener {
165 pub(crate) async fn bind(socket_path: impl Into<PathBuf>) -> Result<Self> {
166 let socket_path = socket_path.into();
167 prepare_socket_path(&socket_path).await?;
168 if let Some(parent) = socket_path.parent() {
169 tokio::fs::create_dir_all(parent)
170 .await
171 .with_context(|| format!("create actor executor directory {}", parent.display()))?;
172 }
173 let listener = UnixListener::bind(&socket_path)
174 .with_context(|| format!("bind actor executor socket {}", socket_path.display()))?;
175 Ok(Self {
176 listener,
177 socket_path,
178 })
179 }
180
181 pub(crate) async fn accept(self) -> Result<ActorExecutorConnection> {
182 let result = self.accept_connection().await;
183 let cleanup = remove_socket(&self.socket_path).await;
184 match (result, cleanup) {
185 (Ok(connection), Ok(())) => Ok(connection),
186 (Err(error), _) => Err(error),
187 (Ok(_), Err(error)) => Err(error),
188 }
189 }
190
191 async fn accept_connection(&self) -> Result<ActorExecutorConnection> {
192 let (stream, _) =
193 self.listener.accept().await.with_context(|| {
194 format!("accept actor executor at {}", self.socket_path.display())
195 })?;
196 let (reader, writer) = stream.into_split();
197 let mut reader = BufReader::new(reader);
198 let attach = match read_client_message(&mut reader).await? {
199 Some(ActorExecutorClientMessage::Attach {
200 protocol,
201 actor_types,
202 }) => {
203 ensure!(
204 protocol == ACTOR_EXECUTOR_PROTOCOL_VERSION,
205 "customer actor executor uses unsupported protocol version {protocol}"
206 );
207 ensure!(
208 !actor_types.is_empty(),
209 "customer actor executor did not advertise any actor types"
210 );
211 actor_types
212 }
213 Some(_) => {
214 anyhow::bail!("first customer actor executor message must attach the process")
215 }
216 None => anyhow::bail!("customer actor executor disconnected before attaching"),
217 };
218
219 let executor = Arc::new(JsActorExecutor::new(writer, attach));
220 let task = tokio::spawn(read_executor_messages(reader, executor.clone()));
221 debug!(
222 socket = %self.socket_path.display(),
223 actor_types = ?executor.actor_types,
224 "customer JavaScript process connected to actor executor"
225 );
226 Ok(ActorExecutorConnection { executor, task })
227 }
228}
229
230pub(crate) struct ActorExecutorConnection {
231 executor: Arc<JsActorExecutor>,
232 task: JoinHandle<Result<()>>,
233}
234
235impl ActorExecutorConnection {
236 pub(crate) fn executor(&self) -> Arc<dyn ActorExecutor> {
237 self.executor.clone()
238 }
239
240 pub(crate) async fn mark_ready(&self) -> Result<()> {
241 self.executor
242 .send(&ActorExecutorServerMessage::Attached {
243 protocol: ACTOR_EXECUTOR_PROTOCOL_VERSION,
244 })
245 .await?;
246 info!(
247 actor_types = ?self.executor.actor_types,
248 "customer JavaScript process attached to actor executor"
249 );
250 Ok(())
251 }
252
253 pub(crate) async fn run(mut self, shutdown: CancellationToken) -> Result<()> {
254 tokio::select! {
255 result = &mut self.task => {
256 match result {
257 Ok(result) => result,
258 Err(error) => Err(error.into()),
259 }
260 }
261 _ = shutdown.cancelled() => {
262 self.executor.close().await;
263 self.task.abort();
264 let _ = (&mut self.task).await;
265 Ok(())
266 }
267 }
268 }
269}
270
271impl Drop for ActorExecutorConnection {
272 fn drop(&mut self) {
273 self.task.abort();
274 }
275}
276
277struct JsActorExecutor {
278 actor_types: HashSet<String>,
279 next_message_id: AtomicU64,
280 pending: Mutex<HashMap<u64, oneshot::Sender<ExecutorReply>>>,
281 writer: AsyncMutex<OwnedWriteHalf>,
282}
283
284#[async_trait]
285impl ActorExecutor for JsActorExecutor {
286 fn supports(&self, actor_type: &str) -> bool {
287 self.actor_types.contains(actor_type)
288 }
289
290 async fn invoke(&self, invocation: ActorMethodInvocation) -> Result<ActorMethodOutcome> {
291 match self.exchange(ExecutorCommand::Invoke(invocation)).await? {
292 ExecutorReply::Invoked {
293 result,
294 state,
295 effects,
296 } => Ok(ActorMethodOutcome::Completed {
297 result,
298 state,
299 effects,
300 }),
301 ExecutorReply::Failed { code, message } => {
302 Ok(ActorMethodOutcome::Failed(ActorInvocationFailure {
303 code,
304 message,
305 }))
306 }
307 ExecutorReply::Evicted => {
308 anyhow::bail!("actor executor returned eviction reply to invocation")
309 }
310 ExecutorReply::WebsocketHandled { .. } => {
311 anyhow::bail!("actor executor returned socket reply to invocation")
312 }
313 }
314 }
315
316 async fn handle_socket(&self, invocation: ActorSocketInvocation) -> Result<ActorSocketOutcome> {
317 match self
318 .exchange(ExecutorCommand::WebsocketEvent(invocation))
319 .await?
320 {
321 ExecutorReply::WebsocketHandled { state, effects } => {
322 Ok(ActorSocketOutcome::Handled { state, effects })
323 }
324 ExecutorReply::Failed { code, message } => {
325 Ok(ActorSocketOutcome::Failed(ActorInvocationFailure {
326 code,
327 message,
328 }))
329 }
330 ExecutorReply::Invoked { .. } | ExecutorReply::Evicted => {
331 anyhow::bail!("actor executor returned the wrong reply to socket event")
332 }
333 }
334 }
335
336 async fn evict(&self, eviction: ActorMethodEviction) -> Result<()> {
337 match self.exchange(ExecutorCommand::Evict(eviction)).await? {
338 ExecutorReply::Evicted => Ok(()),
339 ExecutorReply::Failed { code, message } => {
340 anyhow::bail!("actor executor rejected eviction ({code}): {message}")
341 }
342 ExecutorReply::Invoked { .. } => {
343 anyhow::bail!("actor executor returned the wrong reply to eviction")
344 }
345 ExecutorReply::WebsocketHandled { .. } => {
346 anyhow::bail!("actor executor returned socket reply to eviction")
347 }
348 }
349 }
350}
351
352impl JsActorExecutor {
353 fn new(writer: OwnedWriteHalf, actor_types: Vec<String>) -> Self {
354 Self {
355 actor_types: actor_types.into_iter().collect(),
356 next_message_id: AtomicU64::new(1),
357 pending: Mutex::new(HashMap::new()),
358 writer: AsyncMutex::new(writer),
359 }
360 }
361
362 async fn exchange(&self, command: ExecutorCommand) -> Result<ExecutorReply> {
363 let message_id = self.next_message_id.fetch_add(1, Ordering::Relaxed);
364 let (reply_tx, reply_rx) = oneshot::channel();
365 self.pending
366 .lock()
367 .map_err(|_| anyhow::anyhow!("actor executor pending-reply lock poisoned"))?
368 .insert(message_id, reply_tx);
369
370 let write_result = self
371 .send(&ActorExecutorServerMessage::Command {
372 message_id,
373 command: Box::new(command),
374 })
375 .await;
376 if let Err(error) = write_result {
377 self.remove_pending(message_id)?;
378 if let Some(limit) = error.downcast_ref::<ActorExecutorMessageTooLarge>() {
379 return Ok(ExecutorReply::Failed {
380 code: "resource_exhausted".into(),
381 message: limit.to_string(),
382 });
383 }
384 return Err(error.context("send command to customer actor executor"));
385 }
386
387 reply_rx
388 .await
389 .context("customer actor executor disconnected before replying")
390 }
391
392 async fn send(&self, message: &ActorExecutorServerMessage) -> Result<()> {
393 write_server_message(&mut *self.writer.lock().await, message).await
394 }
395
396 fn deliver(&self, message_id: u64, reply: ExecutorReply) -> Result<()> {
397 let sender = self
398 .pending
399 .lock()
400 .map_err(|_| anyhow::anyhow!("actor executor pending-reply lock poisoned"))?
401 .remove(&message_id)
402 .with_context(|| format!("actor executor replied to unknown message {message_id}"))?;
403 let _ = sender.send(reply);
404 Ok(())
405 }
406
407 fn remove_pending(&self, message_id: u64) -> Result<()> {
408 self.pending
409 .lock()
410 .map_err(|_| anyhow::anyhow!("actor executor pending-reply lock poisoned"))?
411 .remove(&message_id);
412 Ok(())
413 }
414
415 async fn close(&self) {
416 let _ = self.writer.lock().await.shutdown().await;
417 self.disconnect();
418 }
419
420 fn disconnect(&self) {
421 if let Ok(mut pending) = self.pending.lock() {
422 pending.clear();
423 }
424 }
425}
426
427async fn read_executor_messages(
428 mut reader: BufReader<tokio::net::unix::OwnedReadHalf>,
429 executor: Arc<JsActorExecutor>,
430) -> Result<()> {
431 let result = async {
432 loop {
433 match read_client_message(&mut reader).await? {
434 None => {
435 anyhow::bail!("customer JavaScript actor executor disconnected")
436 }
437 Some(ActorExecutorClientMessage::Reply { message_id, reply }) => {
438 executor.deliver(message_id, reply)?;
439 }
440 Some(ActorExecutorClientMessage::Attach { .. }) => {
441 anyhow::bail!("customer actor executor attached more than once")
442 }
443 }
444 }
445 }
446 .await;
447 executor.disconnect();
448 result
449}
450
451async fn read_client_message(
452 reader: &mut BufReader<tokio::net::unix::OwnedReadHalf>,
453) -> Result<Option<ActorExecutorClientMessage>> {
454 let mut line = String::new();
455 let bytes = reader.read_line(&mut line).await?;
456 if bytes == 0 {
457 return Ok(None);
458 }
459 ensure!(
460 bytes <= MAX_ACTOR_EXECUTOR_MESSAGE_BYTES,
461 "customer actor executor message exceeds {MAX_ACTOR_EXECUTOR_MESSAGE_BYTES} bytes"
462 );
463 serde_json::from_str(line.trim_end())
464 .map(Some)
465 .context("decode customer actor executor message")
466}
467
468async fn write_server_message(
469 writer: &mut OwnedWriteHalf,
470 message: &ActorExecutorServerMessage,
471) -> Result<()> {
472 let mut bytes = serde_json::to_vec(message)?;
473 bytes.push(b'\n');
474 if bytes.len() > MAX_ACTOR_EXECUTOR_MESSAGE_BYTES {
475 return Err(ActorExecutorMessageTooLarge.into());
476 }
477 writer.write_all(&bytes).await?;
478 Ok(())
479}
480
481#[derive(Debug)]
482struct ActorExecutorMessageTooLarge;
483
484impl Display for ActorExecutorMessageTooLarge {
485 fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
486 write!(
487 formatter,
488 "actor executor command exceeds {MAX_ACTOR_EXECUTOR_MESSAGE_BYTES} bytes"
489 )
490 }
491}
492
493impl Error for ActorExecutorMessageTooLarge {}
494
495async fn prepare_socket_path(path: &Path) -> Result<()> {
496 match tokio::fs::symlink_metadata(path).await {
497 Ok(metadata) => {
498 ensure!(
499 metadata.file_type().is_socket(),
500 "refusing to replace non-socket actor executor path {}",
501 path.display()
502 );
503 tokio::fs::remove_file(path).await?;
504 }
505 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
506 Err(error) => return Err(error.into()),
507 }
508 Ok(())
509}
510
511async fn remove_socket(path: &Path) -> Result<()> {
512 match tokio::fs::remove_file(path).await {
513 Ok(()) => {
514 debug!(socket = %path.display(), "actor executor socket removed");
515 Ok(())
516 }
517 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
518 Err(error) => Err(error.into()),
519 }
520}
521
522#[derive(Debug, Serialize)]
523#[serde(tag = "type", rename_all = "snake_case")]
524enum ActorExecutorServerMessage {
525 Attached {
526 protocol: u32,
527 },
528 Command {
529 message_id: u64,
530 command: Box<ExecutorCommand>,
531 },
532}
533
534#[derive(Debug, Deserialize)]
535#[serde(tag = "type", rename_all = "snake_case")]
536enum ActorExecutorClientMessage {
537 Attach {
538 protocol: u32,
539 actor_types: Vec<String>,
540 },
541 Reply {
542 message_id: u64,
543 reply: ExecutorReply,
544 },
545}
546
547#[derive(Debug, Serialize)]
548#[serde(tag = "type", rename_all = "snake_case")]
549enum ExecutorCommand {
550 Invoke(ActorMethodInvocation),
551 WebsocketEvent(ActorSocketInvocation),
552 Evict(ActorMethodEviction),
553}
554
555#[derive(Debug, Deserialize)]
556#[serde(tag = "type", rename_all = "snake_case")]
557enum ExecutorReply {
558 Invoked {
559 result: Value,
560 state: Value,
561 #[serde(default)]
562 effects: Vec<ActorSocketEffect>,
563 },
564 WebsocketHandled {
565 state: Value,
566 effects: Vec<ActorSocketEffect>,
567 },
568 Failed {
569 code: String,
570 message: String,
571 },
572 Evicted,
573}
574
575#[cfg(test)]
576mod tests {
577 use super::*;
578 use serde_json::json;
579 use tempfile::TempDir;
580 use tokio::{
581 io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
582 net::UnixStream,
583 };
584
585 #[tokio::test]
586 async fn one_javascript_executor_runs_until_host_shutdown() -> Result<()> {
587 let root = TempDir::new_in("/tmp")?;
588 let socket = root.path().join("actor-executor.sock");
589 let host = ActorExecutorListener::bind(&socket).await?;
590 let customer = tokio::spawn(run_incrementing_customer(socket.clone()));
591 let connection = host.accept().await?;
592 let executor = connection.executor();
593 connection.mark_ready().await?;
594 assert!(executor.supports("counter"));
595
596 let shutdown = CancellationToken::new();
597 let connection_task = tokio::spawn(connection.run(shutdown.clone()));
598 let outcome = executor
599 .invoke(ActorMethodInvocation {
600 request_id: "request-1".into(),
601 actor: ActorKey {
602 namespace_id: "namespace-1".into(),
603 actor_type: "counter".into(),
604 actor_id: "counter-1".into(),
605 },
606 method: "increment".into(),
607 args: vec![json!(2)],
608 state: None,
609 connections: Vec::new(),
610 })
611 .await?;
612 assert_eq!(
613 outcome,
614 ActorMethodOutcome::Completed {
615 result: json!(2),
616 state: json!({ "count": 2 }),
617 effects: Vec::new(),
618 }
619 );
620 let socket_outcome = executor
621 .handle_socket(ActorSocketInvocation {
622 request_id: "socket-request-1".into(),
623 actor: ActorKey {
624 namespace_id: "namespace-1".into(),
625 actor_type: "counter".into(),
626 actor_id: "counter-1".into(),
627 },
628 event: ActorSocketEvent::Connect {
629 connection: ActorSocketConnection {
630 id: "socket-1".into(),
631 metadata: json!({ "userId": "user-1" }),
632 tags: Vec::new(),
633 },
634 },
635 connections: vec![ActorSocketConnection {
636 id: "socket-1".into(),
637 metadata: json!({ "userId": "user-1" }),
638 tags: Vec::new(),
639 }],
640 state: Some(json!({ "count": 2 })),
641 })
642 .await?;
643 assert_eq!(
644 socket_outcome,
645 ActorSocketOutcome::Handled {
646 state: json!({ "count": 3 }),
647 effects: vec![ActorSocketEffect::Send {
648 connection_id: "socket-1".into(),
649 message: ActorSocketMessage::Text {
650 data: "ready".into()
651 },
652 }],
653 }
654 );
655 shutdown.cancel();
656 connection_task.await??;
657 customer.await??;
658 Ok(())
659 }
660
661 #[tokio::test]
662 async fn oversized_commands_are_reported_as_resource_exhausted() -> Result<()> {
663 let root = TempDir::new_in("/tmp")?;
664 let socket = root.path().join("actor-executor.sock");
665 let host = ActorExecutorListener::bind(&socket).await?;
666 let customer = tokio::spawn(run_attached_customer(socket.clone()));
667 let connection = host.accept().await?;
668 let executor = connection.executor();
669 connection.mark_ready().await?;
670
671 let shutdown = CancellationToken::new();
672 let connection_task = tokio::spawn(connection.run(shutdown.clone()));
673 let outcome = executor
674 .invoke(ActorMethodInvocation {
675 request_id: "request-1".into(),
676 actor: ActorKey {
677 namespace_id: "namespace-1".into(),
678 actor_type: "counter".into(),
679 actor_id: "counter-1".into(),
680 },
681 method: "accept".into(),
682 args: vec![json!("x".repeat(MAX_ACTOR_EXECUTOR_MESSAGE_BYTES))],
683 state: None,
684 connections: Vec::new(),
685 })
686 .await?;
687
688 assert!(matches!(
689 outcome,
690 ActorMethodOutcome::Failed(ref failure) if failure.code == "resource_exhausted"
691 ));
692 shutdown.cancel();
693 connection_task.await??;
694 customer.await??;
695 Ok(())
696 }
697
698 async fn run_incrementing_customer(socket: PathBuf) -> Result<()> {
699 let stream = UnixStream::connect(socket).await?;
700 let (reader, mut writer) = stream.into_split();
701 let mut reader = BufReader::new(reader);
702 writer
703 .write_all(b"{\"type\":\"attach\",\"protocol\":12,\"actor_types\":[\"counter\"]}\n")
704 .await?;
705 ensure!(
706 read_json_line(&mut reader).await? == json!({ "type": "attached", "protocol": 12 })
707 );
708
709 let invocation = read_json_line(&mut reader).await?;
710 let invocation_id = invocation["message_id"]
711 .as_u64()
712 .context("invocation message ID")?;
713 ensure!(invocation["command"]["type"] == "invoke");
714 ensure!(invocation["command"].get("timeout_ms").is_none());
715 write_json_line(
716 &mut writer,
717 &json!({
718 "type": "reply",
719 "message_id": invocation_id,
720 "reply": {
721 "type": "invoked",
722 "result": 2,
723 "state": { "count": 2 }
724 }
725 }),
726 )
727 .await?;
728
729 let socket_event = read_json_line(&mut reader).await?;
730 let socket_event_id = socket_event["message_id"]
731 .as_u64()
732 .context("socket event message ID")?;
733 ensure!(socket_event["command"]["type"] == "websocket_event");
734 ensure!(socket_event["command"]["event"]["type"] == "connect");
735 write_json_line(
736 &mut writer,
737 &json!({
738 "type": "reply",
739 "message_id": socket_event_id,
740 "reply": {
741 "type": "websocket_handled",
742 "state": { "count": 3 },
743 "effects": [{
744 "type": "send",
745 "connection_id": "socket-1",
746 "message": { "type": "text", "data": "ready" }
747 }]
748 }
749 }),
750 )
751 .await?;
752
753 let mut trailing = String::new();
754 ensure!(
755 reader.read_line(&mut trailing).await? == 0,
756 "expected Rust host to close the actor executor"
757 );
758 Ok(())
759 }
760
761 async fn run_attached_customer(socket: PathBuf) -> Result<()> {
762 let stream = UnixStream::connect(socket).await?;
763 let (reader, mut writer) = stream.into_split();
764 let mut reader = BufReader::new(reader);
765 writer
766 .write_all(b"{\"type\":\"attach\",\"protocol\":12,\"actor_types\":[\"counter\"]}\n")
767 .await?;
768 ensure!(
769 read_json_line(&mut reader).await? == json!({ "type": "attached", "protocol": 12 })
770 );
771 let mut trailing = String::new();
772 ensure!(
773 reader.read_line(&mut trailing).await? == 0,
774 "oversized command reached the customer actor executor"
775 );
776 Ok(())
777 }
778
779 async fn read_json_line<R>(reader: &mut R) -> Result<Value>
780 where
781 R: tokio::io::AsyncBufRead + Unpin,
782 {
783 let mut line = String::new();
784 ensure!(reader.read_line(&mut line).await? > 0, "expected JSON line");
785 Ok(serde_json::from_str(line.trim_end())?)
786 }
787
788 async fn write_json_line<W>(writer: &mut W, value: &Value) -> Result<()>
789 where
790 W: tokio::io::AsyncWrite + Unpin,
791 {
792 writer
793 .write_all(serde_json::to_string(value)?.as_bytes())
794 .await?;
795 writer.write_all(b"\n").await?;
796 Ok(())
797 }
798}