rzmq 0.5.7

A high performance, fully asynchronous, safe pure-Rust implementation of ZeroMQ (ØMQ) messaging with io_uring and tcp cork acceleration on Linux.
Documentation 
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

use tracing::debug;

use crate::runtime::{ActorType, SystemEvent}; // Adjust imports as needed
use crate::{Context, ZmqError};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;

pub(crate) struct ActorDropGuard {
  context: Context,
  handle_id: usize,
  actor_type: ActorType,
  parent_id: Option<usize>, 
  endpoint_uri: Option<String>,
  // Use an Arc<AtomicBool> to signal normal exit, preventing double publish
  stopped_normally: bool,
  error: Option<ZmqError>,
}

impl ActorDropGuard {
  pub fn new(context: Context, handle_id: usize, actor_type: ActorType, endpoint_uri: Option<String>, parent_handle_id: Option<usize>) -> Self {

    context.publish_actor_started(handle_id, actor_type, parent_handle_id);
    Self {
      context,
      handle_id,
      parent_id: parent_handle_id,
      actor_type,
      endpoint_uri,
      stopped_normally: false,
      error: None,
    }
  }

  pub fn set_error(&mut self, error: ZmqError) {
    self.error = Some(error);
  }

  pub fn waive(&mut self) {
    self.stopped_normally = true;
  }
}

impl Drop for ActorDropGuard {
  fn drop(&mut self) {
    // Only publish stop if the task didn't signal normal completion
    let error;
    if !self.stopped_normally {
      error = self.error.take().or_else(|| Some(ZmqError::Internal("Actor task cancelled/aborted".into())));

      // Log carefully here, as we are in drop context
      debug!(
        // Use eprintln or tracing::error! if subscriber setup handles panics
        "ActorDropGuard: Actor {} ({:?}) stopping abnormally (error: {}). Publishing stop.",
        self.handle_id, self.actor_type, error.as_ref().unwrap(),
      );
    } else {
      error = None;
    }

    // NOTE: publish_actor_stopping involves sending on a broadcast channel.
    // This is *usually* safe from drop if the receiver task is still running,
    // but it's not ideal. A truly robust solution might involve
    // a synchronous mechanism for the final WaitGroup decrement if possible,
    // but the event system is the current design.
    self.context.publish_actor_stopping(
      self.handle_id,
      self.actor_type,
      self.parent_id,
      self.endpoint_uri.take(),
      error,
    );
  }
}