refactor: rewrite websocket to use op2 macro (#20140)

Co-authored-by: Bartek Iwańczuk <biwanczuk@gmail.com>
This commit is contained in:
Divy Srivastava 2023-10-05 03:13:58 +05:30 committed by GitHub
parent cbddf5756e
commit 1a81b2826d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 75 additions and 43 deletions

12
Cargo.lock generated
View file

@ -1205,9 +1205,9 @@ dependencies = [
[[package]] [[package]]
name = "deno_core" name = "deno_core"
version = "0.220.0" version = "0.221.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b9a21d0411c638803d21b5059b8cac2e847933fe7272c7eadafec7890f4421f1" checksum = "cf694d66a55050bdfad2036765f93547392ba7c131adc7fcd5e91fa31e291c51"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"bytes", "bytes",
@ -1594,9 +1594,9 @@ dependencies = [
[[package]] [[package]]
name = "deno_ops" name = "deno_ops"
version = "0.96.0" version = "0.97.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "28f8b32e5575daabc57c42339496bed91236a9af422f500674fb0da01689e35e" checksum = "aab46b645231decaf3b03d9eb99a5207e4e66c8c53b7774000b44e3ae43e0f99"
dependencies = [ dependencies = [
"deno-proc-macro-rules", "deno-proc-macro-rules",
"lazy-regex", "lazy-regex",
@ -4821,9 +4821,9 @@ dependencies = [
[[package]] [[package]]
name = "serde_v8" name = "serde_v8"
version = "0.129.0" version = "0.130.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "00f52380e42a53d4c3884c21969621620fdd0bcc1c165b02c2d80e2f41f5730e" checksum = "2c6078fc5e16615b092b26199e7f1d7d2d336848623d32e00f95e74618274111"
dependencies = [ dependencies = [
"bytes", "bytes",
"derive_more", "derive_more",

View file

@ -40,7 +40,7 @@ repository = "https://github.com/denoland/deno"
[workspace.dependencies] [workspace.dependencies]
deno_ast = { version = "0.29.3", features = ["transpiling"] } deno_ast = { version = "0.29.3", features = ["transpiling"] }
deno_core = { version = "0.220.0" } deno_core = { version = "0.221.0" }
deno_runtime = { version = "0.128.0", path = "./runtime" } deno_runtime = { version = "0.128.0", path = "./runtime" }
napi_sym = { version = "0.50.0", path = "./cli/napi/sym" } napi_sym = { version = "0.50.0", path = "./cli/napi/sym" }

View file

@ -146,6 +146,7 @@ const OP_DETAILS = {
"op_ws_next_event": ["receive the next message on a WebSocket", "closing a `WebSocket` or `WebSocketStream`"], "op_ws_next_event": ["receive the next message on a WebSocket", "closing a `WebSocket` or `WebSocketStream`"],
"op_ws_send_text": ["send a message on a WebSocket", "closing a `WebSocket` or `WebSocketStream`"], "op_ws_send_text": ["send a message on a WebSocket", "closing a `WebSocket` or `WebSocketStream`"],
"op_ws_send_binary": ["send a message on a WebSocket", "closing a `WebSocket` or `WebSocketStream`"], "op_ws_send_binary": ["send a message on a WebSocket", "closing a `WebSocket` or `WebSocketStream`"],
"op_ws_send_binary_ab": ["send a message on a WebSocket", "closing a `WebSocket` or `WebSocketStream`"],
"op_ws_send_ping": ["send a message on a WebSocket", "closing a `WebSocket` or `WebSocketStream`"], "op_ws_send_ping": ["send a message on a WebSocket", "closing a `WebSocket` or `WebSocketStream`"],
"op_ws_send_pong": ["send a message on a WebSocket", "closing a `WebSocket` or `WebSocketStream`"], "op_ws_send_pong": ["send a message on a WebSocket", "closing a `WebSocket` or `WebSocketStream`"],
}; };

View file

@ -26,7 +26,6 @@ const {
ArrayPrototypeJoin, ArrayPrototypeJoin,
ArrayPrototypeMap, ArrayPrototypeMap,
ArrayPrototypeSome, ArrayPrototypeSome,
DataView,
ErrorPrototypeToString, ErrorPrototypeToString,
ObjectDefineProperties, ObjectDefineProperties,
ObjectPrototypeIsPrototypeOf, ObjectPrototypeIsPrototypeOf,
@ -50,6 +49,7 @@ const {
op_ws_create, op_ws_create,
op_ws_close, op_ws_close,
op_ws_send_binary, op_ws_send_binary,
op_ws_send_binary_ab,
op_ws_send_text, op_ws_send_text,
op_ws_next_event, op_ws_next_event,
op_ws_get_buffer, op_ws_get_buffer,
@ -336,11 +336,7 @@ class WebSocket extends EventTarget {
PromisePrototypeThen( PromisePrototypeThen(
// deno-lint-ignore prefer-primordials // deno-lint-ignore prefer-primordials
data.slice().arrayBuffer(), data.slice().arrayBuffer(),
(ab) => (ab) => op_ws_send_binary_ab(this[_rid], ab),
op_ws_send_binary(
this[_rid],
new DataView(ab),
),
); );
} else { } else {
const string = String(data); const string = String(data);

View file

@ -5,6 +5,7 @@ use deno_core::error::invalid_hostname;
use deno_core::error::type_error; use deno_core::error::type_error;
use deno_core::error::AnyError; use deno_core::error::AnyError;
use deno_core::op; use deno_core::op;
use deno_core::op2;
use deno_core::url; use deno_core::url;
use deno_core::AsyncMutFuture; use deno_core::AsyncMutFuture;
use deno_core::AsyncRefCell; use deno_core::AsyncRefCell;
@ -108,11 +109,12 @@ impl Resource for WsCancelResource {
// This op is needed because creating a WS instance in JavaScript is a sync // This op is needed because creating a WS instance in JavaScript is a sync
// operation and should throw error when permissions are not fulfilled, // operation and should throw error when permissions are not fulfilled,
// but actual op that connects WS is async. // but actual op that connects WS is async.
#[op] #[op2]
#[smi]
pub fn op_ws_check_permission_and_cancel_handle<WP>( pub fn op_ws_check_permission_and_cancel_handle<WP>(
state: &mut OpState, state: &mut OpState,
api_name: String, #[string] api_name: String,
url: String, #[string] url: String,
cancel_handle: bool, cancel_handle: bool,
) -> Result<Option<ResourceId>, AnyError> ) -> Result<Option<ResourceId>, AnyError>
where where
@ -167,14 +169,15 @@ async fn handshake<S: AsyncRead + AsyncWrite + Send + Unpin + 'static>(
Ok((stream, response)) Ok((stream, response))
} }
#[op] #[op2(async)]
#[serde]
pub async fn op_ws_create<WP>( pub async fn op_ws_create<WP>(
state: Rc<RefCell<OpState>>, state: Rc<RefCell<OpState>>,
api_name: String, #[string] api_name: String,
url: String, #[string] url: String,
protocols: String, #[string] protocols: String,
cancel_handle: Option<ResourceId>, #[smi] cancel_handle: Option<ResourceId>,
headers: Option<Vec<(ByteString, ByteString)>>, #[serde] headers: Option<Vec<(ByteString, ByteString)>>,
) -> Result<CreateResponse, AnyError> ) -> Result<CreateResponse, AnyError>
where where
WP: WebSocketPermissions + 'static, WP: WebSocketPermissions + 'static,
@ -407,8 +410,7 @@ pub fn ws_create_server_stream(
Ok(rid) Ok(rid)
} }
#[op(fast)] fn send_binary(state: &mut OpState, rid: ResourceId, data: &[u8]) {
pub fn op_ws_send_binary(state: &mut OpState, rid: ResourceId, data: &[u8]) {
let resource = state.resource_table.get::<ServerWebSocket>(rid).unwrap(); let resource = state.resource_table.get::<ServerWebSocket>(rid).unwrap();
let data = data.to_vec(); let data = data.to_vec();
let len = data.len(); let len = data.len();
@ -426,8 +428,30 @@ pub fn op_ws_send_binary(state: &mut OpState, rid: ResourceId, data: &[u8]) {
}); });
} }
#[op(fast)] #[op2(fast)]
pub fn op_ws_send_text(state: &mut OpState, rid: ResourceId, data: String) { pub fn op_ws_send_binary(
state: &mut OpState,
#[smi] rid: ResourceId,
#[anybuffer] data: &[u8],
) {
send_binary(state, rid, data)
}
#[op2(fast)]
pub fn op_ws_send_binary_ab(
state: &mut OpState,
#[smi] rid: ResourceId,
#[arraybuffer] data: &[u8],
) {
send_binary(state, rid, data)
}
#[op2(fast)]
pub fn op_ws_send_text(
state: &mut OpState,
#[smi] rid: ResourceId,
#[string] data: String,
) {
let resource = state.resource_table.get::<ServerWebSocket>(rid).unwrap(); let resource = state.resource_table.get::<ServerWebSocket>(rid).unwrap();
let len = data.len(); let len = data.len();
resource.buffered.set(resource.buffered.get() + len); resource.buffered.set(resource.buffered.get() + len);
@ -487,8 +511,12 @@ pub async fn op_ws_send_text_async(
const EMPTY_PAYLOAD: &[u8] = &[]; const EMPTY_PAYLOAD: &[u8] = &[];
#[op(fast)] #[op2(fast)]
pub fn op_ws_get_buffered_amount(state: &mut OpState, rid: ResourceId) -> u32 { #[smi]
pub fn op_ws_get_buffered_amount(
state: &mut OpState,
#[smi] rid: ResourceId,
) -> u32 {
state state
.resource_table .resource_table
.get::<ServerWebSocket>(rid) .get::<ServerWebSocket>(rid)
@ -497,10 +525,10 @@ pub fn op_ws_get_buffered_amount(state: &mut OpState, rid: ResourceId) -> u32 {
.get() as u32 .get() as u32
} }
#[op] #[op2(async)]
pub async fn op_ws_send_pong( pub async fn op_ws_send_pong(
state: Rc<RefCell<OpState>>, state: Rc<RefCell<OpState>>,
rid: ResourceId, #[smi] rid: ResourceId,
) -> Result<(), AnyError> { ) -> Result<(), AnyError> {
let resource = state let resource = state
.borrow_mut() .borrow_mut()
@ -512,10 +540,10 @@ pub async fn op_ws_send_pong(
.await .await
} }
#[op] #[op2(async)]
pub async fn op_ws_send_ping( pub async fn op_ws_send_ping(
state: Rc<RefCell<OpState>>, state: Rc<RefCell<OpState>>,
rid: ResourceId, #[smi] rid: ResourceId,
) -> Result<(), AnyError> { ) -> Result<(), AnyError> {
let resource = state let resource = state
.borrow_mut() .borrow_mut()
@ -530,12 +558,12 @@ pub async fn op_ws_send_ping(
.await .await
} }
#[op(deferred)] #[op2(async(lazy))]
pub async fn op_ws_close( pub async fn op_ws_close(
state: Rc<RefCell<OpState>>, state: Rc<RefCell<OpState>>,
rid: ResourceId, #[smi] rid: ResourceId,
code: Option<u16>, #[smi] code: Option<u16>,
reason: Option<String>, #[string] reason: Option<String>,
) -> Result<(), AnyError> { ) -> Result<(), AnyError> {
let resource = state let resource = state
.borrow_mut() .borrow_mut()
@ -551,23 +579,29 @@ pub async fn op_ws_close(
Ok(()) Ok(())
} }
#[op] #[op2]
pub fn op_ws_get_buffer(state: &mut OpState, rid: ResourceId) -> ToJsBuffer { #[serde]
pub fn op_ws_get_buffer(
state: &mut OpState,
#[smi] rid: ResourceId,
) -> ToJsBuffer {
let resource = state.resource_table.get::<ServerWebSocket>(rid).unwrap(); let resource = state.resource_table.get::<ServerWebSocket>(rid).unwrap();
resource.buffer.take().unwrap().into() resource.buffer.take().unwrap().into()
} }
#[op] #[op2]
#[string]
pub fn op_ws_get_buffer_as_string( pub fn op_ws_get_buffer_as_string(
state: &mut OpState, state: &mut OpState,
rid: ResourceId, #[smi] rid: ResourceId,
) -> String { ) -> String {
let resource = state.resource_table.get::<ServerWebSocket>(rid).unwrap(); let resource = state.resource_table.get::<ServerWebSocket>(rid).unwrap();
resource.string.take().unwrap() resource.string.take().unwrap()
} }
#[op] #[op2]
pub fn op_ws_get_error(state: &mut OpState, rid: ResourceId) -> String { #[string]
pub fn op_ws_get_error(state: &mut OpState, #[smi] rid: ResourceId) -> String {
let Ok(resource) = state.resource_table.get::<ServerWebSocket>(rid) else { let Ok(resource) = state.resource_table.get::<ServerWebSocket>(rid) else {
return "Bad resource".into(); return "Bad resource".into();
}; };
@ -660,6 +694,7 @@ deno_core::extension!(deno_websocket,
op_ws_get_buffer_as_string, op_ws_get_buffer_as_string,
op_ws_get_error, op_ws_get_error,
op_ws_send_binary, op_ws_send_binary,
op_ws_send_binary_ab,
op_ws_send_text, op_ws_send_text,
op_ws_send_binary_async, op_ws_send_binary_async,
op_ws_send_text_async, op_ws_send_text_async,

View file

@ -31,7 +31,7 @@ fn op_main_module(state: &mut OpState) -> Result<String, AnyError> {
/// This is an op instead of being done at initialization time because /// This is an op instead of being done at initialization time because
/// it's expensive to retrieve the ppid on Windows. /// it's expensive to retrieve the ppid on Windows.
#[op2] #[op2(fast)]
#[bigint] #[bigint]
pub fn op_ppid() -> i64 { pub fn op_ppid() -> i64 {
#[cfg(windows)] #[cfg(windows)]