diff options
Diffstat (limited to 'core/inspector.rs')
| -rw-r--r-- | core/inspector.rs | 82 |
1 files changed, 28 insertions, 54 deletions
diff --git a/core/inspector.rs b/core/inspector.rs index d8a51e601..ef59cf413 100644 --- a/core/inspector.rs +++ b/core/inspector.rs @@ -41,9 +41,7 @@ use std::thread; /// If first argument is `None` then it's a notification, otherwise /// it's a message. pub type SessionProxySender = UnboundedSender<(Option<i32>, String)>; -// TODO(bartlomieju): does it even need to send a Result? -// It seems `Vec<u8>` would be enough -pub type SessionProxyReceiver = UnboundedReceiver<Result<Vec<u8>, AnyError>>; +pub type SessionProxyReceiver = UnboundedReceiver<Vec<u8>>; /// Encapsulates an UnboundedSender/UnboundedReceiver pair that together form /// a duplex channel for sending/receiving messages in V8 session. @@ -89,11 +87,11 @@ pub struct JsRuntimeInspector { impl Drop for JsRuntimeInspector { fn drop(&mut self) { - // Since the waker is cloneable, it might outlive the inspector itself. + // Since the waker is cloneable, it might outlive the inspector itself. // Set the poll state to 'dropped' so it doesn't attempt to request an // interrupt from the isolate. self.waker.update(|w| w.poll_state = PollState::Dropped); - // TODO(bartlomieju): this comment is out of date + // V8 automatically deletes all sessions when an `V8Inspector` instance is // deleted, however InspectorSession also has a drop handler that cleans // up after itself. To avoid a double free, make sure the inspector is @@ -134,9 +132,11 @@ impl v8::inspector::V8InspectorClientImpl for JsRuntimeInspector { } } -/// `JsRuntimeInspector` implements a Future so that it can poll for new incoming -/// connections and messages from the WebSocket server. The Worker that owns -/// this `JsRuntimeInspector` will call this function from `Worker::poll()`. +/// Polling `JsRuntimeInspector` allows inspector to accept new incoming +/// connections and "pump" messages in different sessions. +/// +/// It should be polled on tick of event loop, ie. in `JsRuntime::poll_event_loop` +/// function. impl Future for JsRuntimeInspector { type Output = (); fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<()> { @@ -501,8 +501,7 @@ impl task::ArcWake for InspectorWaker { struct InspectorSession { v8_channel: v8::inspector::ChannelBase, v8_session: Rc<RefCell<v8::UniqueRef<v8::inspector::V8InspectorSession>>>, - proxy_tx: SessionProxySender, - proxy_rx_handler: Pin<Box<dyn Future<Output = ()> + 'static>>, + proxy: InspectorSessionProxy, } impl InspectorSession { @@ -524,15 +523,10 @@ impl InspectorSession { v8::inspector::StringView::empty(), ))); - let (proxy_tx, proxy_rx) = session_proxy.split(); - let proxy_rx_handler = - Self::receive_from_proxy(v8_session.clone(), proxy_rx); - Self { v8_channel, v8_session, - proxy_tx, - proxy_rx_handler, + proxy: session_proxy, } }) } @@ -546,46 +540,13 @@ impl InspectorSession { v8_session_ptr.dispatch_protocol_message(msg); } - // TODO(bartlomieju): this function should be reworked into `impl Future` - // or `impl Stream` - /// Returns a future that receives messages from the proxy and dispatches - /// them to the V8 session. - fn receive_from_proxy( - v8_session_rc: Rc< - RefCell<v8::UniqueRef<v8::inspector::V8InspectorSession>>, - >, - proxy_rx: SessionProxyReceiver, - ) -> Pin<Box<dyn Future<Output = ()> + 'static>> { - async move { - let result = proxy_rx - .map_ok(move |msg| { - let msg = v8::inspector::StringView::from(msg.as_slice()); - let mut v8_session = v8_session_rc.borrow_mut(); - let v8_session_ptr = v8_session.as_mut(); - v8_session_ptr.dispatch_protocol_message(msg); - }) - .try_collect::<()>() - .await; - - // TODO(bartlomieju): ideally these prints should be moved - // to `server.rs` as they are unwanted in context of REPL/coverage collection - // but right now they do not pose a huge problem. Investigate how to - // move them to `server.rs`. - match result { - Ok(_) => eprintln!("Debugger session ended."), - Err(err) => eprintln!("Debugger session ended: {}.", err), - }; - } - .boxed_local() - } - fn send_message( &self, maybe_call_id: Option<i32>, msg: v8::UniquePtr<v8::inspector::StringBuffer>, ) { let msg = msg.unwrap().string().to_string(); - let _ = self.proxy_tx.unbounded_send((maybe_call_id, msg)); + let _ = self.proxy.tx.unbounded_send((maybe_call_id, msg)); } pub fn break_on_next_statement(&mut self) { @@ -626,17 +587,30 @@ impl v8::inspector::ChannelImpl for InspectorSession { fn flush_protocol_notifications(&mut self) {} } +/// This is a "pump" future takes care of receiving messages and dispatching +/// them to the inspector. It resolves when receiver closes. impl Future for InspectorSession { type Output = (); fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> { - self.proxy_rx_handler.poll_unpin(cx) + while let Poll::Ready(maybe_msg) = self.proxy.rx.poll_next_unpin(cx) { + if let Some(msg) = maybe_msg { + let msg = v8::inspector::StringView::from(msg.as_slice()); + let mut v8_session = self.v8_session.borrow_mut(); + let v8_session_ptr = v8_session.as_mut(); + v8_session_ptr.dispatch_protocol_message(msg); + } else { + return Poll::Ready(()); + } + } + + Poll::Pending } } /// A local inspector session that can be used to send and receive protocol messages directly on /// the same thread as an isolate. pub struct LocalInspectorSession { - v8_session_tx: UnboundedSender<Result<Vec<u8>, AnyError>>, + v8_session_tx: UnboundedSender<Vec<u8>>, v8_session_rx: UnboundedReceiver<(Option<i32>, String)>, response_tx_map: HashMap<i32, oneshot::Sender<serde_json::Value>>, next_message_id: i32, @@ -645,7 +619,7 @@ pub struct LocalInspectorSession { impl LocalInspectorSession { pub fn new( - v8_session_tx: UnboundedSender<Result<Vec<u8>, AnyError>>, + v8_session_tx: UnboundedSender<Vec<u8>>, v8_session_rx: UnboundedReceiver<(Option<i32>, String)>, ) -> Self { let response_tx_map = HashMap::new(); @@ -687,7 +661,7 @@ impl LocalInspectorSession { let raw_message = serde_json::to_string(&message).unwrap(); self .v8_session_tx - .unbounded_send(Ok(raw_message.as_bytes().to_vec())) + .unbounded_send(raw_message.as_bytes().to_vec()) .unwrap(); loop { |