call sendData with an empty bytestring

cabal.project

@@ -2,5 +2,5 @@ packages: .
 
 source-repository-package
     type: git
-    location: git://github.com/simplex-chat/aeson.git
+    location: https://github.com/simplex-chat/aeson.git
     tag: 3eb66f9a68f103b5f1489382aad89f5712a64db7

package.yaml

@@ -45,7 +45,7 @@ dependencies:
   - iso8601-time == 0.1.*
   - memory == 0.15.*
   - mtl == 2.2.*
-  - network == 3.1.*
+  - network == 3.1.2.*
   - network-transport == 0.5.*
   - QuickCheck == 2.14.*
   - random >= 1.1 && < 1.3

simplexmq.cabal

@@ -57,6 +57,7 @@ library
       Simplex.Messaging.Server.StoreLog
       Simplex.Messaging.Transport
       Simplex.Messaging.Transport.Client
+      Simplex.Messaging.Transport.KeepAlive
       Simplex.Messaging.Transport.Server
       Simplex.Messaging.Transport.WebSockets
       Simplex.Messaging.Util
@@ -91,7 +92,7 @@ library
     , iso8601-time ==0.1.*
     , memory ==0.15.*
     , mtl ==2.2.*
-    , network ==3.1.*
+    , network ==3.1.2.*
     , network-transport ==0.5.*
     , random >=1.1 && <1.3
     , simple-logger ==0.1.*
@@ -142,7 +143,7 @@ executable smp-agent
     , iso8601-time ==0.1.*
     , memory ==0.15.*
     , mtl ==2.2.*
-    , network ==3.1.*
+    , network ==3.1.2.*
     , network-transport ==0.5.*
     , random >=1.1 && <1.3
     , simple-logger ==0.1.*
@@ -195,7 +196,7 @@ executable smp-server
     , iso8601-time ==0.1.*
     , memory ==0.15.*
     , mtl ==2.2.*
-    , network ==3.1.*
+    , network ==3.1.2.*
     , network-transport ==0.5.*
     , optparse-applicative >=0.15 && <0.17
     , process ==1.6.*
@@ -264,7 +265,7 @@ test-suite smp-server-test
     , iso8601-time ==0.1.*
     , memory ==0.15.*
     , mtl ==2.2.*
-    , network ==3.1.*
+    , network ==3.1.2.*
     , network-transport ==0.5.*
     , random >=1.1 && <1.3
     , simple-logger ==0.1.*

src/Simplex/Messaging/Client.hs

@@ -110,7 +110,7 @@ smpDefaultConfig =
     { qSize = 64,
       defaultTransport = ("5223", transport @TLS),
       tcpTimeout = 4_000_000,
-      smpPing = 30_000_000
+      smpPing = 3_600_000_000 -- 1 hour
     }
 
 data Request = Request

src/Simplex/Messaging/Transport.hs

@@ -80,6 +80,7 @@ import qualified Network.TLS.Extra as TE
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Parsers (dropPrefix, parse, parseRead1, sumTypeJSON)
+import Simplex.Messaging.Transport.KeepAlive
 import Simplex.Messaging.Util (bshow)
 import Simplex.Messaging.Version
 import Test.QuickCheck (Arbitrary (..))
@@ -147,7 +148,8 @@ data TLS = TLS
     tlsPeer :: TransportPeer,
     tlsUniq :: ByteString,
     buffer :: TVar ByteString,
-    getLock :: TMVar ()
+    getLock :: TMVar (),
+    keepAlive :: Maybe KeepAliveThread
   }
 
 connectTLS :: T.TLSParams p => p -> Socket -> IO T.Context
@@ -163,7 +165,7 @@ getTLS tlsPeer cxt = withTlsUnique tlsPeer cxt newTLS
     newTLS tlsUniq = do
       buffer <- newTVarIO ""
       getLock <- newTMVarIO ()
