From d700525959f5a247911195b9f7e43f1eb4c0bdf6 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Marcel=20M=C3=BCller?= <com2040@gmail.com>
Date: Sun, 8 Mar 2020 12:57:43 +0100
Subject: [PATCH] Add msg queuing for all messages for public agents (#1007)

* Add msg queuing for all messages for public agents

Fixes #944

* Send Disconnected and Destroy correctly

They now use the `send_message` so that they also
get queued in case the worker is not ready yet

* Remove todo about connected message
---
 src/agent.rs | 24 +++++++++++++++---------
 1 file changed, 15 insertions(+), 9 deletions(-)

diff --git a/src/agent.rs b/src/agent.rs
index ce9a9bb8574..18839671936 100644
--- a/src/agent.rs
+++ b/src/agent.rs
@@ -594,7 +594,6 @@ impl Discoverer for Public {
                             let msg = FromWorker::<AGN::Output>::unpack(&data);
                             match msg {
                                 FromWorker::WorkerLoaded => {
-                                    // TODO(#944): Send `Connected` message
                                     REMOTE_AGENTS_LOADED.with(|loaded| {
                                         let _ = loaded.borrow_mut().insert(TypeId::of::<AGN>());
                                     });
@@ -640,7 +639,9 @@ impl Discoverer for Public {
                         }),
                     };
                     let launched = RemoteAgent::new(worker, slab);
-                    entry.insert(launched).create_bridge(callback)
+                    let bridge = entry.insert(launched).create_bridge(callback);
+                    bridge.send_message(ToWorker::Connected(bridge.id));
+                    bridge
                 }
             }
         });
@@ -684,6 +685,15 @@ impl<AGN: Agent> PublicBridge<AGN> {
             }
         });
     }
+
+    /// Send a message to the worker, queuing it up if necessary
+    fn send_message(&self, msg: ToWorker<AGN::Input>) {
+        if self.worker_is_loaded() {
+            send_to_remote::<AGN>(&self.worker, msg);
+        } else {
+            self.msg_to_queue(msg.pack());
+        }
+    }
 }
 
 fn send_to_remote<AGN: Agent>(
@@ -705,11 +715,7 @@ fn send_to_remote<AGN: Agent>(
 impl<AGN: Agent> Bridge<AGN> for PublicBridge<AGN> {
     fn send(&mut self, msg: AGN::Input) {
         let msg = ToWorker::ProcessInput(self.id, msg);
-        if self.worker_is_loaded() {
-            send_to_remote::<AGN>(&self.worker, msg);
-        } else {
-            self.msg_to_queue(msg.pack());
-        }
+        self.send_message(msg);
     }
 }
 
@@ -733,11 +739,11 @@ impl<AGN: Agent> Drop for PublicBridge<AGN> {
         });
 
         let disconnected = ToWorker::Disconnected(self.id);
-        send_to_remote::<AGN>(&self.worker, disconnected);
+        self.send_message(disconnected);
 
         if terminate_worker {
             let destroy = ToWorker::Destroy;
-            send_to_remote::<AGN>(&self.worker, destroy);
+            self.send_message(destroy);
 
             REMOTE_AGENTS_LOADED.with(|loaded| {
                 loaded.borrow_mut().remove(&TypeId::of::<AGN>());