Implement the ZeroMQ transport by SamuelSarle · Pull Request #27 · artob/protoflow

This repository was archived by the owner on Jan 22, 2026. It is now read-only.

-Original file line number
+Diff line change
@@ Expand Up / @@ -17,16 +17,25 @@ publish.workspace = true @@
     [features]
     default = ["all", "std"]
     all = ["tracing"]
-    std = ["protoflow-core/std", "tracing?/std"] #, "zeromq/default"]
+    std = ["protoflow-core/std", "tracing?/std"]
     tracing = ["protoflow-core/tracing", "dep:tracing"]
     unstable = ["protoflow-core/unstable"]
     [build-dependencies]
     cfg_aliases.workspace = true
+    prost-build = "0.13.2"
     [dependencies]
     protoflow-core.workspace = true
     tracing = { version = "0.1", default-features = false, optional = true }
-    #zeromq = { version = "0.4", default-features = false }
+    zeromq = { version = "0.4.1", default-features = false, features = [
+    	"tokio-runtime",
+    	"all-transport",
+    ] }
+    tokio = { version = "1.40.0", default-features = false }
+    prost = "0.13.2"
+    prost-types = "0.13.2"
     [dev-dependencies]
+    futures-util = "0.3.31"
+    tracing-subscriber = "0.3.19"

lib/protoflow-zeromq/build.rs

-Original file line number
+Diff line change
@@ -0,0 +1,6 @@
+    use std::io::Result;
+    fn main() -> Result<()> {
+        prost_build::Config::default()
+            .out_dir("src/")
+            .compile_protos(&["proto/transport_event.proto"], &["proto/"])
+    }

lib/protoflow-zeromq/proto/transport_event.proto

-Original file line number
+Diff line change
@@ -0,0 +1,46 @@
+    syntax = "proto3";
+    package protoflow.zmq;
+    message Connect {
+    	int64 output = 1;
+    	int64 input = 2;
+    }
+    message AckConnection {
+    	int64 output = 1;
+    	int64 input = 2;
+    }
+    message Message {
+    	int64 output = 1;
+    	int64 input = 2;
+    	uint64 sequence = 3;
+    	bytes message = 4;
+    }
+    message AckMessage {
+    	int64 output = 1;
+    	int64 input = 2;
+    	uint64 sequence = 3;
+    }
+    message CloseOutput {
+    	int64 output = 1;
+    	int64 input = 2;
+    }
+    message CloseInput {
+    	int64 input = 1;
+    }
+    message Event {
+    	oneof payload {
+    		Connect connect = 1;
+    		AckConnection ack_connection = 2;
+    		Message message = 3;
+    		AckMessage ack_message = 4;
+    		CloseOutput close_output = 5;
+    		CloseInput close_input = 6;
+    	}
+    }

lib/protoflow-zeromq/src/event.rs

