send syn after 500ms if TFO is enabled for server

- ref shadowsocks#232

crates/shadowsocks-service/src/server/tcprelay.rs

@@ -1,6 +1,7 @@
 //! Shadowsocks TCP server
 
 use std::{
+    future::Future,
     io::{self, ErrorKind},
     net::SocketAddr,
     sync::Arc,
@@ -22,7 +23,11 @@ use shadowsocks::{
     ProxyListener,
     ServerConfig,
 };
-use tokio::{net::TcpStream as TokioTcpStream, time};
+use tokio::{
+    io::{AsyncReadExt, AsyncWriteExt},
+    net::TcpStream as TokioTcpStream,
+    time,
+};
 
 use crate::net::{utils::ignore_until_end, MonProxyStream};
 
@@ -77,6 +82,20 @@ impl TcpServer {
     }
 }
 
+#[inline]
+async fn timeout_fut<F, R>(duration: Option<Duration>, f: F) -> io::Result<R>
+where
+    F: Future<Output = io::Result<R>>,
+{
+    match duration {
+        None => f.await,
+        Some(d) => match time::timeout(d, f).await {
+            Ok(o) => o,
+            Err(..) => Err(ErrorKind::TimedOut.into()),
+        },
+    }
+}
+
 struct TcpServerClient {
     context: Arc<ServiceContext>,
     method: CipherKind,
@@ -123,38 +142,56 @@ impl TcpServerClient {
             return Ok(());
         }
 
-        let mut remote_stream = match self.timeout {
-            Some(d) => {
-                match time::timeout(
-                    d,
-                    OutboundTcpStream::connect_remote_with_opts(
-                        self.context.context_ref(),
-                        &target_addr,
-                        self.context.connect_opts_ref(),
-                    ),
-                )
-                .await
-                {
-                    Ok(Ok(s)) => s,
-                    Ok(Err(e)) => return Err(e),
-                    Err(..) => {
-                        return Err(io::Error::new(
-                            ErrorKind::TimedOut,
-                            format!("connect {} timeout", target_addr),
-                        ))
-                    }
-                }
-            }
-            None => {
-                OutboundTcpStream::connect_remote_with_opts(
-                    self.context.context_ref(),
-                    &target_addr,
-                    self.context.connect_opts_ref(),
-                )
-                .await?
+        let mut remote_stream = match timeout_fut(
+            self.timeout,
+            OutboundTcpStream::connect_remote_with_opts(
+                self.context.context_ref(),
+                &target_addr,
+                self.context.connect_opts_ref(),
+            ),
+        )
+        .await
+        {
+            Ok(s) => s,
+            Err(err) => {
+                error!(
+                    "tcp tunnel {} -> {} connect failed, error: {}",
+                    self.peer_addr, target_addr, err
+                );
+                return Err(err);
             }
         };
 
+        // https://github.com/shadowsocks/shadowsocks-rust/issues/232
+        //
+        // Protocols like FTP, clients will wait for servers to send Welcome Message without sending anything.
+        //
+        // Wait at most 500ms, and then sends handshake packet to remote servers.
+        if self.context.connect_opts_ref().tcp.fastopen {
+            let mut buffer = [0u8; 8192];
+            match time::timeout(Duration::from_millis(500), self.stream.read(&mut buffer)).await {
+                Ok(Ok(0)) => {
+                    // EOF. Just terminate right here.
+                    return Ok(());
+                }
+                Ok(Ok(n)) => {
+                    // Send the first packet.
+                    timeout_fut(self.timeout, remote_stream.write_all(&buffer[..n])).await?;
+                }
+                Ok(Err(err)) => return Err(err),
+                Err(..) => {
+                    // Timeout. Send handshake to server.
+                    timeout_fut(self.timeout, remote_stream.write(&[])).await?;
+
+                    trace!(
+                        "tcp tunnel {} -> {} sent TFO connect without data",
+                        self.peer_addr,
+                        target_addr
+                    );
+                }
+            }
+        }
+
         let (mut lr, mut lw) = self.stream.into_split();
         let (mut rr, mut rw) = remote_stream.split();