Fix RPC
This commit is contained in:
@@ -22,6 +22,10 @@ use crate::{
|
|||||||
state::{AppState, RpcConnectionGuard},
|
state::{AppState, RpcConnectionGuard},
|
||||||
};
|
};
|
||||||
|
|
||||||
|
// jsonrpc2-ws 1.0.0-beta23 passes JavaScript `null` as the browser
|
||||||
|
// WebSocket subprotocol, which Chromium serializes as the token "null".
|
||||||
|
const JSONRPC2_WS_BROWSER_PROTOCOL: &str = "null";
|
||||||
|
|
||||||
#[derive(Debug, Deserialize)]
|
#[derive(Debug, Deserialize)]
|
||||||
struct RpcRequest {
|
struct RpcRequest {
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
@@ -35,6 +39,7 @@ struct RpcRequest {
|
|||||||
|
|
||||||
pub async fn upgrade(State(state): State<Arc<AppState>>, websocket: WebSocketUpgrade) -> Response {
|
pub async fn upgrade(State(state): State<Arc<AppState>>, websocket: WebSocketUpgrade) -> Response {
|
||||||
websocket
|
websocket
|
||||||
|
.protocols([JSONRPC2_WS_BROWSER_PROTOCOL])
|
||||||
.max_message_size(1024 * 1024)
|
.max_message_size(1024 * 1024)
|
||||||
.on_upgrade(move |socket| connection(socket, state))
|
.on_upgrade(move |socket| connection(socket, state))
|
||||||
}
|
}
|
||||||
@@ -202,11 +207,21 @@ fn update_rooms(
|
|||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use std::collections::HashSet;
|
use std::{collections::HashSet, time::Duration};
|
||||||
|
|
||||||
|
use axum::{Router, routing::get};
|
||||||
|
use mirakurun_core::config::ConfigPaths;
|
||||||
use serde_json::json;
|
use serde_json::json;
|
||||||
|
use tempfile::tempdir;
|
||||||
|
use tokio::{
|
||||||
|
io::{AsyncReadExt, AsyncWriteExt},
|
||||||
|
net::{TcpListener, TcpStream},
|
||||||
|
};
|
||||||
|
use tokio_util::sync::CancellationToken;
|
||||||
|
|
||||||
use super::update_rooms;
|
use crate::state::AppState;
|
||||||
|
|
||||||
|
use super::{JSONRPC2_WS_BROWSER_PROTOCOL, update_rooms, upgrade};
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn joins_and_leaves_rooms() {
|
fn joins_and_leaves_rooms() {
|
||||||
@@ -216,4 +231,64 @@ mod tests {
|
|||||||
update_rooms(&mut rooms, &json!({"rooms": ["events:program"]}), false).expect("leave room");
|
update_rooms(&mut rooms, &json!({"rooms": ["events:program"]}), false).expect("leave room");
|
||||||
assert!(rooms.is_empty());
|
assert!(rooms.is_empty());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn selects_jsonrpc2_ws_browser_subprotocol() {
|
||||||
|
let directory = tempdir().expect("create tempdir");
|
||||||
|
let state = AppState::load(ConfigPaths {
|
||||||
|
server: directory.path().join("server.yml"),
|
||||||
|
tuners: directory.path().join("tuners.yml"),
|
||||||
|
channels: directory.path().join("channels.yml"),
|
||||||
|
services_db: directory.path().join("services.json"),
|
||||||
|
programs_db: directory.path().join("programs.json"),
|
||||||
|
logo_data_dir: directory.path().join("logo-data"),
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.expect("load state");
|
||||||
|
let app = Router::new().route("/rpc", get(upgrade)).with_state(state);
|
||||||
|
let listener = TcpListener::bind("127.0.0.1:0")
|
||||||
|
.await
|
||||||
|
.expect("bind listener");
|
||||||
|
let address = listener.local_addr().expect("listener address");
|
||||||
|
let shutdown = CancellationToken::new();
|
||||||
|
let server_shutdown = shutdown.clone();
|
||||||
|
let server = tokio::spawn(async move {
|
||||||
|
axum::serve(listener, app)
|
||||||
|
.with_graceful_shutdown(server_shutdown.cancelled_owned())
|
||||||
|
.await
|
||||||
|
});
|
||||||
|
let mut socket = TcpStream::connect(address).await.expect("connect server");
|
||||||
|
socket
|
||||||
|
.write_all(
|
||||||
|
format!(
|
||||||
|
"GET /rpc HTTP/1.1\r\n\
|
||||||
|
Host: {address}\r\n\
|
||||||
|
Connection: Upgrade\r\n\
|
||||||
|
Upgrade: websocket\r\n\
|
||||||
|
Sec-WebSocket-Version: 13\r\n\
|
||||||
|
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\
|
||||||
|
Sec-WebSocket-Protocol: {JSONRPC2_WS_BROWSER_PROTOCOL}\r\n\
|
||||||
|
\r\n"
|
||||||
|
)
|
||||||
|
.as_bytes(),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("write upgrade request");
|
||||||
|
let mut response = vec![0; 2048];
|
||||||
|
let length = tokio::time::timeout(Duration::from_secs(1), socket.read(&mut response))
|
||||||
|
.await
|
||||||
|
.expect("upgrade response timeout")
|
||||||
|
.expect("read upgrade response");
|
||||||
|
let response = String::from_utf8_lossy(&response[..length]).to_ascii_lowercase();
|
||||||
|
assert!(response.starts_with("http/1.1 101 switching protocols\r\n"));
|
||||||
|
assert!(response.contains("\r\nsec-websocket-protocol: null\r\n"));
|
||||||
|
|
||||||
|
drop(socket);
|
||||||
|
shutdown.cancel();
|
||||||
|
tokio::time::timeout(Duration::from_secs(1), server)
|
||||||
|
.await
|
||||||
|
.expect("server shutdown timeout")
|
||||||
|
.expect("join server")
|
||||||
|
.expect("run server");
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -215,7 +215,7 @@ class State extends EventEmitter<StateEventTypes> {
|
|||||||
|
|
||||||
private _initRPC() {
|
private _initRPC() {
|
||||||
const rpc = this._rpc = new RPCClient(`${location.protocol === "https:" ? "wss:" : "ws:"}//${location.host}/rpc`, {
|
const rpc = this._rpc = new RPCClient(`${location.protocol === "https:" ? "wss:" : "ws:"}//${location.host}/rpc`, {
|
||||||
protocols: null,
|
protocols: [],
|
||||||
bufferSendingMessages: false
|
bufferSendingMessages: false
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user