fix(pubsub): handle subscription response on reconnects (#105) (#107)

crates/pubsub/src/managers/sub.rs

@@ -47,7 +47,7 @@ impl SubscriptionManager {
 
         // If we already know a subscription with the exact params,
         // we can just update the server_id and get a new listener.
-        if self.local_to_server.contains_left(&local_id) {
+        if self.local_to_sub.contains_left(&local_id) {
             self.change_server_id(local_id, server_id);
             self.get_rx(local_id).expect("checked existence")
         } else {

crates/pubsub/src/service.rs

@@ -91,13 +91,19 @@ where
 
         // Drop all server IDs. We'll re-insert them as we get responses.
         self.subs.drop_server_ids();
+
         // Dispatch all subscription requests
-        self.subs
-            .iter()
-            .map(|(_, sub)| sub.request().serialized().to_owned())
-            .collect::<Vec<_>>()
-            .into_iter()
-            .try_for_each(|brv| self.dispatch_request(brv))?;
+        self.subs.iter().try_for_each(|(_, sub)| {
+            let req = sub.request().to_owned();
+            let (in_flight, _) = InFlight::new(req.clone());
+            self.in_flights.insert(in_flight);
+
+            self.handle
+                .to_socket
+                .send(req.serialized().to_owned())
+                .map(drop)
+                .map_err(|_| TransportErrorKind::backend_gone())
+        })?;
 
         Ok(())
     }