如何制作 Rust Websocket 客户端？

Question

我尝试过使用不同的库和不同的实现，但我无法在 Rust 中获得有效的 WebSocket 客户端/侦听器。

我尝试编写一个处理程序：

extern crate ws;

use ws::{connect, listen, Handler, Sender, Handshake, Result, Message, CloseCode};

struct Client {
    out: Sender,
}

impl Handler for Client {
    fn on_open(&mut self, _: Handshake) -> Result<()> {
        self.out.send(r#"{"action": "authenticate","data": {"key_id": "<API_KEY>","secret_key": "<API_SECRET>"}}"#);
        self.out.send(r#"{"action": "listen","data": {"streams": ["AM.SPY"]}}"#)
    }

    fn on_message(&mut self, msg: Message) -> Result<()> {
        println!("message: {}", msg);
        Ok(())
    }
}

fn main() {
    if let Err(error) = listen("wss://data.alpaca.markets/stream", |out| {
        Client { out: out }
    }) {
        println!("Failed to create WebSocket due to: {:?}", error);
    }
}

我也尝试过这个：

extern crate ws;

use ws::{connect, CloseCode};

fn main() {
    if let Err(error) = connect("wss://data.alpaca.markets/stream", |out| {
        if out.send(r#"{"action": "authenticate","data": {"key_id": "<API_KEY>","secret_key": "<API_SECRET>"}}"#).is_err() {
            println!("Websocket couldn't queue an initial message.")
        } else {
            println!("Client sent message 'Hello WebSocket'. ")
        };

        if out.send(r#"{"action": "listen","data": {"streams": ["AM.SPY"]}}"#).is_err() {
            println!("Websocket couldn't queue an initial message.")
        } else {
            println!("Client sent message 'Hello WebSocket'. ")
        };

        move |msg| {
            println!("message: '{}'. ", msg);

            Ok(())
        }
    }) {
        println!("Failed to create WebSocket due to: {:?}", error);
    }
}

为了确保我尝试连接的连接不是问题，我在 JS 中编写了相同的代码。这确实有效。

const ws = require("ws");

const stream = new ws("wss://data.alpaca.markets/stream");

stream.on("open", () => {
    stream.send('{"action": "authenticate","data": {"key_id": "<API_KEY>","secret_key": "API_SECRET"}}');
    stream.send('{"action": "listen","data": {"streams": ["AM.SPY"]}}');
});

stream.on("message", (bar) => {
    process.stdout.write(`${bar}\n`);
});

在 Rust 代码的两个实例中，代码都会编译并运行，但 on_open 函数和 lambda 函数从未被调用。

先感谢您。

Answer 1

对于面临同样问题的任何人，我建议使用tungstenite和异步 websockets tokio-tungstenite

这是最终为我工作的代码：

use url::Url;
use tungstenite::{connect, Message};

let (mut socket, response) = connect(
    Url::parse("wss://data.alpaca.markets/stream").unwrap()
).expect("Can't connect");

socket.write_message(Message::Text(r#"{
    "action": "authenticate",
    "data": {
        "key_id": "API-KEY",
        "secret_key": "SECRET-KEY"
    }
}"#.into()));

socket.write_message(Message::Text(r#"{
    "action": "listen",
    "data": {
        "streams": ["AM.SPY"]
    }
}"#.into()));

loop {
    let msg = socket.read_message().expect("Error reading message");
    println!("Received: {}", msg);
}

在 Cargo.toml 中：

[dependencies]
tungstenite = {version = "0.16.0", features = ["native-tls"]}
url = "2.2.2"

我面临的问题是我使用的方法不适用于 TLS 流，而是 TCP 流。对于 tungstenite，如果启用 native-tls 功能，则 TCP 和 TLS 流都可以通过 connect 方法正确处理。