chore(deps): update rust crate interprocess to v2 (#1915)

Cargo.toml

@@ -92,7 +92,7 @@ humantime = "2.1.0"
 http = "1.1.0"
 httpmock = "0.7"
 hyper = "1.0"
-interprocess = { version = "1", default-features = false }
+interprocess = { version = "2", default-features = false }
 indoc = "2"
 lazycell = "1"
 lazy_static = "1.4"

src/command/dev/do_dev.rs

@@ -1,16 +1,16 @@
 use anyhow::{anyhow, Context};
 use camino::Utf8PathBuf;
-use rover_std::Emoji;
+use crossbeam_channel::bounded as sync_channel;
 
-use super::protocol::{FollowerChannel, FollowerMessenger, LeaderChannel, LeaderSession};
-use super::router::RouterConfigHandler;
-use super::Dev;
+use rover_std::Emoji;
 
 use crate::command::dev::protocol::FollowerMessage;
 use crate::utils::client::StudioClientConfig;
 use crate::{RoverError, RoverOutput, RoverResult};
 
-use crossbeam_channel::bounded as sync_channel;
+use super::protocol::{FollowerChannel, FollowerMessenger, LeaderChannel, LeaderSession};
+use super::router::RouterConfigHandler;
+use super::Dev;
 
 pub fn log_err_and_continue(err: RoverError) -> RoverError {
     let _ = err.print();
@@ -29,7 +29,7 @@ impl Dev {
 
         let router_config_handler = RouterConfigHandler::try_from(&self.opts.supergraph_opts)?;
         let router_address = router_config_handler.get_router_address();
-        let ipc_socket_addr = router_config_handler.get_ipc_address()?;
+        let raw_socket_name = router_config_handler.get_raw_socket_name();
         let leader_channel = LeaderChannel::new();
         let follower_channel = FollowerChannel::new();
 
@@ -116,14 +116,14 @@ impl Dev {
                 .join()
                 .expect("could not wait for subgraph watcher thread");
         } else {
-            let follower_messenger = FollowerMessenger::from_attached_session(&ipc_socket_addr);
+            let follower_messenger = FollowerMessenger::from_attached_session(&raw_socket_name);
             let mut subgraph_refresher = self.opts.subgraph_opts.get_subgraph_watcher(
                 router_address,
                 &client_config,
                 follower_messenger.clone(),
             )?;
             tracing::info!(
-                "connecting to existing `rover dev` process by communicating via the interprocess socket located at {ipc_socket_addr}"
+                "connecting to existing `rover dev` process by communicating via the interprocess socket located at {raw_socket_name}",
             );
 
             // start the interprocess socket health check in the background

src/command/dev/protocol/follower/messenger.rs

@@ -1,14 +1,15 @@
+use std::{fmt::Debug, io::BufReader, time::Duration};
+
 use anyhow::anyhow;
 use apollo_federation_types::build::SubgraphDefinition;
 use crossbeam_channel::{Receiver, Sender};
-use interprocess::local_socket::LocalSocketStream;
-use std::{fmt::Debug, io::BufReader, time::Duration};
-
-use crate::{RoverError, RoverErrorSuggestion, RoverResult, PKG_VERSION};
+use interprocess::local_socket::traits::Stream;
 
 use crate::command::dev::protocol::{
-    socket_read, socket_write, FollowerMessage, LeaderMessageKind, SubgraphKeys, SubgraphName,
+    create_socket_name, socket_read, socket_write, FollowerMessage, LeaderMessageKind,
+    SubgraphKeys, SubgraphName,
 };
+use crate::{RoverError, RoverErrorSuggestion, RoverResult, PKG_VERSION};
 
 #[derive(Clone, Debug)]
 pub struct FollowerMessenger {
@@ -30,9 +31,9 @@ impl FollowerMessenger {
     }
 
     /// Create a [`FollowerMessenger`] for an attached session that can talk to the main session via a socket.
-    pub fn from_attached_session(ipc_socket_addr: &str) -> Self {
+    pub fn from_attached_session(raw_socket_name: &str) -> Self {
         Self {
-            kind: FollowerMessengerKind::from_attached_session(ipc_socket_addr.to_string()),
+            kind: FollowerMessengerKind::from_attached_session(raw_socket_name.to_string()),
         }
     }
 
@@ -108,7 +109,7 @@ enum FollowerMessengerKind {
         leader_message_receiver: Receiver<LeaderMessageKind>,
     },
     FromAttachedSession {
-        ipc_socket_addr: String,
+        raw_socket_name: String,
     },
 }
 
@@ -123,8 +124,8 @@ impl FollowerMessengerKind {
         }
     }
 
-    fn from_attached_session(ipc_socket_addr: String) -> Self {
-        Self::FromAttachedSession { ipc_socket_addr }
+    fn from_attached_session(raw_socket_name: String) -> Self {
+        Self::FromAttachedSession { raw_socket_name }
     }
 
     fn message_leader(
@@ -149,8 +150,9 @@ impl FollowerMessengerKind {
 
                 leader_message
             }
-            FromAttachedSession { ipc_socket_addr } => {
-                let stream = LocalSocketStream::connect(&**ipc_socket_addr).map_err(|_| {
+            FromAttachedSession { raw_socket_name } => {
+                let socket_name = create_socket_name(raw_socket_name)?;
+                let stream = Stream::connect(socket_name).map_err(|_| {
                     let mut err = RoverError::new(anyhow!(
                         "there is not a main `rover dev` process to report updates to"
                     ));

src/command/dev/protocol/leader.rs

@@ -1,34 +1,38 @@
-use crate::{
-    command::dev::{
-        compose::ComposeRunner,
-        do_dev::log_err_and_continue,
-        router::{RouterConfigHandler, RouterRunner},
-        OVERRIDE_DEV_COMPOSITION_VERSION,
-    },
-    options::PluginOpts,
-    utils::client::StudioClientConfig,
-    RoverError, RoverErrorSuggestion, RoverResult, PKG_VERSION,
+use std::{
+    collections::{hash_map::Entry::Vacant, HashMap},
+    fmt::Debug,
+    io::BufReader,
+    net::TcpListener,
 };
+
 use anyhow::{anyhow, Context};
 use apollo_federation_types::{
     build::SubgraphDefinition,
     config::{FederationVersion, SupergraphConfig},
 };
 use camino::Utf8PathBuf;
 use crossbeam_channel::{bounded, Receiver, Sender};
-use interprocess::local_socket::{LocalSocketListener, LocalSocketStream};
-use rover_std::Emoji;
+use interprocess::local_socket::traits::{ListenerExt, Stream};
+use interprocess::local_socket::ListenerOptions;
 use semver::Version;
 use serde::{Deserialize, Serialize};
 
-use std::{
-    collections::{hash_map::Entry::Vacant, HashMap},
-    fmt::Debug,
-    io::BufReader,
-    net::TcpListener,
+use rover_std::Emoji;
+
+use crate::{
+    command::dev::{
+        compose::ComposeRunner,
+        do_dev::log_err_and_continue,
+        router::{RouterConfigHandler, RouterRunner},
+        OVERRIDE_DEV_COMPOSITION_VERSION,
+    },
+    options::PluginOpts,
+    utils::client::StudioClientConfig,
+    RoverError, RoverErrorSuggestion, RoverResult, PKG_VERSION,
 };
 
 use super::{
+    create_socket_name,
     socket::{handle_socket_error, socket_read, socket_write},
     types::{
         CompositionResult, SubgraphEntry, SubgraphKey, SubgraphKeys, SubgraphName, SubgraphSdl,
@@ -39,7 +43,7 @@ use super::{
 #[derive(Debug)]
 pub struct LeaderSession {
     subgraphs: HashMap<SubgraphKey, SubgraphSdl>,
-    ipc_socket_addr: String,
+    raw_socket_name: String,
     compose_runner: ComposeRunner,
     router_runner: RouterRunner,
     follower_channel: FollowerChannel,
@@ -63,10 +67,12 @@ impl LeaderSession {
         plugin_opts: PluginOpts,
         router_config_handler: RouterConfigHandler,
     ) -> RoverResult<Option<Self>> {
-        let ipc_socket_addr = router_config_handler.get_ipc_address()?;
+        let raw_socket_name = router_config_handler.get_raw_socket_name();
         let router_socket_addr = router_config_handler.get_router_address();
-        if let Ok(stream) = LocalSocketStream::connect(&*ipc_socket_addr) {
-            // write to the socket so we don't make the other session deadlock waiting on a message
+        let socket_name = create_socket_name(&raw_socket_name)?;
+
+        if let Ok(stream) = Stream::connect(socket_name.clone()) {
+            // write to the socket, so we don't make the other session deadlock waiting on a message
             let mut stream = BufReader::new(stream);
             socket_write(&FollowerMessage::health_check(false)?, &mut stream)?;
             let _ = LeaderSession::socket_read(&mut stream);
@@ -80,7 +86,7 @@ impl LeaderSession {
         //
         // remove the socket file before starting in case it was here from last time
         // if we can't connect to it, it's safe to remove
-        let _ = std::fs::remove_file(&ipc_socket_addr);
+        let _ = std::fs::remove_file(&raw_socket_name);
 
         if TcpListener::bind(router_socket_addr).is_err() {
             let mut err =
@@ -126,7 +132,7 @@ impl LeaderSession {
 
         Ok(Some(Self {
             subgraphs: HashMap::new(),
-            ipc_socket_addr,
+            raw_socket_name,
             compose_runner,
             router_runner,
             follower_channel,
@@ -165,15 +171,19 @@ impl LeaderSession {
 
     /// Listen on the socket for incoming [`FollowerMessageKind`] messages.
     fn receive_messages_from_attached_sessions(&self) -> RoverResult<()> {
-        let listener = LocalSocketListener::bind(&*self.ipc_socket_addr).with_context(|| {
-            format!(
-                "could not start local socket server at {}",
-                &self.ipc_socket_addr
-            )
-        })?;
+        let socket_name = create_socket_name(&self.raw_socket_name)?;
+        let listener = ListenerOptions::new()
+            .name(socket_name)
+            .create_sync()
+            .with_context(|| {
+                format!(
+                    "could not start local socket server at {:?}",
+                    &self.raw_socket_name
+                )
+            })?;
         tracing::info!(
             "connected to socket {}, waiting for messages",
-            &self.ipc_socket_addr
+            &self.raw_socket_name
         );
 
         let follower_message_sender = self.follower_channel.sender.clone();
@@ -307,7 +317,9 @@ impl LeaderSession {
     }
 
     /// Reads a [`FollowerMessage`] from an open socket connection.
-    fn socket_read(stream: &mut BufReader<LocalSocketStream>) -> RoverResult<FollowerMessage> {
+    fn socket_read(
+        stream: &mut BufReader<interprocess::local_socket::Stream>,
+    ) -> RoverResult<FollowerMessage> {
         socket_read(stream)
             .map(|message| {
                 tracing::debug!("leader received message {:?}", &message);
@@ -322,7 +334,7 @@ impl LeaderSession {
     /// Writes a [`LeaderMessageKind`] to an open socket connection.
     fn socket_write(
         message: LeaderMessageKind,
-        stream: &mut BufReader<LocalSocketStream>,
+        stream: &mut BufReader<interprocess::local_socket::Stream>,
     ) -> RoverResult<()> {
         tracing::debug!("leader sending message {:?}", message);
         socket_write(&message, stream)
@@ -350,7 +362,7 @@ impl LeaderSession {
     /// Shuts the router down, removes the socket file, and exits the process.
     pub fn shutdown(&mut self) {
         let _ = self.router_runner.kill().map_err(log_err_and_continue);
-        let _ = std::fs::remove_file(&self.ipc_socket_addr);
+        let _ = std::fs::remove_file(&self.raw_socket_name);
         std::process::exit(1)
     }
 

src/command/dev/protocol/mod.rs

@@ -1,9 +1,15 @@
-mod follower;
-mod leader;
-mod socket;
-mod types;
+use interprocess::local_socket::{GenericFilePath, Name, ToFsName};
 
 pub use follower::*;
 pub use leader::*;
 pub(crate) use socket::*;
 pub use types::*;
+
+mod follower;
+mod leader;
+mod socket;
+mod types;
+
+pub(crate) fn create_socket_name(raw_socket_name: &str) -> std::io::Result<Name> {
+    raw_socket_name.to_fs_name::<GenericFilePath>()
+}