devtools: Track multiple clients better, and cleanup streams when a client isn't reachable.

components/devtools/actor.rs

@@ -3,6 +3,7 @@
  * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
 
 /// General actor system infrastructure.
+use crate::StreamId;
 use devtools_traits::PreciseTime;
 use serde_json::{Map, Value};
 use std::any::Any;
@@ -21,18 +22,20 @@ pub enum ActorMessageStatus {
 /// A common trait for all devtools actors that encompasses an immutable name
 /// and the ability to process messages that are directed to particular actors.
 /// TODO: ensure the name is immutable
-pub trait Actor: Any + ActorAsAny {
+pub(crate) trait Actor: Any + ActorAsAny {
     fn handle_message(
         &self,
         registry: &ActorRegistry,
         msg_type: &str,
         msg: &Map<String, Value>,
         stream: &mut TcpStream,
+        id: StreamId,
     ) -> Result<ActorMessageStatus, ()>;
     fn name(&self) -> String;
+    fn cleanup(&self, _id: StreamId) {}
 }
 
-pub trait ActorAsAny {
+pub(crate) trait ActorAsAny {
     fn actor_as_any(&self) -> &dyn Any;
     fn actor_as_any_mut(&mut self) -> &mut dyn Any;
 }
@@ -71,6 +74,12 @@ impl ActorRegistry {
         }
     }
 
+    pub(crate) fn cleanup(&self, id: StreamId) {
+        for actor in self.actors.values() {
+            actor.cleanup(id);
+        }
+    }
+
     /// Creating shareable registry
     pub fn create_shareable(self) -> Arc<Mutex<ActorRegistry>> {
         if let Some(shareable) = self.shareable {
@@ -131,11 +140,11 @@ impl ActorRegistry {
     }
 
     /// Add an actor to the registry of known actors that can receive messages.
-    pub fn register(&mut self, actor: Box<dyn Actor + Send>) {
+    pub(crate) fn register(&mut self, actor: Box<dyn Actor + Send>) {
         self.actors.insert(actor.name(), actor);
     }
 
-    pub fn register_later(&self, actor: Box<dyn Actor + Send>) {
+    pub(crate) fn register_later(&self, actor: Box<dyn Actor + Send>) {
         let mut actors = self.new_actors.borrow_mut();
         actors.push(actor);
     }
@@ -154,10 +163,11 @@ impl ActorRegistry {
 
     /// Attempt to process a message as directed by its `to` property. If the actor is not
     /// found or does not indicate that it knew how to process the message, ignore the failure.
-    pub fn handle_message(
+    pub(crate) fn handle_message(
         &mut self,
         msg: &Map<String, Value>,
         stream: &mut TcpStream,
+        id: StreamId,
     ) -> Result<(), ()> {
         let to = match msg.get("to") {
             Some(to) => to.as_str().unwrap(),
@@ -171,7 +181,7 @@ impl ActorRegistry {
             None => debug!("message received for unknown actor \"{}\"", to),
             Some(actor) => {
                 let msg_type = msg.get("type").unwrap().as_str().unwrap();
-                if actor.handle_message(self, msg_type, msg, stream)? !=
+                if actor.handle_message(self, msg_type, msg, stream, id)? !=
                     ActorMessageStatus::Processed
                 {
                     debug!(

components/devtools/actors/browsing_context.rs

@@ -17,6 +17,7 @@ use crate::actors::tab::TabDescriptorActor;
 use crate::actors::thread::ThreadActor;
 use crate::actors::timeline::TimelineActor;
 use crate::protocol::JsonPacketStream;
+use crate::StreamId;
 use devtools_traits::DevtoolScriptControlMsg::{self, WantsLiveNotifications};
 use devtools_traits::DevtoolsPageInfo;
 use devtools_traits::NavigationState;
@@ -24,6 +25,7 @@ use ipc_channel::ipc::IpcSender;
 use msg::constellation_msg::{BrowsingContextId, PipelineId};
 use serde_json::{Map, Value};
 use std::cell::{Cell, RefCell};
+use std::collections::HashMap;
 use std::net::TcpStream;
 
 #[derive(Serialize)]
@@ -118,7 +120,7 @@ pub struct BrowsingContextActorMsg {
     manifestActor: String,*/
 }
 
-pub struct BrowsingContextActor {
+pub(crate) struct BrowsingContextActor {
     pub name: String,
     pub title: RefCell<String>,
     pub url: RefCell<String>,
@@ -131,7 +133,7 @@ pub struct BrowsingContextActor {
     pub styleSheets: String,
     pub thread: String,
     pub tab: String,
-    pub streams: RefCell<Vec<TcpStream>>,
+    pub streams: RefCell<HashMap<StreamId, TcpStream>>,
     pub browsing_context_id: BrowsingContextId,
     pub active_pipeline: Cell<PipelineId>,
     pub script_chan: IpcSender<DevtoolScriptControlMsg>,
@@ -148,6 +150,7 @@ impl Actor for BrowsingContextActor {
         msg_type: &str,
         msg: &Map<String, Value>,
         stream: &mut TcpStream,
+        id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(match msg_type {
             "reconfigure" => {
@@ -181,26 +184,26 @@ impl Actor for BrowsingContextActor {
                         watchpoints: false,
                     },
                 };
-                self.streams.borrow_mut().push(stream.try_clone().unwrap());
+
-                stream.write_json_packet(&msg);
+                if stream.write_json_packet(&msg).is_err() {
+                    return Ok(ActorMessageStatus::Processed);
+                }
+                self.streams
+                    .borrow_mut()
+                    .insert(id, stream.try_clone().unwrap());
                 self.script_chan
                     .send(WantsLiveNotifications(self.active_pipeline.get(), true))
                     .unwrap();
                 ActorMessageStatus::Processed
             },
 
-            //FIXME: The current implementation won't work for multiple connections. Need to ensure
-            //       that the correct stream is removed.
             "detach" => {
                 let msg = BrowsingContextDetachedReply {
                     from: self.name(),
                     type_: "detached".to_owned(),
                 };
-                self.streams.borrow_mut().pop();
+                let _ = stream.write_json_packet(&msg);
-                stream.write_json_packet(&msg);
+                self.cleanup(id);
-                self.script_chan
-                    .send(WantsLiveNotifications(self.active_pipeline.get(), false))
-                    .unwrap();
                 ActorMessageStatus::Processed
             },
 
@@ -224,13 +227,22 @@ impl Actor for BrowsingContextActor {
                     from: self.name(),
                     workers: vec![],
                 };
-                stream.write_json_packet(&msg);
+                let _ = stream.write_json_packet(&msg);
                 ActorMessageStatus::Processed
             },
 
             _ => ActorMessageStatus::Ignored,
         })
     }
+
+    fn cleanup(&self, id: StreamId) {
+        self.streams.borrow_mut().remove(&id);
+        if self.streams.borrow().is_empty() {
+            self.script_chan
+                .send(WantsLiveNotifications(self.active_pipeline.get(), false))
+                .unwrap();
+        }
+    }
 }
 
 impl BrowsingContextActor {
@@ -284,7 +296,7 @@ impl BrowsingContextActor {
             styleSheets: styleSheets.name(),
             tab: tabdesc.name(),
             thread: thread.name(),
-            streams: RefCell::new(Vec::new()),
+            streams: RefCell::new(HashMap::new()),
             browsing_context_id: id,
             active_pipeline: Cell::new(pipeline),
         };
@@ -347,8 +359,8 @@ impl BrowsingContextActor {
             state: state.to_owned(),
             isFrameSwitching: false,
         };
-        for stream in &mut *self.streams.borrow_mut() {
+        for stream in self.streams.borrow_mut().values_mut() {
-            stream.write_json_packet(&msg);
+            let _ = stream.write_json_packet(&msg);
         }
     }
 

components/devtools/actors/console.rs

@@ -12,7 +12,7 @@ use crate::actors::browsing_context::BrowsingContextActor;
 use crate::actors::object::ObjectActor;
 use crate::actors::worker::WorkerActor;
 use crate::protocol::JsonPacketStream;
-use crate::UniqueId;
+use crate::{StreamId, UniqueId};
 use devtools_traits::CachedConsoleMessage;
 use devtools_traits::ConsoleMessage;
 use devtools_traits::EvaluateJSReply::{ActorValue, BooleanValue, StringValue};
@@ -23,7 +23,7 @@ use devtools_traits::{
 use ipc_channel::ipc::{self, IpcSender};
 use msg::constellation_msg::TEST_PIPELINE_ID;
 use serde_json::{self, Map, Number, Value};
-use std::cell::{RefCell, RefMut};
+use std::cell::RefCell;
 use std::collections::HashMap;
 use std::net::TcpStream;
 use time::precise_time_ns;
@@ -130,15 +130,20 @@ impl ConsoleActor {
         }
     }
 
-    fn streams_mut<'a>(&self, registry: &'a ActorRegistry) -> RefMut<'a, Vec<TcpStream>> {
+    fn streams_mut<'a>(&self, registry: &'a ActorRegistry, cb: impl Fn(&mut TcpStream)) {
         match &self.root {
             Root::BrowsingContext(bc) => registry
                 .find::<BrowsingContextActor>(bc)
                 .streams
-                .borrow_mut(),
+                .borrow_mut()
-            Root::DedicatedWorker(worker) => {
+                .values_mut()
-                registry.find::<WorkerActor>(worker).streams.borrow_mut()
+                .for_each(cb),
-            },
+            Root::DedicatedWorker(worker) => registry
+                .find::<WorkerActor>(worker)
+                .streams
+                .borrow_mut()
+                .values_mut()
+                .for_each(cb),
         }
     }
 
@@ -255,9 +260,9 @@ impl ConsoleActor {
                 type_: "pageError".to_owned(),
                 pageError: page_error,
             };
-            for stream in &mut *self.streams_mut(registry) {
+            self.streams_mut(registry, |stream| {
-                stream.write_json_packet(&msg);
+                let _ = stream.write_json_packet(&msg);
-            }
+            });
         }
     }
 
@@ -303,9 +308,9 @@ impl ConsoleActor {
                     columnNumber: console_message.columnNumber,
                 },
             };
-            for stream in &mut *self.streams_mut(registry) {
+            self.streams_mut(registry, |stream| {
-                stream.write_json_packet(&msg);
+                let _ = stream.write_json_packet(&msg);
-            }
+            });
         }
     }
 }
@@ -321,6 +326,7 @@ impl Actor for ConsoleActor {
         msg_type: &str,
         msg: &Map<String, Value>,
         stream: &mut TcpStream,
+        _id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(match msg_type {
             "clearMessagesCache" => {

components/devtools/actors/device.rs

@@ -5,6 +5,7 @@
 use crate::actor::{Actor, ActorMessageStatus, ActorRegistry};
 use crate::protocol::JsonPacketStream;
 use crate::protocol::{ActorDescription, Method};
+use crate::StreamId;
 use serde_json::{Map, Value};
 use std::net::TcpStream;
 
@@ -34,6 +35,7 @@ impl Actor for DeviceActor {
         msg_type: &str,
         _msg: &Map<String, Value>,
         stream: &mut TcpStream,
+        _id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(match msg_type {
             "getDescription" => {

components/devtools/actors/emulation.rs

@@ -3,6 +3,7 @@
  * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
 
 use crate::actor::{Actor, ActorMessageStatus, ActorRegistry};
+use crate::StreamId;
 use serde_json::{Map, Value};
 use std::net::TcpStream;
 
@@ -21,6 +22,7 @@ impl Actor for EmulationActor {
         msg_type: &str,
         _msg: &Map<String, Value>,
         _stream: &mut TcpStream,
+        _id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(match msg_type {
             _ => ActorMessageStatus::Ignored,

components/devtools/actors/framerate.rs

@@ -4,6 +4,7 @@
 
 use crate::actor::{Actor, ActorMessageStatus, ActorRegistry};
 use crate::actors::timeline::HighResolutionStamp;
+use crate::StreamId;
 use devtools_traits::DevtoolScriptControlMsg;
 use ipc_channel::ipc::IpcSender;
 use msg::constellation_msg::PipelineId;
@@ -32,6 +33,7 @@ impl Actor for FramerateActor {
         _msg_type: &str,
         _msg: &Map<String, Value>,
         _stream: &mut TcpStream,
+        _id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(ActorMessageStatus::Ignored)
     }

components/devtools/actors/inspector.rs

@@ -8,6 +8,7 @@
 use crate::actor::{Actor, ActorMessageStatus, ActorRegistry};
 use crate::actors::browsing_context::BrowsingContextActor;
 use crate::protocol::JsonPacketStream;
+use crate::StreamId;
 use devtools_traits::DevtoolScriptControlMsg::{GetChildren, GetDocumentElement, GetRootNode};
 use devtools_traits::DevtoolScriptControlMsg::{GetLayout, ModifyAttribute};
 use devtools_traits::{ComputedNodeLayout, DevtoolScriptControlMsg, NodeInfo};
@@ -68,6 +69,7 @@ impl Actor for HighlighterActor {
         msg_type: &str,
         _msg: &Map<String, Value>,
         stream: &mut TcpStream,
+        _id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(match msg_type {
             "showBoxModel" => {
@@ -103,6 +105,7 @@ impl Actor for NodeActor {
         msg_type: &str,
         msg: &Map<String, Value>,
         stream: &mut TcpStream,
+        _id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(match msg_type {
             "modifyAttributes" => {
@@ -289,6 +292,7 @@ impl Actor for WalkerActor {
         msg_type: &str,
         msg: &Map<String, Value>,
         stream: &mut TcpStream,
+        _id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(match msg_type {
             "querySelector" => {
@@ -471,6 +475,7 @@ impl Actor for PageStyleActor {
         msg_type: &str,
         msg: &Map<String, Value>,
         stream: &mut TcpStream,
+        _id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(match msg_type {
             "getApplied" => {
@@ -596,6 +601,7 @@ impl Actor for InspectorActor {
         msg_type: &str,
         _msg: &Map<String, Value>,
         stream: &mut TcpStream,
+        _id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         let browsing_context = registry.find::<BrowsingContextActor>(&self.browsing_context);
         let pipeline = browsing_context.active_pipeline.get();

components/devtools/actors/memory.rs

@@ -3,6 +3,7 @@
  * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
 
 use crate::actor::{Actor, ActorMessageStatus, ActorRegistry};
+use crate::StreamId;
 use serde_json::{Map, Value};
 use std::net::TcpStream;
 
@@ -34,6 +35,7 @@ impl Actor for MemoryActor {
         _msg_type: &str,
         _msg: &Map<String, Value>,
         _stream: &mut TcpStream,
+        _id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(ActorMessageStatus::Ignored)
     }

components/devtools/actors/network_event.rs

@@ -8,6 +8,7 @@
 
 use crate::actor::{Actor, ActorMessageStatus, ActorRegistry};
 use crate::protocol::JsonPacketStream;
+use crate::StreamId;
 use devtools_traits::HttpRequest as DevtoolsHttpRequest;
 use devtools_traits::HttpResponse as DevtoolsHttpResponse;
 use headers::{ContentType, Cookie, HeaderMapExt};
@@ -180,6 +181,7 @@ impl Actor for NetworkEventActor {
         msg_type: &str,
         _msg: &Map<String, Value>,
         stream: &mut TcpStream,
+        _id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(match msg_type {
             "getRequestHeaders" => {

components/devtools/actors/object.rs

@@ -3,6 +3,7 @@
  * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
 
 use crate::actor::{Actor, ActorMessageStatus, ActorRegistry};
+use crate::StreamId;
 use serde_json::{Map, Value};
 use std::net::TcpStream;
 
@@ -21,6 +22,7 @@ impl Actor for ObjectActor {
         _: &str,
         _: &Map<String, Value>,
         _: &mut TcpStream,
+        _: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(ActorMessageStatus::Ignored)
     }

components/devtools/actors/performance.rs

@@ -4,6 +4,7 @@
 
 use crate::actor::{Actor, ActorMessageStatus, ActorRegistry};
 use crate::protocol::{ActorDescription, JsonPacketStream, Method};
+use crate::StreamId;
 use serde_json::{Map, Value};
 use std::net::TcpStream;
 
@@ -57,6 +58,7 @@ impl Actor for PerformanceActor {
         msg_type: &str,
         _msg: &Map<String, Value>,
         stream: &mut TcpStream,
+        _id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(match msg_type {
             "connect" => {

components/devtools/actors/preference.rs

@@ -4,6 +4,7 @@
 
 use crate::actor::{Actor, ActorMessageStatus, ActorRegistry};
 use crate::protocol::JsonPacketStream;
+use crate::StreamId;
 use serde_json::{Map, Value};
 use std::net::TcpStream;
 
@@ -28,6 +29,7 @@ impl Actor for PreferenceActor {
         msg_type: &str,
         _msg: &Map<String, Value>,
         stream: &mut TcpStream,
+        _id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(match msg_type {
             "getBoolPref" => {

components/devtools/actors/process.rs

@@ -4,6 +4,7 @@
 
 use crate::actor::{Actor, ActorMessageStatus, ActorRegistry};
 use crate::protocol::JsonPacketStream;
+use crate::StreamId;
 use serde_json::{Map, Value};
 use std::net::TcpStream;
 
@@ -34,6 +35,7 @@ impl Actor for ProcessActor {
         msg_type: &str,
         _msg: &Map<String, Value>,
         stream: &mut TcpStream,
+        _id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(match msg_type {
             "listWorkers" => {

components/devtools/actors/profiler.rs

@@ -3,6 +3,7 @@
  * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
 
 use crate::actor::{Actor, ActorMessageStatus, ActorRegistry};
+use crate::StreamId;
 use serde_json::{Map, Value};
 use std::net::TcpStream;
 
@@ -21,6 +22,7 @@ impl Actor for ProfilerActor {
         _msg_type: &str,
         _msg: &Map<String, Value>,
         _stream: &mut TcpStream,
+        _id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(ActorMessageStatus::Ignored)
     }

components/devtools/actors/root.rs

@@ -12,6 +12,7 @@ use crate::actors::performance::PerformanceActor;
 use crate::actors::tab::{TabDescriptorActor, TabDescriptorActorMsg};
 use crate::actors::worker::{WorkerActor, WorkerMsg};
 use crate::protocol::{ActorDescription, JsonPacketStream};
+use crate::StreamId;
 use serde_json::{Map, Value};
 use std::net::TcpStream;
 
@@ -124,6 +125,7 @@ impl Actor for RootActor {
         msg_type: &str,
         _msg: &Map<String, Value>,
         stream: &mut TcpStream,
+        _id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(match msg_type {
             "listAddons" => {

components/devtools/actors/stylesheets.rs

@@ -4,6 +4,7 @@
 
 use crate::actor::{Actor, ActorMessageStatus, ActorRegistry};
 use crate::protocol::JsonPacketStream;
+use crate::StreamId;
 use serde_json::{Map, Value};
 use std::net::TcpStream;
 
@@ -27,6 +28,7 @@ impl Actor for StyleSheetsActor {
         msg_type: &str,
         _msg: &Map<String, Value>,
         stream: &mut TcpStream,
+        _id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(match msg_type {
             "getStyleSheets" => {

components/devtools/actors/tab.rs

@@ -6,6 +6,7 @@ use crate::actor::{Actor, ActorMessageStatus, ActorRegistry};
 use crate::actors::browsing_context::{BrowsingContextActor, BrowsingContextActorMsg};
 use crate::actors::root::RootActor;
 use crate::protocol::JsonPacketStream;
+use crate::StreamId;
 use serde_json::{Map, Value};
 use std::net::TcpStream;
 
@@ -48,6 +49,7 @@ impl Actor for TabDescriptorActor {
         msg_type: &str,
         _msg: &Map<String, Value>,
         stream: &mut TcpStream,
+        _id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(match msg_type {
             "getTarget" => {

components/devtools/actors/thread.rs

@@ -4,6 +4,7 @@
 
 use crate::actor::{Actor, ActorMessageStatus, ActorRegistry};
 use crate::protocol::JsonPacketStream;
+use crate::StreamId;
 use serde_json::{Map, Value};
 use std::net::TcpStream;
 
@@ -84,6 +85,7 @@ impl Actor for ThreadActor {
         msg_type: &str,
         _msg: &Map<String, Value>,
         stream: &mut TcpStream,
+        _id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(match msg_type {
             "attach" => {

components/devtools/actors/timeline.rs

@@ -6,6 +6,7 @@ use crate::actor::{Actor, ActorMessageStatus, ActorRegistry};
 use crate::actors::framerate::FramerateActor;
 use crate::actors::memory::{MemoryActor, TimelineMemoryReply};
 use crate::protocol::JsonPacketStream;
+use crate::StreamId;
 use devtools_traits::DevtoolScriptControlMsg;
 use devtools_traits::DevtoolScriptControlMsg::{DropTimelineMarkers, SetTimelineMarkers};
 use devtools_traits::{PreciseTime, TimelineMarker, TimelineMarkerType};
@@ -188,6 +189,7 @@ impl Actor for TimelineActor {
         msg_type: &str,
         msg: &Map<String, Value>,
         stream: &mut TcpStream,
+        _id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(match msg_type {
             "start" => {
@@ -202,6 +204,7 @@ impl Actor for TimelineActor {
                     ))
                     .unwrap();
 
+                //TODO: support multiple connections by using root actor's streams instead.
                 *self.stream.borrow_mut() = stream.try_clone().ok();
 
                 // init memory actor
@@ -256,6 +259,7 @@ impl Actor for TimelineActor {
                     ))
                     .unwrap();
 
+                //TODO: move this to the cleanup method.
                 if let Some(ref actor_name) = *self.framerate_actor.borrow() {
                     registry.drop_actor_later(actor_name.clone());
                 }

components/devtools/actors/worker.rs

@@ -4,6 +4,7 @@
 
 use crate::actor::{Actor, ActorMessageStatus, ActorRegistry};
 use crate::protocol::JsonPacketStream;
+use crate::StreamId;
 use devtools_traits::DevtoolScriptControlMsg::WantsLiveNotifications;
 use devtools_traits::{DevtoolScriptControlMsg, WorkerId};
 use ipc_channel::ipc::IpcSender;
@@ -11,6 +12,7 @@ use msg::constellation_msg::TEST_PIPELINE_ID;
 use serde_json::{Map, Value};
 use servo_url::ServoUrl;
 use std::cell::RefCell;
+use std::collections::HashMap;
 use std::net::TcpStream;
 
 #[derive(Clone, Copy)]
@@ -21,7 +23,7 @@ pub enum WorkerType {
     Service = 2,
 }
 
-pub struct WorkerActor {
+pub(crate) struct WorkerActor {
     pub name: String,
     pub console: String,
     pub thread: String,
@@ -29,7 +31,7 @@ pub struct WorkerActor {
     pub url: ServoUrl,
     pub type_: WorkerType,
     pub script_chan: IpcSender<DevtoolScriptControlMsg>,
-    pub streams: RefCell<Vec<TcpStream>>,
+    pub streams: RefCell<HashMap<StreamId, TcpStream>>,
 }
 
 impl WorkerActor {
@@ -58,6 +60,7 @@ impl Actor for WorkerActor {
         msg_type: &str,
         _msg: &Map<String, Value>,
         stream: &mut TcpStream,
+        id: StreamId,
     ) -> Result<ActorMessageStatus, ()> {
         Ok(match msg_type {
             "attach" => {
@@ -66,8 +69,12 @@ impl Actor for WorkerActor {
                     type_: "attached".to_owned(),
                     url: self.url.as_str().to_owned(),
                 };
-                self.streams.borrow_mut().push(stream.try_clone().unwrap());
+                if stream.write_json_packet(&msg).is_err() {
-                stream.write_json_packet(&msg);
+                    return Ok(ActorMessageStatus::Processed);
+                }
+                self.streams
+                    .borrow_mut()
+                    .insert(id, stream.try_clone().unwrap());
                 // FIXME: fix messages to not require forging a pipeline for worker messages
                 self.script_chan
                     .send(WantsLiveNotifications(TEST_PIPELINE_ID, true))
@@ -91,18 +98,23 @@ impl Actor for WorkerActor {
                     from: self.name(),
                     type_: "detached".to_string(),
                 };
-                // FIXME: we should ensure we're removing the correct stream.
+                let _ = stream.write_json_packet(&msg);
-                self.streams.borrow_mut().pop();
+                self.cleanup(id);
-                stream.write_json_packet(&msg);
-                self.script_chan
-                    .send(WantsLiveNotifications(TEST_PIPELINE_ID, false))
-                    .unwrap();
                 ActorMessageStatus::Processed
             },
 
             _ => ActorMessageStatus::Ignored,
         })
     }
+
+    fn cleanup(&self, id: StreamId) {
+        self.streams.borrow_mut().remove(&id);
+        if self.streams.borrow().is_empty() {
+            self.script_chan
+                .send(WantsLiveNotifications(TEST_PIPELINE_ID, false))
+                .unwrap();
+        }
+    }
 }
 
 #[derive(Serialize)]

components/devtools/lib.rs

@@ -127,6 +127,9 @@ pub fn start_server(port: u16, embedder: EmbedderProxy) -> Sender<DevtoolsContro
     sender
 }
 
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+pub(crate) struct StreamId(u32);
+
 fn run_server(
     sender: Sender<DevtoolsControlMsg>,
     receiver: Receiver<DevtoolsControlMsg>,
@@ -188,7 +191,7 @@ fn run_server(
     let mut actor_workers: HashMap<WorkerId, String> = HashMap::new();
 
     /// Process the input from a single devtools client until EOF.
-    fn handle_client(actors: Arc<Mutex<ActorRegistry>>, mut stream: TcpStream) {
+    fn handle_client(actors: Arc<Mutex<ActorRegistry>>, mut stream: TcpStream, id: StreamId) {
         debug!("connection established to {}", stream.peer_addr().unwrap());
         {
             let actors = actors.lock().unwrap();
@@ -202,11 +205,11 @@ fn run_server(
         'outer: loop {
             match stream.read_json_packet() {
                 Ok(Some(json_packet)) => {
-                    if let Err(()) = actors
+                    if let Err(()) = actors.lock().unwrap().handle_message(
-                        .lock()
+                        json_packet.as_object().unwrap(),
-                        .unwrap()
+                        &mut stream,
-                        .handle_message(json_packet.as_object().unwrap(), &mut stream)
+                        id,
-                    {
+                    ) {
                         debug!("error: devtools actor stopped responding");
                         let _ = stream.shutdown(Shutdown::Both);
                         break 'outer;
@@ -222,6 +225,8 @@ fn run_server(
                 },
             }
         }
+
+        actors.lock().unwrap().cleanup(id);
     }
 
     fn handle_framerate_tick(actors: Arc<Mutex<ActorRegistry>>, actor_name: String, tick: f64) {
@@ -586,15 +591,18 @@ fn run_server(
         })
         .expect("Thread spawning failed");
 
+    let mut next_id = StreamId(0);
     while let Ok(msg) = receiver.recv() {
         debug!("{:?}", msg);
         match msg {
             DevtoolsControlMsg::FromChrome(ChromeToDevtoolsControlMsg::AddClient(stream)) => {
                 let actors = actors.clone();
+                let id = next_id;
+                next_id = StreamId(id.0 + 1);
                 accepted_connections.push(stream.try_clone().unwrap());
                 thread::Builder::new()
                     .name("DevtoolsClientHandler".to_owned())
-                    .spawn(move || handle_client(actors, stream.try_clone().unwrap()))
+                    .spawn(move || handle_client(actors, stream.try_clone().unwrap(), id))
                     .expect("Thread spawning failed");
             },
             DevtoolsControlMsg::FromScript(ScriptToDevtoolsControlMsg::FramerateTick(