-Original file line number
+Diff line change
@@ -0,0 +1,160 @@
+    // This is free and unencumbered software released into the public domain.
+    use protoflow_core::{
+        prelude::{Bytes, Vec},
+        InputPortID, OutputPortID,
+    };
+    use zeromq::ZmqMessage;
+    pub type SequenceID = u64;
+    /// ZmqTransportEvent represents the data that goes over the wire from one port to another.
+    #[derive(Clone, Debug, PartialEq)]
+    pub enum ZmqTransportEvent {
+        Connect(OutputPortID, InputPortID),
+        AckConnection(OutputPortID, InputPortID),
+        Message(OutputPortID, InputPortID, SequenceID, Bytes),
+        AckMessage(OutputPortID, InputPortID, SequenceID),
+        CloseOutput(OutputPortID, InputPortID),
+        CloseInput(InputPortID),
+    }
+    impl ZmqTransportEvent {
+        fn write_topic<W: std::io::Write + ?Sized>(&self, f: &mut W) -> Result<(), std::io::Error> {
+            use ZmqTransportEvent::*;
+            match self {
+                Connect(o, i) => write!(f, "{}:conn:{}", i, o),
+                AckConnection(o, i) => write!(f, "{}:ackConn:{}", i, o),
+                Message(o, i, seq, _) => write!(f, "{}:msg:{}:{}", i, o, seq),
+                AckMessage(o, i, seq) => write!(f, "{}:ackMsg:{}:{}", i, o, seq),
+                CloseOutput(o, i) => write!(f, "{}:closeOut:{}", i, o),
+                CloseInput(i) => write!(f, "{}:closeIn", i),
+            }
+        }
+    }
+    impl From<ZmqTransportEvent> for ZmqMessage {
+        fn from(value: ZmqTransportEvent) -> Self {
+            let mut topic = Vec::new();
+            value.write_topic(&mut topic).unwrap();
+            // first frame of the message is the topic
+            let mut msg = ZmqMessage::from(topic);
+            fn map_id<T>(id: T) -> i64
+            where
+                isize: From<T>,
+            {
+                isize::from(id) as i64
+            }
+            // second frame of the message is the payload
+            use crate::protoflow_zmq::{self, event::Payload, Event};
+            use prost::Message;
+            use ZmqTransportEvent::*;
+            let payload = match value {
+                Connect(output, input) => Payload::Connect(protoflow_zmq::Connect {
+                    output: map_id(output),
+                    input: map_id(input),
+                }),
+                AckConnection(output, input) => Payload::AckConnection(protoflow_zmq::AckConnection {
+                    output: map_id(output),
+                    input: map_id(input),
+                }),
+                Message(output, input, sequence, message) => Payload::Message(protoflow_zmq::Message {
+                    output: map_id(output),
+                    input: map_id(input),
+                    sequence,
+                    message: message.to_vec(),
+                }),
+                AckMessage(output, input, sequence) => Payload::AckMessage(protoflow_zmq::AckMessage {
+                    output: map_id(output),
+                    input: map_id(input),
+                    sequence,
+                }),
+                CloseOutput(output, input) => Payload::CloseOutput(protoflow_zmq::CloseOutput {
+                    output: map_id(output),
+                    input: map_id(input),
+                }),
+                CloseInput(input) => Payload::CloseInput(protoflow_zmq::CloseInput {
+                    input: map_id(input),
+                }),
+            };
+            let bytes = Event {
+                payload: Some(payload),
+            }
+            .encode_to_vec();
+            msg.push_back(bytes.into());
+            msg
+        }
+    }
+    impl TryFrom<ZmqMessage> for ZmqTransportEvent {
+        type Error = protoflow_core::DecodeError;
+        fn try_from(value: ZmqMessage) -> Result<Self, Self::Error> {
+            use crate::protoflow_zmq::{self, event::Payload, Event};
+            use prost::Message;
+            use protoflow_core::DecodeError;
+            fn map_id<T>(id: i64) -> Result<T, DecodeError>
+            where
+                T: TryFrom<isize>,
+                std::borrow::Cow<'static, str>: From<<T as TryFrom<isize>>::Error>,
+            {
+                (id as isize).try_into().map_err(DecodeError::new)
+            }
+            value
+                .get(1)
+                .ok_or_else(|| {
+                    protoflow_core::DecodeError::new("message contains less than two frames")
+                })
+                .and_then(|bytes| {
+                    let event = Event::decode(bytes.as_ref())?;
+                    use ZmqTransportEvent::*;
+                    Ok(match event.payload {
+                        None => {
+                            return Err(protoflow_core::DecodeError::new("message payload is empty"))
+                        }
+                        Some(Payload::Connect(protoflow_zmq::Connect { output, input })) => {
+                            Connect(map_id(output)?, map_id(input)?)
+                        }
+                        Some(Payload::AckConnection(protoflow_zmq::AckConnection {
+                            output,
+                            input,
+                        })) => AckConnection(map_id(output)?, map_id(input)?),
+                        Some(Payload::Message(protoflow_zmq::Message {
+                            output,
+                            input,
+                            sequence,
+                            message,
+                        })) => Message(
+                            map_id(output)?,
+                            map_id(input)?,
+                            sequence,
+                            Bytes::from(message),
+                        ),
+                        Some(Payload::AckMessage(protoflow_zmq::AckMessage {
+                            output,
+                            input,
+                            sequence,
+                        })) => AckMessage(map_id(output)?, map_id(input)?, sequence),
+                        Some(Payload::CloseOutput(protoflow_zmq::CloseOutput { output, input })) => {
+                            CloseOutput(map_id(output)?, map_id(input)?)
+                        }
+                        Some(Payload::CloseInput(protoflow_zmq::CloseInput { input })) => {
+                            CloseInput(map_id(input)?)
+                        }
+                    })
+                })
+        }
+    }

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Implement the ZeroMQ transport #27

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!