-      pure TLS {tlsContext = cxt, tlsPeer, tlsUniq, buffer, getLock}
+      pure TLS {tlsContext = cxt, tlsPeer, tlsUniq, buffer, getLock, keepAlive = Nothing}
 
 withTlsUnique :: TransportPeer -> T.Context -> (ByteString -> IO c) -> IO c
 withTlsUnique peer cxt f =
@@ -195,12 +197,17 @@ instance Transport TLS where
   transportName _ = "TLS"
   transportPeer = tlsPeer
   getServerConnection = getTLS TServer
-  getClientConnection = getTLS TClient
+  getClientConnection cxt = do
+    tls <- getTLS TClient cxt
+    keepAlive <- Just <$> startKeepAlive (tlsContext tls)
+    pure tls {keepAlive}
   tlsUnique = tlsUniq
-  closeConnection tls = closeTLS $ tlsContext tls
+  closeConnection TLS {tlsContext, keepAlive} = do
+    mapM_ stopKeepAlive keepAlive 
+    closeTLS tlsContext
 
   cGet :: TLS -> Int -> IO ByteString
-  cGet TLS {tlsContext, buffer, getLock} n =
+  cGet TLS {tlsContext, buffer, getLock, keepAlive} n =
     E.bracket_
       (atomically $ takeTMVar getLock)
       (atomically $ putTMVar getLock ())
@@ -213,16 +220,21 @@ instance Transport TLS where
       readChunks :: ByteString -> IO ByteString
       readChunks b
         | B.length b >= n = pure b
-        | otherwise = readChunks . (b <>) =<< T.recvData tlsContext `E.catch` handleEOF
+        | otherwise = do
+          chunk <- T.recvData tlsContext `E.catch` handleEOF
+          mapM_ touchKeepAlive keepAlive
+          readChunks $ b <> chunk
       handleEOF = \case
         T.Error_EOF -> E.throwIO TEBadBlock
         e -> E.throwIO e
 
   cPut :: TLS -> ByteString -> IO ()
-  cPut tls = T.sendData (tlsContext tls) . BL.fromStrict
+  cPut TLS {tlsContext, keepAlive} s = do
+    mapM_ touchKeepAlive keepAlive
+    T.sendData tlsContext $ BL.fromStrict s
 
   getLn :: TLS -> IO ByteString
-  getLn TLS {tlsContext, buffer, getLock} = do
+  getLn TLS {tlsContext, buffer, getLock, keepAlive} = do
     E.bracket_
       (atomically $ takeTMVar getLock)
       (atomically $ putTMVar getLock ())
@@ -236,9 +248,9 @@ instance Transport TLS where
       readChunks b
         | B.elem '\n' b = pure b
         | otherwise = readChunks . (b <>) =<< T.recvData tlsContext `E.catch` handleEOF
-      handleEOF = \case
+      handleEOF e = mapM_ stopKeepAlive keepAlive >> case e of
         T.Error_EOF -> E.throwIO TEBadBlock
-        e -> E.throwIO e
+        _ -> E.throwIO e
 
 -- | Trim trailing CR from ByteString.
 trimCR :: ByteString -> ByteString

src/Simplex/Messaging/Transport/Client.hs

@@ -20,6 +20,7 @@ import Network.Socket
 import qualified Network.TLS as T
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Transport
+import Simplex.Messaging.Transport.KeepAlive
 import System.IO.Error
 import UnliftIO.Exception (IOException)
 import qualified UnliftIO.Exception as E
@@ -51,6 +52,7 @@ startTCPClient host port clientParams = withSocketsDo $ resolve >>= tryOpen err
     open addr = do
       sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)
       connect sock $ addrAddress addr
+      setSocketKeepAlive sock defaultKeepAlive
       ctx <- connectTLS clientParams sock
       getClientConnection ctx
 

src/Simplex/Messaging/Transport/KeepAlive.hs

