rdelfin / async-zmq Goto Github PK

So that dealer should return sink and stream separately. Current API seem to make using one exclude another (consume mut ref) while the underlying socket only require ref.

Is there a new release planned for this crate? I was planning on using the vendored zmq feature but I see the latest release doesn't support it. When is a new release expected?

In general, only one context is needed in an application. For inproc sockets, the same context must be used. It should be possible to provide the context when creating a socket.

After reading this post from Stjepang, I realize this crate doesn't need to depend on any runtime at all. It already has its own reactor. Although initial intention is because async-std doesn't public its reactor, it looks like it would be better just port reactor only as time goes by. We can keep our dependencies and size small in this way. The Major benefit from this is users don't need to worry about what runtime to choose and how to configure it.

A problem with multipart messages in the current implementation is that senders need to allocate all of the message parts to a VecDeque before sending. Likewise, receivers must allocate message parts before they can be processed. Applications such as an XPUB/XSUB proxy would be very inefficient with this API.

Ideally, a sender should be able to stream parts from another source, without an intermediate allocation. Receivers should be able to start processing the first part, before the rest arrive.

The API that I am envisioning could look something like this:

• socket.send(msg) can send only single messages
• Use socket.send_multipart(iter) to send messages from an IntoIterator
• Use socket.send_multipart_stream(stream) to stream messages
• socket.recv() returns an error if the message is multipart
• socket.recv_multipart() returns a Stream. It works for either multipart or single message.

This is quite a big change, which will impact the public API and internal structures. It will need some careful design, and my intention with opening this issue is to trigger some conversation first.

hey is it possible to have mac os and windows support ?

Since the introduction of the runtime agnostic code (I guess!) the poll_event() function no longer returns Poll::Pending, only Poll::Ready, possibly with a Err(Error::EAGAIN) value. This makes the examples fail with

thread 'main' panicked at 'called `Result::unwrap()` on an `Err` value: Unexpected(Resource temporarily unavailable)', examples/reply.rs:12:19

Please update to latest zmq version 0.10.0

In README, it states:

Another common issue when people adopting a library is to deal with its error handling flow. To prevent introducing more overhead, async_zmq uses the exact same Result/Error type in zmq crate and re-export them.

I believe this is a mistake. There is a very standard way of modeling errors in Rust, which is quite different from the zmq::Error type. It is more important to be consistent with other crates in the Rust ecosystem than to mirror the underlying C types.

In my experience of working with the zmq crate, it is very hard to know which of the 28 error variants are possible for any given operation. Most are generally not applicable, or are applicable to certain operations but not others. Instead of having to refer to the ZMQ C documention, I would much rather lean on the Rust type system to tell me exactly which errors I need to handle in a particular case.

A good example of this done well is the design of error handling for channels in std, or crossbeam_channel. Each operation has its own error type, which represents exactly the errors that could occur.

I encountered an issue with the struct Reply when trying to use it in a concurrent context. I want to know why it is not Send and if there is a way to make it work. Thank you for the help.

Here is the code I am trying to run, the src/main.rs file:

use async_zmq;
use futures_util::stream::StreamExt;
use tokio;

async fn foo() {
    let mut socket = async_zmq::reply("tcp://localhost:1234")
        .unwrap()
        .bind()
        .unwrap();

    while let Some(Ok(msg)) = socket.next().await {
        let f = socket.send(msg);
        f.await.unwrap();  // <-- Commenting this line out makes the code compile.
    }

}

#[tokio::main]
async fn main() {
    // The spawn(x) requires x to be Send, and the compiler complains about the foo() function not
    // being Send. The compiler gives a hint that "within `Reply<std::vec::IntoIter<Message>,
    // Message>`, the trait `Sync` is not implemented for `*mut c_void`". I don't understand where
    // does the `*mut c_void` come from.
    let handle = tokio::spawn(foo());

    // tokio::spawn(bar());
    // tokio::spawn(other_tasks());

    handle.await.unwrap();
}

And the Cargo.toml file:

[package]
name = "test"
version = "0.1.0"
edition = "2021"

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[dependencies]

# common utils
futures = "0.3"
futures-util = "0.3"
tokio = { version = "1.0", features = ["full"] }
async_zmq = "0.3.3"