foster/deno
0
0
Fork 0
mirror of https://github.com/denoland/deno.git synced 2025-03-03 09:31:22 -05:00
Code Issues Packages 1 Wiki Activity

chore(ext/websocket): remove ping frame handling (#18847)

Browse source 
Automatically done in the fastwebsockets crate
This commit is contained in:
Divy Srivastava 2023-04-26 15:37:38 +05:30 • committed by GitHub
parent 106e4c475f
commit 9d7e3f84c8
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 53 additions and 92 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
ext/websocket/01_websocket.js
View file

@ -478,7 +478,7 @@ class WebSocket extends EventTarget {
this[_serverHandleIdleTimeout]();
break;
}
case 5: {
case 3: {
/* error */
this[_readyState] = CLOSED;
@ -492,10 +492,6 @@ class WebSocket extends EventTarget {
core.tryClose(this[_rid]);
break;
}
case 3: {
/* ping */
break;
}
default: {
/* close */
const code = kind;

20
ext/websocket/02_websocketstream.js
View file

@ -236,7 +236,7 @@ class WebSocketStream {
},
});
const pull = async (controller) => {
const { 0: kind, 1: value } = await core.opAsync(
const { 0: kind, 1: value } = await core.opAsync2(
"op_ws_next_event",
this[_rid],
);
@ -249,7 +249,11 @@ class WebSocketStream {
controller.enqueue(value);
break;
}
case 5: {
case 2: {
/* pong */
break;
}
case 3: {
/* error */
const err = new Error(value);
this[_closed].reject(err);
@ -257,17 +261,7 @@ class WebSocketStream {
core.tryClose(this[_rid]);
break;
}
case 3: {
/* ping */
await core.opAsync("op_ws_send_pong", this[_rid]);
await pull(controller);
break;
}
case 2: {
/* pong */
break;
}
case 6: {
case 4: {
/* closed */
this[_closed].resolve(undefined);
core.tryClose(this[_rid]);

119
ext/websocket/lib.rs
View file

@ -27,7 +27,6 @@ use http::Method;
use http::Request;
use http::Uri;
use hyper::Body;
use serde::Deserialize;
use serde::Serialize;
use std::borrow::Cow;
use std::cell::Cell;
@ -85,15 +84,6 @@ impl Resource for WsCancelResource {
}
}
#[derive(Deserialize)]
#[serde(tag = "kind", content = "value", rename_all = "camelCase")]
pub enum SendValue {
Text(String),
Binary(ZeroCopyBuf),
Pong,
Ping,
}
// This op is needed because creating a WS instance in JavaScript is a sync
// operation and should throw error when permissions are not fulfilled,
// but actual op that connects WS is async.
@ -301,9 +291,8 @@ pub enum MessageKind {
Text = 0,
Binary = 1,
Pong = 2,
Ping = 3,
Error = 5,
Closed = 6,
Error = 3,
Closed = 4,
}
pub struct ServerWebSocket {
@ -405,20 +394,6 @@ pub async fn op_ws_send_text(
.await
}
#[op]
pub async fn op_ws_send_ping(
state: Rc<RefCell<OpState>>,
rid: ResourceId,
) -> Result<(), AnyError> {
let resource = state
.borrow_mut()
.resource_table
.get::<ServerWebSocket>(rid)?;
resource
.write_frame(Frame::new(true, OpCode::Ping, None, vec![]))
.await
}
#[op]
pub async fn op_ws_send_pong(
state: Rc<RefCell<OpState>>,
@ -463,58 +438,55 @@ pub async fn op_ws_next_event(
.get::<ServerWebSocket>(rid)?;
let mut ws = RcRef::map(&resource, |r| &r.ws).borrow_mut().await;
let val = match ws.read_frame().await {
Ok(val) => val,
Err(err) => {
// No message was received, socket closed while we waited.
// Try close the stream, ignoring any errors, and report closed status to JavaScript.
if resource.closed.get() {
let _ = state.borrow_mut().resource_table.close(rid);
loop {
let val = match ws.read_frame().await {
Ok(val) => val,
Err(err) => {
// No message was received, socket closed while we waited.
// Try close the stream, ignoring any errors, and report closed status to JavaScript.
if resource.closed.get() {
let _ = state.borrow_mut().resource_table.close(rid);
return Ok((
MessageKind::Closed as u16,
StringOrBuffer::Buffer(vec![].into()),
));
}
return Ok((
MessageKind::Closed as u16,
StringOrBuffer::Buffer(vec![].into()),
MessageKind::Error as u16,
StringOrBuffer::String(err.to_string()),
));
}
};
return Ok((
MessageKind::Error as u16,
StringOrBuffer::String(err.to_string()),
));
}
};
break Ok(match val.opcode {
OpCode::Text => (
MessageKind::Text as u16,
StringOrBuffer::String(String::from_utf8(val.payload).unwrap()),
),
OpCode::Binary => (
MessageKind::Binary as u16,
StringOrBuffer::Buffer(val.payload.into()),
),
OpCode::Close => {
if val.payload.len() < 2 {
return Ok((1005, StringOrBuffer::String("".to_string())));
}
let res = match val.opcode {
OpCode::Text => (
MessageKind::Text as u16,
StringOrBuffer::String(String::from_utf8(val.payload).unwrap()),
),
OpCode::Binary => (
MessageKind::Binary as u16,
StringOrBuffer::Buffer(val.payload.into()),
),
OpCode::Close => {
if val.payload.len() < 2 {
return Ok((1005, StringOrBuffer::String("".to_string())));
let close_code =
CloseCode::from(u16::from_be_bytes([val.payload[0], val.payload[1]]));
let reason = String::from_utf8(val.payload[2..].to_vec()).unwrap();
(close_code.into(), StringOrBuffer::String(reason))
}
let close_code =
CloseCode::from(u16::from_be_bytes([val.payload[0], val.payload[1]]));
let reason = String::from_utf8(val.payload[2..].to_vec()).unwrap();
(close_code.into(), StringOrBuffer::String(reason))
}
OpCode::Ping => (
MessageKind::Ping as u16,
StringOrBuffer::Buffer(vec![].into()),
),
OpCode::Pong => (
MessageKind::Pong as u16,
StringOrBuffer::Buffer(vec![].into()),
),
OpCode::Continuation => {
return Err(type_error("Unexpected continuation frame"))
}
};
Ok(res)
OpCode::Pong => (
MessageKind::Pong as u16,
StringOrBuffer::Buffer(vec![].into()),
),
OpCode::Continuation | OpCode::Ping => {
continue;
}
});
}
}
deno_core::extension!(deno_websocket,
@ -527,7 +499,6 @@ deno_core::extension!(deno_websocket,
op_ws_next_event,
op_ws_send_binary,
op_ws_send_text,
op_ws_send_ping,
op_ws_send_pong,
op_ws_server_create,
],