websocket

[Menzer05]

I have to retrieve json data from my database to send them to an actix websocket client, all my attempts were unsuccessful, please help me

[mohsen-alizadeh]

Can you post your sample code?

[Menzer05]

main.rs
use actix::prelude::{Actor, ActorContext, StreamHandler};
use actix_files as fs;
use actix_web::{middleware, web, App, Error, HttpRequest, HttpResponse, HttpServer};
use actix_web_actors::ws;
mod souche;

async fn ws_index(r: HttpRequest, stream: web::Payload) -> Result<HttpResponse, Error> {
println!("{:?}", r);
let res = ws::start(MyWebSocket {}, &r, stream);
println!("{:?}", res);
res
}
struct MyWebSocket {

}

impl Actor for MyWebSocket {
type Context = ws::WebsocketContext;
fn started(&mut self, ctx: &mut Self::Context) {
ctx;
}
}

impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for MyWebSocket {
fn handle( &mut self, msg: Result<ws::Message, ws::ProtocolError>,ctx: &mut Self::Context,) {

    let bindex = r#"{"username":"hamid","code":200}"#; // just an example and it s work
    let mindex = r#"{"username":"zoza","code":2020}"#; // just an example  //
  
    let bb = souche::maresp().unwrap();
   
    match msg {           
        Ok(ws::Message::Text(msg)) => {
           if msg == "monmsg"{
            ctx.text(mindex);    // this msg work
           }else if &msg == "tonmsg"{
             ctx.text(bindex);  // this msg work too
           }
           else if &msg == "vosmsg"{ 
            //ctx.text(bb);           // this msg dont work
          }
        } 
        
        Ok(ws::Message::Close(reason)) => {
            ctx.close(reason);
            ctx.stop();
        }
        _ => ctx.stop(),
    }
    
}

}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
std::env::set_var("RUST_LOG", "actix_server=info,actix_web=info");
env_logger::init();
HttpServer::new(|| {
App::new()
.wrap(middleware::Logger::default())// enable logger
.service(web::resource("/ws/").route(web::get().to(ws_index)))// websocket route
.service(fs::Files::new("/", "static/").index_file("index.html"))// static files
})
// start http server on http://127.0.0.1:8000
.bind("127.0.0.1:8000")?
.run()
.await
}

souche.rs

use reqwest::*;
use actix_web::{web};
use serde_derive::{Deserialize, Serialize};
use actix::prelude::{Message};

#[derive(Debug, Deserialize, Serialize, Message)]
#[rtype(result = "()")]
pub struct Measurement {
origin: String,
}

#[derive(Debug, Deserialize, Serialize, Message)]
#[rtype(result = "()")]
pub struct Monjson {
userId: isize,
id: isize,
title: String,
completed: bool,
}

async fn index(info: web::Json) -> String {
format!("Welcome {}!", info.origin)
}

pub async fn newclient() -> Result { // sortie demandée: struct reqwest::response
let res = reqwest::get("https://jsonplaceholder.typicode.com/todos/1")
.await?
.json::()
.await?;
println!(" Monjson titre:\n\n{:#?}", res.title);
println!(" tout Monjson:\n\n{:#?}", res);
Ok(res) // sortie res = struct reqwest::response : Ok(T) ou Err(error)
}
// une autre reqwest

pub async fn maresp() -> Result {
let resp = reqwest::get("https://httpbin.org/ip")
.await?
.json::()
.await?;
Ok(resp)
}

static/index.html

<title>WebSocket demo</title>

<div>
    <form>
        <label for="numberfield">Number</label>
        <input type="text" id="numberfield" placeholder="12"/><br />
        <button type="button" id="sendBtn">Send</button>
    </form>
</div>
<div id="container"></div>