@@ -0,0 +1,72 @@
+{-# LANGUAGE CApiFFI #-}
+{-# LANGUAGE CPP #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+
+module Simplex.Messaging.Transport.KeepAlive where
+
+import Control.Concurrent
+import Control.Concurrent.STM
+import Control.Monad
+import Data.Time.Clock.System (SystemTime (..), getSystemTime)
+import Foreign.C (CInt (..))
+import Network.Socket
+import qualified Network.TLS as T
+
+foreign import capi "netinet/tcp.h value TCP_KEEPCNT" tcpKeepCnt :: CInt
+
+foreign import capi "netinet/tcp.h value TCP_KEEPINTVL" tcpKeepIntvl :: CInt
+
+#if defined(darwin_HOST_OS)
+foreign import capi "netinet/tcp.h value TCP_KEEPALIVE" tcpKeepIdle :: CInt
+foreign import capi "netinet/in.h value IPPROTO_TCP" solTcp :: CInt
+#else
+foreign import capi "netinet/tcp.h value TCP_KEEPIDLE" tcpKeepIdle :: CInt
+foreign import capi "netinet/tcp.h value SOL_TCP" solTcp :: CInt
+#endif
+
+data KeepAliveOpts = KeepAliveOpts
+  { keepCnt :: Int,
+    keepIdle :: Int,
+    keepIntvl :: Int
+  }
+
+defaultKeepAlive :: KeepAliveOpts
+defaultKeepAlive =
+  KeepAliveOpts
+    { keepCnt = 4,
+      keepIdle = 60,
+      keepIntvl = 30
+    }
+
+setSocketKeepAlive :: Socket -> KeepAliveOpts -> IO ()
+setSocketKeepAlive sock KeepAliveOpts {keepCnt, keepIdle, keepIntvl} = do
+  setSocketOption sock KeepAlive 1
+  setSocketOption sock (SockOpt solTcp tcpKeepCnt) keepCnt
+  setSocketOption sock (SockOpt solTcp tcpKeepIdle) keepIdle
+  setSocketOption sock (SockOpt solTcp tcpKeepIntvl) keepIntvl
+
+data KeepAliveThread = KeepAliveThread
+  { threadId :: ThreadId,
+    dataTs :: TVar SystemTime
+  }
+
+startKeepAlive :: T.Context -> IO KeepAliveThread
+startKeepAlive cxt = do
+  dataTs <- newTVarIO =<< getSystemTime
+  threadId <- forkIO . forever $ do
+    threadDelay 30000000
+    ts' <- getSystemTime
+    doPing <- atomically $ do
+      ts <- readTVar dataTs
+      let ping = systemSeconds ts' - systemSeconds ts >= 30
+      when ping $ writeTVar dataTs ts'
+      pure ping
+    when doPing $ putStrLn "*** ping ***" >> T.sendData cxt ""
+  pure KeepAliveThread {threadId, dataTs}
+
+touchKeepAlive :: KeepAliveThread -> IO ()
+touchKeepAlive KeepAliveThread {dataTs} = atomically . writeTVar dataTs =<< getSystemTime
+
+stopKeepAlive :: KeepAliveThread -> IO ()
+stopKeepAlive KeepAliveThread {threadId} = killThread threadId

stack.yaml

@@ -36,6 +36,7 @@ packages:
 #
 extra-deps:
   - cryptostore-0.2.1.0@sha256:9896e2984f36a1c8790f057fd5ce3da4cbcaf8aa73eb2d9277916886978c5b19,3881
+  - network-3.1.2.7@sha256:e3d78b13db9512aeb106e44a334ab42b7aa48d26c097299084084cb8be5c5568,4888
   - simple-logger-0.1.0@sha256:be8ede4bd251a9cac776533bae7fb643369ebd826eb948a9a18df1a8dd252ff8,1079
   - tls-1.5.7@sha256:1cc30253a9696b65a9cafc0317fbf09f7dcea15e3a145ed6c9c0e28c632fa23a,6991
 # below dependancies are to update Aeson to 2.0.3