Flush and close `swap_setup` substreams

When swapping on testnet we ran into a problem where the CLI started the swap after sending all messages successfully, but the ASB ran into a `connection closed` error at the end of the `swap_setup` and the swap state machine was never actually triggered.

Flushing and closing the stream on both sides should ensure that we don't run into this problem and both parties gracefully exit the protocol.

swap/src/network/swap_setup/alice.rs

@@ -8,7 +8,7 @@ use crate::protocol::{Message0, Message2, Message4};
 use crate::{asb, bitcoin, env, monero};
 use anyhow::{anyhow, Context, Result};
 use futures::future::{BoxFuture, OptionFuture};
-use futures::FutureExt;
+use futures::{AsyncWriteExt, FutureExt};
 use libp2p::core::connection::ConnectionId;
 use libp2p::core::upgrade;
 use libp2p::swarm::{
@@ -383,6 +383,15 @@ where
                 .receive(message4)
                 .context("Failed to transition state2 -> state3 using message4")?;
 
+            substream
+                .flush()
+                .await
+                .context("Failed to flush substream after all messages were sent")?;
+            substream
+                .close()
+                .await
+                .context("Failed to close substream after all messages were sent")?;
+
             Ok((swap_id, state3))
         });
 

swap/src/network/swap_setup/bob.rs

@@ -7,7 +7,7 @@ use crate::protocol::{Message1, Message3};
 use crate::{bitcoin, cli, env, monero};
 use anyhow::Result;
 use futures::future::{BoxFuture, OptionFuture};
-use futures::FutureExt;
+use futures::{AsyncWriteExt, FutureExt};
 use libp2p::core::connection::ConnectionId;
 use libp2p::core::upgrade;
 use libp2p::swarm::{
@@ -188,6 +188,9 @@ impl ProtocolsHandler for Handler {
 
             write_cbor_message(&mut substream, state2.next_message()).await?;
 
+            substream.flush().await?;
+            substream.close().await?;
+
             Ok(state2)
         });