Pauan / rust-signals

Zero-cost functional reactive Signals for Rust
MIT License
673 stars 37 forks source link

from TcpStream to signal example #58

Open nothingIsSomething opened 2 years ago

nothingIsSomething commented 2 years ago

hello I'm new in rust, very used to rxjs, could you help me with a simple example please? , I have already read the crate tutorial but I'm still a bit lost, I'm trying to create a signal from a tcpStream.

use tokio;
use futures;
use futures::{SinkExt, StreamExt};
use futures_signals::signal::Broadcaster;
use tokio_tungstenite::{connect_async};
use futures_signals::signal::SignalExt; //for Iterator trait (gives for_each)
use futures_signals::signal::from_stream;

pub async fn new(url: &str) {
let (ws_stream, _response) = connect_async(url).await.expect("Failed to connect");
let (mut write, read) = ws_stream.split();

    //1. obsevable
    let signal_from_stream = from_stream(read);//.to_stream();

    //2. broadcast to clone the signal.
    let broadcaster = Broadcaster::new(signal_from_stream);

    //let signal = broadcaster.signal();

   //3. create observers 
   let observer_A = broadcaster.signal().for_each(|message| {
    println!("Observer A receiving... {}", message);

    let msg = message.unwrap().to_string().unwrap(); 
    println!("Received: {:?}", msg);
    async {}
    });

    let observer_B = broadcaster.signal().for_each(|message| {
        println!("Observer B receiving... {}", message);

        let msg = message.unwrap().to_string().unwrap(); 
        println!("Received: {:?}", msg);
        async {}
    });

    tokio::spawn(observer_A);
    tokio::spawn(observer_B);
}
Pauan commented 2 years ago

What's the issue that you're having? That code seems reasonable to me, though it usually doesn't make sense to convert a Stream into a Signal, because when you convert a Stream into a Signal, the Signal will only contain the most-recent message, all other messages are dropped.

nothingIsSomething commented 2 years ago

As it says in the tutorial, it doesn't make sense if you need the intermediate values, well, in this case I'm only interested in the last message, not in buffering all the incoming messages, I guess I gain in performance with the signal, I need to learn more about the streams and signals, I'm trying to find the differences and similarities with rxjs to gain understanding faster.

this is the error that I get:

the method `signal` exists for struct `futures_signals::signal::Broadcaster<futures_signals::signal::FromStream<futures::stream::SplitStream<tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>>>>`, but its trait bounds were not satisfied

method cannot be called on `futures_signals::signal::Broadcaster<futures_signals::signal::FromStream<futures::stream::SplitStream<tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>>>>` due to unsatisfied trait bounds

note: the following trait bounds were not satisfied:
      `std::option::Option<std::result::Result<tokio_tungstenite::tungstenite::Message, tokio_tungstenite::tungstenite::Error>>: std::marker::Copy`rustc(E0599)
client.rs(22, 33): method cannot be called on `futures_signals::signal::Broadcaster<futures_signals::signal::FromStream<futures::stream::SplitStream<tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>>>>` due to unsatisfied trait bounds
option.rs(515, 1): doesn't satisfy `_: std::marker::Copy`
Pauan commented 2 years ago

I guess I gain in performance with the signal

That depends on what you're trying to do. When you convert a Stream into a Signal, it's still pulling all of the Stream values like usual, so the performance is the same, it's just dropping the intermediate values.

The buffer comes from the Stream itself, so the buffer still exists either way. And when you use StreamExt methods like map it doesn't create a new buffer, so the performance should be the same in most situations.

In addition to that, Broadcaster is pretty costly in performance, so in your code snippet it might actually be faster to use Streams.

this is the error that I get

That error is saying that the message type doesn't implement Copy, and so you can't use it with signal().

If the message type implements Clone then you can use signal_cloned() instead.

If the message type doesn't implement Copy or Clone then you have to use signal_ref(|x| { ... }) which gives you a reference to the message.

All of this is only needed because you're using Broadcaster, if you didn't use Broadcaster then it becomes much simpler and faster:

let future = from_stream(read).for_each(move |message| {
    if let Some(message) = message {
        // ...
    }
    async {}
});

If you want to continue to use Broadcaster, then I suggest doing something like this:

let broadcaster = Broadcaster::new(from_stream(read).map(|message| {
    message.map(|message| message.to_string())
}));

This converts the message into a String, so now you can use broadcaster.signal_cloned(), because String implements Clone.

Pauan commented 2 years ago

I'm trying to find the differences and similarities with rxjs to gain understanding faster.

RxJS is closer to Streams than it is to Signals. In RxJS they combine the concept of Streams and Signals together, which causes tons of bugs and problems. In Rust the two concepts are very cleanly separated:

You can think of it as being like the difference between having a mutable Vec and a mutable variable:

In this analogy, a Stream is like an asynchronous Vec, whereas a Signal is like an asynchronous variable. They have very different behavior and APIs, because they do different things, they serve different purposes. Signals do not replace Streams, instead they complement Streams, because both are useful.

nothingIsSomething commented 2 years ago

it's all very clear to me now! thank you so much! :)