fn create_websocket_proxy(
  websocket: deno_websocket::tokio_tungstenite::WebSocketStream<
    hyper::upgrade::Upgraded,
  >,
) -> (WebSocketProxy, impl Future<Output = ()> + Send) {
  // The 'outbound' channel carries messages sent to the websocket.
  let (outbound_tx, outbound_rx) = mpsc::unbounded();

  // The 'inbound' channel carries messages received from the websocket.
  let (inbound_tx, inbound_rx) = mpsc::unbounded();

  let proxy = WebSocketProxy {
    tx: outbound_tx,
    rx: inbound_rx,
  };

  // The pump future takes care of forwarding messages between the websocket
  // and channels. It resolves to () when either side disconnects, ignoring any
  // errors.
  let pump = async move {
    let (websocket_tx, websocket_rx) = websocket.split();

    let outbound_pump =
      outbound_rx.map(Ok).forward(websocket_tx).map_err(|_| ());

    let inbound_pump = websocket_rx
      .map(|msg| inbound_tx.unbounded_send(msg))
      .map_err(|_| ())
      .try_collect::<()>();

    let _ = future::try_join(outbound_pump, inbound_pump).await;
  };

  (proxy, pump)
}
/// Creates a future that proxies messages sent and received on a warp WebSocket
/// to a UnboundedSender/UnboundedReceiver pair. We need this to sidestep
/// Tokio's task budget, which causes issues when DenoInspector::poll_sessions()
/// needs to block the thread because JavaScript execution is paused.
///
/// This works because UnboundedSender/UnboundedReceiver are implemented in the
/// 'futures' crate, therefore they can't participate in Tokio's cooperative
/// task yielding.
///
/// A tuple is returned, where the first element is a duplex channel that can
/// be used to send/receive messages on the websocket, and the second element
/// is a future that does the forwarding.
Creates a future that proxies messages sent and received on a warp WebSocket
to a UnboundedSender/UnboundedReceiver pair. We need this to sidestep
Tokio's task budget, which causes issues when DenoInspector::poll_sessions()
needs to block the thread because JavaScript execution is paused.

This works because UnboundedSender/UnboundedReceiver are implemented in the
'futures' crate, therefore they can't participate in Tokio's cooperative
task yielding.

A tuple is returned, where the first element is a duplex channel that can
be used to send/receive messages on the websocket, and the second element
is a future that does the forwarding.
Creates a future that proxies messages sent and received on a warp WebSocket to a UnboundedSender / UnboundedReceiver pair . We need this to sidestep Tokio ' s task budget which causes issues when DenoInspector :: poll_sessions () needs to block the thread because JavaScript execution is paused . This works because UnboundedSender / UnboundedReceiver are implemented in the ' futures ' crate therefore they can ' t participate in Tokio ' s cooperative task yielding . A tuple is returned where the first element is a duplex channel that can be used to send / receive messages on the websocket and the second element is a future that does the forwarding .
fn create_websocket_proxy(
  websocket: deno_websocket::tokio_tungstenite::WebSocketStream<
    hyper::upgrade::Upgraded,
  >,
) -> (WebSocketProxy, impl Future<Output = ()> + Send) {
  // The 'outbound' channel carries messages sent to the websocket.
  let (outbound_tx, outbound_rx) = mpsc::unbounded();
  // The 'inbound' channel carries messages received from the websocket.
  let (inbound_tx, inbound_rx) = mpsc::unbounded();
  let proxy = WebSocketProxy {
    tx: outbound_tx,
    rx: inbound_rx,
  };
  // The pump future takes care of forwarding messages between the websocket
  // and channels. It resolves to () when either side disconnects, ignoring any
  // errors.
  let pump = async move {
    let (websocket_tx, websocket_rx) = websocket.split();
    let outbound_pump =
      outbound_rx.map(Ok).forward(websocket_tx).map_err(|_| ());
    let inbound_pump = websocket_rx
      .map(|msg| inbound_tx.unbounded_send(msg))
      .map_err(|_| ())
      .try_collect::<()>();
    let _ = future::try_join(outbound_pump, inbound_pump).await;
  };
  (proxy, pump)
}
fn create_websocket_proxy ( websocket : deno_websocket :: tokio_tungstenite :: WebSocketStream < hyper :: upgrade :: Upgraded , > , ) -> ( WebSocketProxy , impl Future < Output = ( ) > + Send ) { // The 'outbound' channel carries messages sent to the websocket. let ( outbound_tx , outbound_rx ) = mpsc :: unbounded ( ) ; // The 'inbound' channel carries messages received from the websocket. let ( inbound_tx , inbound_rx ) = mpsc :: unbounded ( ) ; let proxy = WebSocketProxy { tx : outbound_tx , rx : inbound_rx , } ; // The pump future takes care of forwarding messages between the websocket // and channels. It resolves to () when either side disconnects, ignoring any // errors. let pump = async move { let ( websocket_tx , websocket_rx ) = websocket . split ( ) ; let outbound_pump = outbound_rx . map ( Ok ) . forward ( websocket_tx ) . map_err ( | _ | ( ) ) ; let inbound_pump = websocket_rx . map ( | msg | inbound_tx . unbounded_send ( msg ) ) . map_err ( | _ | ( ) ) . try_collect :: < ( ) > ( ) ; let _ = future :: try_join ( outbound_pump , inbound_pump ) . await ; } ; ( proxy , pump ) }
Creates a future that proxies messages sent and received on a warp WebSocket
to a UnboundedSender/UnboundedReceiver pair.
Creates a future that proxies messages sent and received on a warp WebSocket to a UnboundedSender / UnboundedReceiver pair .
websocket
deno_websocket::tokio_tungstenite::WebSocketStream<
    hyper::upgrade::Upgraded,
  >
websocket
deno_websocket::tokio_tungstenite::WebSocketStream<
    hyper::upgrade::Upgraded,
  >