<script type="text/javascript" src="http://ajax.googleapis.com/ajax/libs/jquery/1.10.2/jquery.min.js"></script>
<script type="text/javascript">
    $(function () {
        var ws;
        if (window.WebSocket === undefined) {
            $("#container").append("Your browser does not support WebSockets");
            return;
        } else {
            ws = initWS();
        }  // "ws://localhost:8080/ws"
        function initWS() {
            var wsUri = (window.location.protocol=='https:'&&'wss://'||'ws://')+window.location.host + '/ws/';
            var adresse = "ws://localhost:8000/ws/";
            //conn = new WebSocket(wsUri);
            //var socket = new WebSocket(wsUri);
            var socket = new WebSocket(adresse);
            //var socket = new WebSocket("ws://localhost:8000");
                container = $("#container");
            socket.onopen = function() {
                container.append("<p>Socket is open</p>");
            };

            socket.onmessage = function (e) {
                container.append("<p> ceci est la reponse:" + e.data + "</p>");
                var message2 = JSON.parse(e.data);
                container.append("<p> ceci est la reponse:" + message2.code + "</p>");
                console.log (message2.code);
            };
            socket.onclose = function () {
                container.append("<p>Socket closed</p>");
            };
            return socket;
        }
        $("#sendBtn").click(function (e) {
            e.preventDefault();
           // ws.send(JSON.stringify({ Num: parseInt($("#numberfield").val()) }));
           ws.send($("#numberfield").val()) });
        });
    
</script>

[fakeshadow]

future(async block) does nothing unless you poll(await) it. Your maresp function actually never run.
There is a brief explaining for this you can give a read https://rust-lang.github.io/async-book/09_example/01_running_async_code.html

Back to your question you would want to await on maresp to get the actual output and actor future can help you to do that with following pattern.

else if &msg == "vosmsg"{ 
   souche::maresp()
        // Transform async {}.await to actor future.
        .into_actor(self)
        // map the output. res is `souche::maresp()`'s output. act is current actor(MyWebSocket). context is actor's ctx.
        .map(|res, act, ctx| {
            let bb = res.unwrap();
            ctx.text(bb);
        })
        // spawn the actor future on actor context.
        .spawn(ctx);
}

For how to use actor future you can check out actix doc.
https://docs.rs/actix/0.11.1/actix/fut/future/trait.ActorFuture.html
https://docs.rs/actix/0.11.1/actix/type.ResponseActFuture.html

[fakeshadow]

BTW for the code to work you would want to import actix crate and add use actix::prelude::* to your module where these APIs are called;

[Menzer05]

thank you fakeshadow for your kind cooperation, however I get this message,(rustc --explain E0599 ) (the method into_actor exists for enum std::result::Result<souche::Measurement, reqwest::Error>, but its trait bounds were not satisfied
method cannot be called on std::result::Result<souche::Measurement, reqwest::Error> due to unsatisfied trait bounds) it seems that (into_actor) is not recognized here, I use actix 0.10 and of course actix :: prelude:: * as you recommended to me

[fakeshadow]

That's not how ? work. If you have problem figuring out how ? function. This is a good read. https://doc.rust-lang.org/edition-guide/rust-2018/error-handling-and-panics/the-question-mark-operator-for-easier-error-handling.html

If you want to return error as websocket message you have to convert the message to String or binary format like Vec<u8>. and send it with WebsocketContext::text or WebsocketContext::binary.

[Menzer05]

I will therefore try to modify the structure of my files to follow the solution that you recommend to me, once again thank you to you also know that I am moving from golang because RUST is great but very hard

[fakeshadow]

You are welcome. reqwest::Error type impls Debug and Display trait so the easiest way to convert it to String would be like with previous example. And you can send the string with ctx.text to client.

        request()
            .into_actor(self)
            .map(|res, act, ctx| {
                let string: String = res.unwrap_or_else(|e| format!("{}", e));
                println!("response is {:?}", string);
                System::current().stop();
            })
            .spawn(ctx);

That said a more general used way is to use serde for serialize the error with a pattern you like.

[Menzer05]

bonjour amigo, maintenant mon code compile mais l'application echoue à l'execution! voilà le message que je reçois :"thread 'actix-rt:worker:0' panicked at 'there is no reactor running, must be called from the context of a Tokio 1.x runtime'"

else if &msg == "vosmsg"{
souche::maresp()

                 .into_actor(self)
                
                 .map(|res, act, ctx| {
                     let bb = res.unwrap();
                     let br = format!("{:?}",bb);
                     println!("mon br:{}", br);
                     ctx.text(br);
                 })
                 .spawn(ctx);
            }
        }

[Menzer05]

finally i found the error, i downgraded "reqwest" to version 10 because version 11 uses tokio 1. your help was really precious to me, a thousand thanks!
my final code:
else if &msg == "vosmsg"{
souche::maresp()
.into_actor(self)
.map(|res, act, ctx| {
let bro: String = res.unwrap_or_else(|e| format!("{}", e));
ctx.text(bro);
System::current().stop();
}).spawn(ctx);
}