websockets transport (#139)

* example websockets server

* example of ws client

* type class TConnection for  generic TCP/WebSockets implementation

* support WebSockets transport

* rename TConnection methods

* revert runClient to not need transport arg

* pass the list of ports and transports via SMP server config

* remove TypeApplications

* s/Transport/TProxy/, s/TConnection/Transport/

* fix server with multiple transports, make SMP client use WS transport with port 80 (TODO fallback to WS)

apps/smp-agent/Main.hs

@@ -1,5 +1,6 @@
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE TypeApplications #-}
 
 module Main where
 
@@ -8,6 +9,7 @@ import qualified Data.List.NonEmpty as L
 import Simplex.Messaging.Agent (runSMPAgent)
 import Simplex.Messaging.Agent.Env.SQLite
 import Simplex.Messaging.Client (smpDefaultConfig)
+import Simplex.Messaging.Transport (TCP, Transport (..))
 
 cfg :: AgentConfig
 cfg =
@@ -28,4 +30,4 @@ main :: IO ()
 main = do
   putStrLn $ "SMP agent listening on port " ++ tcpPort (cfg :: AgentConfig)
   setLogLevel LogInfo -- LogError
-  withGlobalLogging logCfg $ runSMPAgent cfg
+  withGlobalLogging logCfg $ runSMPAgent (transport @TCP) cfg

apps/smp-server/Main.hs

@@ -3,10 +3,11 @@
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE TypeApplications #-}
 
 module Main where
 
-import Control.Monad (unless, when)
+import Control.Monad (forM_, unless, when)
 import qualified Crypto.Store.PKCS8 as S
 import Data.ByteString.Base64 (encode)
 import qualified Data.ByteString.Char8 as B
@@ -20,6 +21,8 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Server (runSMPServer)
 import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Server.StoreLog (StoreLog, openReadStoreLog)
+import Simplex.Messaging.Transport (ATransport (..), TCP, Transport (..))
+import Simplex.Messaging.Transport.WebSockets (WS)
 import System.Directory (createDirectoryIfMissing, doesFileExist)
 import System.Exit (exitFailure)
 import System.FilePath (combine)
@@ -28,7 +31,7 @@ import System.IO (IOMode (..), hFlush, stdout)
 cfg :: ServerConfig
 cfg =
   ServerConfig
-    { tcpPort = "5223",
+    { transports = [("5223", transport @TCP), ("80", transport @WS)],
       tbqSize = 16,
       queueIdBytes = 12,
       msgIdBytes = 6,
@@ -57,7 +60,8 @@ main = do
   storeLog <- openStoreLog ini
   pk <- readCreateKey
   B.putStrLn $ "transport key hash: " <> serverKeyHash pk
-  putStrLn $ "listening on port " <> tcpPort cfg
+  forM_ (transports cfg) $ \(port, ATransport t) ->
+    putStrLn $ "listening on port " <> port <> " (" <> transportName t <> ")"
   runSMPServer cfg {serverPrivateKey = pk, storeLog}
 
 data IniOpts = IniOpts

package.yaml

@@ -51,6 +51,7 @@ dependencies:
   - transformers == 0.5.*
   - unliftio == 0.2.*
   - unliftio-core == 0.1.*
+  - websockets == 0.12.*
   - x509 == 1.7.*
 
 library:

src/Simplex/Messaging/Agent.hs

@@ -51,9 +51,8 @@ import Simplex.Messaging.Client (SMPServerTransmission)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol (CorrId (..), MsgBody, SenderPublicKey)
 import qualified Simplex.Messaging.Protocol as SMP
-import Simplex.Messaging.Transport (putLn, runTCPServer)
+import Simplex.Messaging.Transport (ATransport (..), TProxy, Transport (..), runTransportServer)
 import Simplex.Messaging.Util (bshow)
-import System.IO (Handle)
 import System.Random (randomR)
 import UnliftIO.Async (race_)
 import qualified UnliftIO.Exception as E
@@ -62,18 +61,20 @@ import UnliftIO.STM
 -- | Runs an SMP agent as a TCP service using passed configuration.
 --
 -- See a full agent executable here: https://github.com/simplex-chat/simplexmq/blob/master/apps/smp-agent/Main.hs
-runSMPAgent :: (MonadRandom m, MonadUnliftIO m) => AgentConfig -> m ()
-runSMPAgent cfg = newEmptyTMVarIO >>= (`runSMPAgentBlocking` cfg)
+runSMPAgent :: (MonadRandom m, MonadUnliftIO m) => ATransport -> AgentConfig -> m ()
+runSMPAgent t cfg = do
+  started <- newEmptyTMVarIO
+  runSMPAgentBlocking t started cfg
 
 -- | Runs an SMP agent as a TCP service using passed configuration with signalling.
 --
 -- This function uses passed TMVar to signal when the server is ready to accept TCP requests (True)
 -- and when it is disconnected from the TCP socket once the server thread is killed (False).
-runSMPAgentBlocking :: (MonadRandom m, MonadUnliftIO m) => TMVar Bool -> AgentConfig -> m ()
-runSMPAgentBlocking started cfg@AgentConfig {tcpPort} = runReaderT smpAgent =<< newSMPAgentEnv cfg
+runSMPAgentBlocking :: (MonadRandom m, MonadUnliftIO m) => ATransport -> TMVar Bool -> AgentConfig -> m ()
+runSMPAgentBlocking (ATransport t) started cfg@AgentConfig {tcpPort} = runReaderT (smpAgent t) =<< newSMPAgentEnv cfg
   where
-    smpAgent :: (MonadUnliftIO m', MonadReader Env m') => m' ()
-    smpAgent = runTCPServer started tcpPort $ \h -> do
+    smpAgent :: forall c m'. (Transport c, MonadUnliftIO m', MonadReader Env m') => TProxy c -> m' ()
+    smpAgent _ = runTransportServer started tcpPort $ \(h :: c) -> do
       liftIO $ putLn h "Welcome to SMP v0.3.1 agent"
       c <- getSMPAgentClient
       logConnection c True
@@ -87,7 +88,7 @@ getSMPAgentClient = do
   cfg <- asks config
   atomically $ newAgentClient n cfg
 
-connectClient :: MonadUnliftIO m => Handle -> AgentClient -> m ()
+connectClient :: Transport c => MonadUnliftIO m => c -> AgentClient -> m ()
 connectClient h c = race_ (send h c) (receive h c)
 
 logConnection :: MonadUnliftIO m => AgentClient -> Bool -> m ()
@@ -103,7 +104,7 @@ runSMPAgentClient c = do
   s2 <- connectSQLiteStore db
   race_ (subscriber c s1) (client c s2)
 
-receive :: forall m. MonadUnliftIO m => Handle -> AgentClient -> m ()
+receive :: forall c m. (Transport c, MonadUnliftIO m) => c -> AgentClient -> m ()
 receive h c@AgentClient {rcvQ, sndQ} = forever $ do
   (corrId, cAlias, cmdOrErr) <- tGet SClient h
   case cmdOrErr of
@@ -115,7 +116,7 @@ receive h c@AgentClient {rcvQ, sndQ} = forever $ do
       logClient c "-->" t
       atomically $ writeTBQueue q t
 
-send :: MonadUnliftIO m => Handle -> AgentClient -> m ()
+send :: (Transport c, MonadUnliftIO m) => c -> AgentClient -> m ()
 send h c@AgentClient {sndQ} = forever $ do
   t <- atomically $ readTBQueue sndQ
   tPut h t

src/Simplex/Messaging/Agent/Protocol.hs

@@ -102,9 +102,8 @@ import Simplex.Messaging.Protocol
     SenderPublicKey,
   )
 import qualified Simplex.Messaging.Protocol as SMP
-import Simplex.Messaging.Transport (TransportError, getLn, putLn, serializeTransportError, transportErrorP)
+import Simplex.Messaging.Transport (Transport (..), TransportError, serializeTransportError, transportErrorP)
 import Simplex.Messaging.Util
-import System.IO
 import Test.QuickCheck (Arbitrary (..))
 import Text.Read
 import UnliftIO.Exception
@@ -540,23 +539,23 @@ serializeMsg :: ByteString -> ByteString
 serializeMsg body = bshow (B.length body) <> "\n" <> body
 
 -- | Send raw (unparsed) SMP agent protocol transmission to TCP connection.
-tPutRaw :: Handle -> ARawTransmission -> IO ()
+tPutRaw :: Transport c => c -> ARawTransmission -> IO ()
 tPutRaw h (corrId, connAlias, command) = do
   putLn h corrId
   putLn h connAlias
   putLn h command
 
 -- | Receive raw (unparsed) SMP agent protocol transmission from TCP connection.
-tGetRaw :: Handle -> IO ARawTransmission
+tGetRaw :: Transport c => c -> IO ARawTransmission
 tGetRaw h = (,,) <$> getLn h <*> getLn h <*> getLn h
 
 -- | Send SMP agent protocol command (or response) to TCP connection.
-tPut :: MonadIO m => Handle -> ATransmission p -> m ()
+tPut :: (Transport c, MonadIO m) => c -> ATransmission p -> m ()
 tPut h (CorrId corrId, connAlias, command) =
   liftIO $ tPutRaw h (corrId, connAlias, serializeCommand command)
 
 -- | Receive client and agent transmissions from TCP connection.
-tGet :: forall m p. MonadIO m => SAParty p -> Handle -> m (ATransmissionOrError p)
+tGet :: forall c m p. (Transport c, MonadIO m) => SAParty p -> c -> m (ATransmissionOrError p)
 tGet party h = liftIO (tGetRaw h) >>= tParseLoadBody
   where
     tParseLoadBody :: ARawTransmission -> m (ATransmissionOrError p)
@@ -595,7 +594,7 @@ tGet party h = liftIO (tGetRaw h) >>= tParseLoadBody
         ':' : body -> return . Right $ B.pack body
         str -> case readMaybe str :: Maybe Int of
           Just size -> liftIO $ do
-            body <- B.hGet h size
+            body <- cGet h size
             s <- getLn h
             return $ if B.null s then Right body else Left $ CMD SIZE
           Nothing -> return . Left $ CMD SYNTAX

src/Simplex/Messaging/Client.hs

@@ -7,6 +7,7 @@
 {-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeApplications #-}
 
 -- |
 -- Module      : Simplex.Messaging.Client
@@ -60,9 +61,9 @@ import Numeric.Natural
 import Simplex.Messaging.Agent.Protocol (SMPServer (..))
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol
-import Simplex.Messaging.Transport (THandle (..), TransportError, clientHandshake, runTCPClient)
+import Simplex.Messaging.Transport (ATransport (..), TCP, THandle (..), TProxy, Transport (..), TransportError, clientHandshake, runTransportClient)
+import Simplex.Messaging.Transport.WebSockets (WS)
 import Simplex.Messaging.Util (bshow, liftError, raceAny_)
-import System.IO
 import System.Timeout
 
 -- | 'SMPClient' is a handle used to send commands to a specific SMP server.
@@ -92,7 +93,7 @@ data SMPClientConfig = SMPClientConfig
   { -- | size of TBQueue to use for server commands and responses
     qSize :: Natural,
     -- | default SMP server port if port is not specified in SMPServer
-    defaultPort :: ServiceName,
+    defaultTransport :: (ServiceName, ATransport),
     -- | timeout of TCP commands (microseconds)
     tcpTimeout :: Int,
     -- | period for SMP ping commands (microseconds)
@@ -107,7 +108,7 @@ smpDefaultConfig :: SMPClientConfig
 smpDefaultConfig =
   SMPClientConfig
     { qSize = 16,
-      defaultPort = "5223",
+      defaultTransport = ("5223", transport @TCP),
       tcpTimeout = 4_000_000,
       smpPing = 30_000_000,
       smpCommandSize = 256
@@ -126,88 +127,90 @@ type Response = Either SMPClientError Cmd
 -- A single queue can be used for multiple 'SMPClient' instances,
 -- as 'SMPServerTransmission' includes server information.
 getSMPClient :: SMPServer -> SMPClientConfig -> TBQueue SMPServerTransmission -> IO () -> IO (Either SMPClientError SMPClient)
-getSMPClient
-  smpServer@SMPServer {host, port, keyHash}
-  SMPClientConfig {qSize, defaultPort, tcpTimeout, smpPing}
-  msgQ
-  disconnected = do
-    c <- atomically mkSMPClient
-    thVar <- newEmptyTMVarIO
-    action <-
-      async $
-        runTCPClient host (fromMaybe defaultPort port) (client c thVar)
-          `finally` atomically (putTMVar thVar $ Left SMPNetworkError)
-    tHandle <- tcpTimeout `timeout` atomically (takeTMVar thVar)
-    pure $ case tHandle of
-      Just (Right THandle {blockSize}) -> Right c {action, blockSize}
-      Just (Left e) -> Left e
-      Nothing -> Left SMPNetworkError
-    where
-      mkSMPClient :: STM SMPClient
-      mkSMPClient = do
-        connected <- newTVar False
-        clientCorrId <- newTVar 0
-        sentCommands <- newTVar M.empty
-        sndQ <- newTBQueue qSize
-        rcvQ <- newTBQueue qSize
-        return
-          SMPClient
-            { action = undefined,
-              blockSize = undefined,
-              connected,
-              smpServer,
-              tcpTimeout,
-              clientCorrId,
-              sentCommands,
-              sndQ,
-              rcvQ,
-              msgQ
-            }
+getSMPClient smpServer cfg@SMPClientConfig {qSize, tcpTimeout, smpPing} msgQ disconnected =
+  atomically mkSMPClient >>= runClient useTransport
+  where
+    mkSMPClient :: STM SMPClient
+    mkSMPClient = do
+      connected <- newTVar False
+      clientCorrId <- newTVar 0
+      sentCommands <- newTVar M.empty
+      sndQ <- newTBQueue qSize
+      rcvQ <- newTBQueue qSize
+      return
+        SMPClient
+          { action = undefined,
+            blockSize = undefined,
+            connected,
+            smpServer,
+            tcpTimeout,
+            clientCorrId,
+            sentCommands,
+            sndQ,
+            rcvQ,
+            msgQ
+          }
 
-      client :: SMPClient -> TMVar (Either SMPClientError THandle) -> Handle -> IO ()
-      client c thVar h =
-        runExceptT (clientHandshake h keyHash) >>= \case
-          Right th -> clientTransport c thVar th
-          Left e -> atomically . putTMVar thVar . Left $ SMPTransportError e
+    runClient :: (ServiceName, ATransport) -> SMPClient -> IO (Either SMPClientError SMPClient)
+    runClient (port', ATransport t) c = do
+      thVar <- newEmptyTMVarIO
+      action <-
+        async $
+          runTransportClient (host smpServer) port' (client t c thVar)
+            `finally` atomically (putTMVar thVar $ Left SMPNetworkError)
+      bSize <- tcpTimeout `timeout` atomically (takeTMVar thVar)
+      pure $ case bSize of
+        Just (Right blockSize) -> Right c {action, blockSize}
+        Just (Left e) -> Left e
+        Nothing -> Left SMPNetworkError
 
-      clientTransport :: SMPClient -> TMVar (Either SMPClientError THandle) -> THandle -> IO ()
-      clientTransport c thVar th = do
-        atomically $ do
-          writeTVar (connected c) True
-          putTMVar thVar $ Right th
-        raceAny_ [send c th, process c, receive c th, ping c]
-          `finally` disconnected
+    useTransport :: (ServiceName, ATransport)
+    useTransport = case port smpServer of
+      Nothing -> defaultTransport cfg
+      Just "80" -> ("80", transport @WS)
+      Just p -> (p, transport @TCP)
 
-      send :: SMPClient -> THandle -> IO ()
-      send SMPClient {sndQ} h = forever $ atomically (readTBQueue sndQ) >>= tPut h
+    client :: forall c. Transport c => TProxy c -> SMPClient -> TMVar (Either SMPClientError Int) -> c -> IO ()
+    client _ c thVar h =
+      runExceptT (clientHandshake h $ keyHash smpServer) >>= \case
+        Left e -> atomically . putTMVar thVar . Left $ SMPTransportError e
+        Right th -> do
+          atomically $ do
+            writeTVar (connected c) True
+            putTMVar thVar . Right $ blockSize (th :: THandle c)
+          raceAny_ [send c th, process c, receive c th, ping c]
+            `finally` disconnected
 
-      receive :: SMPClient -> THandle -> IO ()
-      receive SMPClient {rcvQ} h = forever $ tGet fromServer h >>= atomically . writeTBQueue rcvQ
+    send :: Transport c => SMPClient -> THandle c -> IO ()
+    send SMPClient {sndQ} h = forever $ atomically (readTBQueue sndQ) >>= tPut h
 
-      ping :: SMPClient -> IO ()
-      ping c = forever $ do
-        threadDelay smpPing
-        runExceptT $ sendSMPCommand c Nothing "" (Cmd SSender PING)
+    receive :: Transport c => SMPClient -> THandle c -> IO ()
+    receive SMPClient {rcvQ} h = forever $ tGet fromServer h >>= atomically . writeTBQueue rcvQ
 
-      process :: SMPClient -> IO ()
-      process SMPClient {rcvQ, sentCommands} = forever $ do
-        (_, (corrId, qId, respOrErr)) <- atomically $ readTBQueue rcvQ
-        cs <- readTVarIO sentCommands
-        case M.lookup corrId cs of
-          Nothing -> do
-            case respOrErr of
-              Right (Cmd SBroker cmd) -> atomically $ writeTBQueue msgQ (smpServer, qId, cmd)
-              -- TODO send everything else to errQ and log in agent
-              _ -> return ()
-          Just Request {queueId, responseVar} -> atomically $ do
-            modifyTVar sentCommands $ M.delete corrId
-            putTMVar responseVar $
-              if queueId == qId
-                then case respOrErr of
-                  Left e -> Left $ SMPResponseError e
-                  Right (Cmd _ (ERR e)) -> Left $ SMPServerError e
-                  Right r -> Right r
-                else Left SMPUnexpectedResponse
+    ping :: SMPClient -> IO ()
+    ping c = forever $ do
+      threadDelay smpPing
+      runExceptT $ sendSMPCommand c Nothing "" (Cmd SSender PING)
+
+    process :: SMPClient -> IO ()
+    process SMPClient {rcvQ, sentCommands} = forever $ do
+      (_, (corrId, qId, respOrErr)) <- atomically $ readTBQueue rcvQ
+      cs <- readTVarIO sentCommands
+      case M.lookup corrId cs of
+        Nothing -> do
+          case respOrErr of
+            Right (Cmd SBroker cmd) -> atomically $ writeTBQueue msgQ (smpServer, qId, cmd)
+            -- TODO send everything else to errQ and log in agent
+            _ -> return ()
+        Just Request {queueId, responseVar} -> atomically $ do
+          modifyTVar sentCommands $ M.delete corrId
+          putTMVar responseVar $
+            if queueId == qId
+              then case respOrErr of
+                Left e -> Left $ SMPResponseError e
+                Right (Cmd _ (ERR e)) -> Left $ SMPServerError e
+                Right r -> Right r
+              else Left SMPUnexpectedResponse
 
 -- | Disconnects SMP client from the server and terminates client threads.
 closeSMPClient :: SMPClient -> IO ()

src/Simplex/Messaging/Protocol.hs

@@ -80,7 +80,7 @@ import GHC.Generics (Generic)
 import Generic.Random (genericArbitraryU)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Parsers
-import Simplex.Messaging.Transport (THandle, TransportError (..), tGetEncrypted, tPutEncrypted)
+import Simplex.Messaging.Transport (THandle, Transport, TransportError (..), tGetEncrypted, tPutEncrypted)
 import Simplex.Messaging.Util
 import Test.QuickCheck (Arbitrary (..))
 
@@ -293,7 +293,7 @@ serializeErrorType :: ErrorType -> ByteString
 serializeErrorType = bshow
 
 -- | Send signed SMP transmission to TCP transport.
-tPut :: THandle -> SignedRawTransmission -> IO (Either TransportError ())
+tPut :: Transport c => THandle c -> SignedRawTransmission -> IO (Either TransportError ())
 tPut th (C.Signature sig, t) =
   tPutEncrypted th $ encode sig <> " " <> t <> " "
 
@@ -315,14 +315,14 @@ fromServer = \case
   _ -> Left $ CMD PROHIBITED
 
 -- | Receive and parse transmission from the TCP transport.
-tGetParse :: THandle -> IO (Either TransportError RawTransmission)
+tGetParse :: Transport c => THandle c -> IO (Either TransportError RawTransmission)
 tGetParse th = (>>= parse transmissionP TEBadBlock) <$> tGetEncrypted th
 
 -- | Receive client and server transmissions.
 --
 -- The first argument is used to limit allowed senders.
 -- 'fromClient' or 'fromServer' should be used here.
-tGet :: forall m. MonadIO m => (Cmd -> Either ErrorType Cmd) -> THandle -> m SignedTransmissionOrError
+tGet :: forall c m. (Transport c, MonadIO m) => (Cmd -> Either ErrorType Cmd) -> THandle c -> m SignedTransmissionOrError
 tGet fromParty th = liftIO (tGetParse th) >>= decodeParseValidate
   where
     decodeParseValidate :: Either TransportError RawTransmission -> m SignedTransmissionOrError

src/Simplex/Messaging/Server.hs

@@ -36,6 +36,7 @@ import qualified Data.ByteString.Char8 as B
 import Data.Functor (($>))
 import qualified Data.Map.Strict as M
 import Data.Time.Clock
+import Network.Socket (ServiceName)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.Env.STM
@@ -46,7 +47,6 @@ import Simplex.Messaging.Server.QueueStore.STM (QueueStore)
 import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Util
-import UnliftIO.Async
 import UnliftIO.Concurrent
 import UnliftIO.Exception
 import UnliftIO.IO
@@ -56,24 +56,29 @@ import UnliftIO.STM
 --
 -- See a full server here: https://github.com/simplex-chat/simplexmq/blob/master/apps/smp-server/Main.hs
 runSMPServer :: (MonadRandom m, MonadUnliftIO m) => ServerConfig -> m ()
-runSMPServer cfg = newEmptyTMVarIO >>= (`runSMPServerBlocking` cfg)
+runSMPServer cfg = do
+  started <- newEmptyTMVarIO
+  runSMPServerBlocking started cfg
 
 -- | Runs an SMP server using passed configuration with signalling.
 --
 -- This function uses passed TMVar to signal when the server is ready to accept TCP requests (True)
 -- and when it is disconnected from the TCP socket once the server thread is killed (False).
 runSMPServerBlocking :: (MonadRandom m, MonadUnliftIO m) => TMVar Bool -> ServerConfig -> m ()
-runSMPServerBlocking started cfg@ServerConfig {tcpPort} = do
+runSMPServerBlocking started cfg@ServerConfig {transports} = do
   env <- newEnv cfg
   runReaderT smpServer env
   where
-    smpServer :: (MonadUnliftIO m, MonadReader Env m) => m ()
+    smpServer :: (MonadUnliftIO m', MonadReader Env m') => m' ()
     smpServer = do
       s <- asks server
-      race_ (runTCPServer started tcpPort runClient) (serverThread s)
+      raceAny_ (serverThread s : map runServer transports)
         `finally` withLog closeStoreLog
 
-    serverThread :: MonadUnliftIO m => Server -> m ()
+    runServer :: (MonadUnliftIO m', MonadReader Env m') => (ServiceName, ATransport) -> m' ()
+    runServer (tcpPort, ATransport t) = runTransportServer started tcpPort (runClient t)
+
+    serverThread :: MonadUnliftIO m' => Server -> m' ()
     serverThread Server {subscribedQ, subscribers} = forever . atomically $ do
       (rId, clnt) <- readTBQueue subscribedQ
       cs <- readTVar subscribers
@@ -82,14 +87,14 @@ runSMPServerBlocking started cfg@ServerConfig {tcpPort} = do
         Nothing -> return ()
       writeTVar subscribers $ M.insert rId clnt cs
 
-runClient :: (MonadUnliftIO m, MonadReader Env m) => Handle -> m ()
-runClient h = do
+runClient :: (Transport c, MonadUnliftIO m, MonadReader Env m) => TProxy c -> c -> m ()
+runClient _ h = do
   keyPair <- asks serverKeyPair
   liftIO (runExceptT $ serverHandshake h keyPair) >>= \case
     Right th -> runClientTransport th
     Left _ -> pure ()
 
-runClientTransport :: (MonadUnliftIO m, MonadReader Env m) => THandle -> m ()
+runClientTransport :: (Transport c, MonadUnliftIO m, MonadReader Env m) => THandle c -> m ()
 runClientTransport th = do
   q <- asks $ tbqSize . config
   c <- atomically $ newClient q
@@ -106,7 +111,7 @@ cancelSub = \case
   Sub {subThread = SubThread t} -> killThread t
   _ -> return ()
 
-receive :: (MonadUnliftIO m, MonadReader Env m) => THandle -> Client -> m ()
+receive :: (Transport c, MonadUnliftIO m, MonadReader Env m) => THandle c -> Client -> m ()
 receive h Client {rcvQ} = forever $ do
   (signature, (corrId, queueId, cmdOrError)) <- tGet fromClient h
   t <- case cmdOrError of
@@ -114,7 +119,7 @@ receive h Client {rcvQ} = forever $ do
     Right cmd -> verifyTransmission (signature, (corrId, queueId, cmd))
   atomically $ writeTBQueue rcvQ t
 
-send :: MonadUnliftIO m => THandle -> Client -> m ()
+send :: (Transport c, MonadUnliftIO m) => THandle c -> Client -> m ()
 send h Client {sndQ} = forever $ do
   t <- atomically $ readTBQueue sndQ
   liftIO $ tPut h ("", serializeTransmission t)

src/Simplex/Messaging/Server/Env/STM.hs

@@ -18,11 +18,12 @@ import Simplex.Messaging.Server.MsgStore.STM
 import Simplex.Messaging.Server.QueueStore (QueueRec (..))
 import Simplex.Messaging.Server.QueueStore.STM
 import Simplex.Messaging.Server.StoreLog
+import Simplex.Messaging.Transport (ATransport)
 import System.IO (IOMode (..))
 import UnliftIO.STM
 
 data ServerConfig = ServerConfig
-  { tcpPort :: ServiceName,
+  { transports :: [(ServiceName, ATransport)],
     tbqSize :: Natural,
     queueIdBytes :: Int,
     msgIdBytes :: Int,

src/Simplex/Messaging/Transport.hs

@@ -9,6 +9,7 @@
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeApplications #-}
 
 -- |
 -- Module      : Simplex.Messaging.Transport
@@ -23,12 +24,17 @@
 --
 -- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#appendix-a
 module Simplex.Messaging.Transport
-  ( -- * TCP transport
-    runTCPServer,
-    runTCPClient,
-    putLn,
-    getLn,
-    trimCR,
+  ( -- * Transport connection class
+    Transport (..),
+    TProxy (..),
+    ATransport (..),
+
+    -- * Transport over TCP
+    runTransportServer,
+    runTransportClient,
+
+    -- * TCP transport
+    TCP (..),
 
     -- * SMP encrypted transport
     THandle (..),
@@ -39,6 +45,9 @@ module Simplex.Messaging.Transport
     tGetEncrypted,
     serializeTransportError,
     transportErrorP,
+
+    -- * Trim trailing CR
+    trimCR,
   )
 where
 
@@ -72,20 +81,53 @@ import Test.QuickCheck (Arbitrary (..))
 import UnliftIO.Concurrent
 import UnliftIO.Exception (Exception, IOException)
 import qualified UnliftIO.Exception as E
-import qualified UnliftIO.IO as IO
 import UnliftIO.STM
 
--- * TCP transport
+-- * Transport connection class
 
--- | Run TCP server on passed port and signal when server started and stopped via passed TMVar.
+class Transport c where
+  transport :: ATransport
+  transport = ATransport (TProxy @c)
+
+  transportName :: TProxy c -> String
+
+  -- | Upgrade client socket to connection (used in the server)
+  getServerConnection :: Socket -> IO c
+
+  -- | Upgrade server socket to connection (used in the client)
+  getClientConnection :: Socket -> IO c
+
+  -- | Close connection
+  closeConnection :: c -> IO ()
+
+  -- | Read fixed number of bytes from connection
+  cGet :: c -> Int -> IO ByteString
+
+  -- | Write bytes to connection
+  cPut :: c -> ByteString -> IO ()
+
+  -- | Receive ByteString from connection, allowing LF or CRLF termination.
+  getLn :: c -> IO ByteString
+
+  -- | Send ByteString to connection terminating it with CRLF.
+  putLn :: c -> ByteString -> IO ()
+  putLn c = cPut c . (<> "\r\n")
+
+data TProxy c = TProxy
+
+data ATransport = forall c. Transport c => ATransport (TProxy c)
+
+-- * Transport over TCP
+
+-- | Run transport server (plain TCP or WebSockets) on passed TCP port and signal when server started and stopped via passed TMVar.
 --
--- All accepted TCP connection handles are passed to the passed function.
-runTCPServer :: MonadUnliftIO m => TMVar Bool -> ServiceName -> (Handle -> m ()) -> m ()
-runTCPServer started port server = do
+-- All accepted connections are passed to the passed function.
+runTransportServer :: (Transport c, MonadUnliftIO m) => TMVar Bool -> ServiceName -> (c -> m ()) -> m ()
+runTransportServer started port server = do
   clients <- newTVarIO S.empty
   E.bracket (liftIO $ startTCPServer started port) (liftIO . closeServer clients) \sock -> forever $ do
-    h <- liftIO $ acceptTCPConn sock
-    tid <- forkFinally (server h) (const $ IO.hClose h)
+    c <- liftIO $ acceptConnection sock
+    tid <- forkFinally (server c) (const $ liftIO $ closeConnection c)
     atomically . modifyTVar clients $ S.insert tid
   where
     closeServer :: TVar (Set ThreadId) -> Socket -> IO ()
@@ -93,6 +135,8 @@ runTCPServer started port server = do
       readTVarIO clients >>= mapM_ killThread
       close sock
       void . atomically $ tryPutTMVar started False
+    acceptConnection :: Transport c => Socket -> IO c
+    acceptConnection sock = accept sock >>= getServerConnection . fst
 
 startTCPServer :: TMVar Bool -> ServiceName -> IO Socket
 startTCPServer started port = withSocketsDo $ resolve >>= open >>= setStarted
@@ -107,18 +151,15 @@ startTCPServer started port = withSocketsDo $ resolve >>= open >>= setStarted
       bind sock $ addrAddress addr
       listen sock 1024
       return sock
-    setStarted sock = atomically (putTMVar started True) >> pure sock
-
-acceptTCPConn :: Socket -> IO Handle
-acceptTCPConn sock = accept sock >>= getSocketHandle . fst
+    setStarted sock = atomically (tryPutTMVar started True) >> pure sock
 
 -- | Connect to passed TCP host:port and pass handle to the client.
-runTCPClient :: MonadUnliftIO m => HostName -> ServiceName -> (Handle -> m a) -> m a
-runTCPClient host port client = do
-  h <- liftIO $ startTCPClient host port
-  client h `E.finally` IO.hClose h
+runTransportClient :: Transport c => MonadUnliftIO m => HostName -> ServiceName -> (c -> m a) -> m a
+runTransportClient host port client = do
+  c <- liftIO $ startTCPClient host port
+  client c `E.finally` liftIO (closeConnection c)
 
-startTCPClient :: HostName -> ServiceName -> IO Handle
+startTCPClient :: forall c. Transport c => HostName -> ServiceName -> IO c
 startTCPClient host port = withSocketsDo $ resolve >>= tryOpen err
   where
     err :: IOException
@@ -129,16 +170,29 @@ startTCPClient host port = withSocketsDo $ resolve >>= tryOpen err
       let hints = defaultHints {addrSocketType = Stream}
        in getAddrInfo (Just hints) (Just host) (Just port)
 
-    tryOpen :: IOException -> [AddrInfo] -> IO Handle
+    tryOpen :: IOException -> [AddrInfo] -> IO c
     tryOpen e [] = E.throwIO e
     tryOpen _ (addr : as) =
       E.try (open addr) >>= either (`tryOpen` as) pure
 
-    open :: AddrInfo -> IO Handle
+    open :: AddrInfo -> IO c
     open addr = do
       sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)
       connect sock $ addrAddress addr
-      getSocketHandle sock
+      getClientConnection sock
+
+-- * TCP transport
+
+newtype TCP = TCP {tcpHandle :: Handle}
+
+instance Transport TCP where
+  transportName _ = "TCP"
+  getServerConnection = fmap TCP . getSocketHandle
+  getClientConnection = getServerConnection
+  closeConnection = hClose . tcpHandle
+  cGet = B.hGet . tcpHandle
+  cPut = B.hPut . tcpHandle
+  getLn = fmap trimCR . B.hGetLine . tcpHandle
 
 getSocketHandle :: Socket -> IO Handle
 getSocketHandle conn = do
@@ -148,14 +202,6 @@ getSocketHandle conn = do
   hSetBuffering h LineBuffering
   return h
 
--- | Send ByteString to TCP connection handle terminating it with CRLF.
-putLn :: Handle -> ByteString -> IO ()
-putLn h = B.hPut h . (<> "\r\n")
-
--- | Receive ByteString from TCP connection handle, allowing LF or CRLF termination.
-getLn :: Handle -> IO ByteString
-getLn h = trimCR <$> B.hGetLine h
-
 -- | Trim trailing CR from ByteString.
 trimCR :: ByteString -> ByteString
 trimCR "" = ""
@@ -180,9 +226,9 @@ smpVersionP =
   let ver = A.decimal <* A.char '.'
    in SMPVersion <$> ver <*> ver <*> ver <*> A.decimal
 
--- | The handle for SMP encrypted transport connection over TCP.
-data THandle = THandle
-  { handle :: Handle,
+-- | The handle for SMP encrypted transport connection over Transport .
+data THandle c = THandle
+  { connection :: c,
     sndKey :: SessionKey,
     rcvKey :: SessionKey,
     blockSize :: Int
@@ -255,16 +301,16 @@ serializeTransportError = \case
   TEHandshake e -> bshow e
 
 -- | Encrypt and send block to SMP encrypted transport.
-tPutEncrypted :: THandle -> ByteString -> IO (Either TransportError ())
-tPutEncrypted THandle {handle = h, sndKey, blockSize} block =
+tPutEncrypted :: Transport c => THandle c -> ByteString -> IO (Either TransportError ())
+tPutEncrypted THandle {connection = c, sndKey, blockSize} block =
   encryptBlock sndKey (blockSize - C.authTagSize) block >>= \case
     Left _ -> pure $ Left TEEncrypt
-    Right (authTag, msg) -> Right <$> B.hPut h (C.authTagToBS authTag <> msg)
+    Right (authTag, msg) -> Right <$> cPut c (C.authTagToBS authTag <> msg)
 
 -- | Receive and decrypt block from SMP encrypted transport.
-tGetEncrypted :: THandle -> IO (Either TransportError ByteString)
-tGetEncrypted THandle {handle = h, rcvKey, blockSize} =
-  B.hGet h blockSize >>= decryptBlock rcvKey >>= \case
+tGetEncrypted :: Transport c => THandle c -> IO (Either TransportError ByteString)
+tGetEncrypted THandle {connection = c, rcvKey, blockSize} =
+  cGet c blockSize >>= decryptBlock rcvKey >>= \case
     Left _ -> pure $ Left TEDecrypt
     Right "" -> ioe_EOF
     Right msg -> pure $ Right msg
@@ -294,14 +340,14 @@ makeNextIV SessionKey {baseIV, counter} = atomically $ do
 -- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#appendix-a
 --
 -- The numbers in function names refer to the steps in the document.
-serverHandshake :: Handle -> C.FullKeyPair -> ExceptT TransportError IO THandle
-serverHandshake h (k, pk) = do
+serverHandshake :: forall c. Transport c => c -> C.FullKeyPair -> ExceptT TransportError IO (THandle c)
+serverHandshake c (k, pk) = do
   liftIO sendHeaderAndPublicKey_1
   encryptedKeys <- receiveEncryptedKeys_4
   -- TODO server currently ignores blockSize returned by the client
   -- this is reserved for future support of streams
   ClientHandshake {blockSize = _, sndKey, rcvKey} <- decryptParseKeys_5 encryptedKeys
-  th <- liftIO $ transportHandle h rcvKey sndKey transportBlockSize -- keys are swapped here
+  th <- liftIO $ transportHandle c rcvKey sndKey transportBlockSize -- keys are swapped here
   sendWelcome_6 th
   pure th
   where
@@ -309,17 +355,18 @@ serverHandshake h (k, pk) = do
     sendHeaderAndPublicKey_1 = do
       let sKey = C.encodePubKey k
           header = ServerHeader {blockSize = transportBlockSize, keySize = B.length sKey}
-      B.hPut h $ binaryServerHeader header <> sKey
+      cPut c $ binaryServerHeader header
+      cPut c sKey
     receiveEncryptedKeys_4 :: ExceptT TransportError IO ByteString
     receiveEncryptedKeys_4 =
-      liftIO (B.hGet h $ C.publicKeySize k) >>= \case
+      liftIO (cGet c $ C.publicKeySize k) >>= \case
         "" -> throwE $ TEHandshake TERMINATED
         ks -> pure ks
     decryptParseKeys_5 :: ByteString -> ExceptT TransportError IO ClientHandshake
     decryptParseKeys_5 encKeys =
       liftError (const $ TEHandshake DECRYPT) (C.decryptOAEP pk encKeys)
         >>= liftEither . parseClientHandshake
-    sendWelcome_6 :: THandle -> ExceptT TransportError IO ()
+    sendWelcome_6 :: THandle c -> ExceptT TransportError IO ()
     sendWelcome_6 th = ExceptT . tPutEncrypted th $ serializeSMPVersion currentSMPVersion <> " "
 
 -- | Client SMP encrypted transport handshake.
@@ -327,23 +374,23 @@ serverHandshake h (k, pk) = do
 -- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#appendix-a
 --
 -- The numbers in function names refer to the steps in the document.
-clientHandshake :: Handle -> Maybe C.KeyHash -> ExceptT TransportError IO THandle
-clientHandshake h keyHash = do
+clientHandshake :: forall c. Transport c => c -> Maybe C.KeyHash -> ExceptT TransportError IO (THandle c)
+clientHandshake c keyHash = do
   (k, blkSize) <- getHeaderAndPublicKey_1_2
   -- TODO currently client always uses the blkSize returned by the server
   keys@ClientHandshake {sndKey, rcvKey} <- liftIO $ generateKeys_3 blkSize
   sendEncryptedKeys_4 k keys
-  th <- liftIO $ transportHandle h sndKey rcvKey blkSize
+  th <- liftIO $ transportHandle c sndKey rcvKey blkSize
   getWelcome_6 th >>= checkVersion
   pure th
   where
     getHeaderAndPublicKey_1_2 :: ExceptT TransportError IO (C.PublicKey, Int)
     getHeaderAndPublicKey_1_2 = do
-      header <- liftIO (B.hGet h serverHeaderSize)
+      header <- liftIO (cGet c serverHeaderSize)
       ServerHeader {blockSize, keySize} <- liftEither $ parse serverHeaderP (TEHandshake HEADER) header
       when (blockSize < transportBlockSize || blockSize > maxTransportBlockSize) $
         throwError $ TEHandshake HEADER
-      s <- liftIO $ B.hGet h keySize
+      s <- liftIO $ cGet c keySize
       maybe (pure ()) (validateKeyHash_2 s) keyHash
       key <- liftEither $ parseKey s
       pure (key, blockSize)
@@ -363,8 +410,8 @@ clientHandshake h keyHash = do
     sendEncryptedKeys_4 :: C.PublicKey -> ClientHandshake -> ExceptT TransportError IO ()
     sendEncryptedKeys_4 k keys =
       liftError (const $ TEHandshake ENCRYPT) (C.encryptOAEP k $ serializeClientHandshake keys)
-        >>= liftIO . B.hPut h
-    getWelcome_6 :: THandle -> ExceptT TransportError IO SMPVersion
+        >>= liftIO . cPut c
+    getWelcome_6 :: THandle c -> ExceptT TransportError IO SMPVersion
     getWelcome_6 th = ExceptT $ (>>= parseSMPVersion) <$> tGetEncrypted th
     parseSMPVersion :: ByteString -> Either TransportError SMPVersion
     parseSMPVersion = first (const $ TEHandshake VERSION) . A.parseOnly (smpVersionP <* A.space)
@@ -428,13 +475,13 @@ binaryRsaTransportP = binaryRsa =<< int16
 parseClientHandshake :: ByteString -> Either TransportError ClientHandshake
 parseClientHandshake = parse clientHandshakeP $ TEHandshake AES_KEYS
 
-transportHandle :: Handle -> SessionKey -> SessionKey -> Int -> IO THandle
-transportHandle h sk rk blockSize = do
+transportHandle :: c -> SessionKey -> SessionKey -> Int -> IO (THandle c)
+transportHandle c sk rk blockSize = do
   sndCounter <- newTVarIO 0
   rcvCounter <- newTVarIO 0
   pure
     THandle
-      { handle = h,
+      { connection = c,
         sndKey = sk {counter = sndCounter},
         rcvKey = rk {counter = rcvCounter},
         blockSize

src/Simplex/Messaging/Transport/WebSockets.hs

@@ -0,0 +1,62 @@
+{-# LANGUAGE InstanceSigs #-}
+
+module Simplex.Messaging.Transport.WebSockets (WS (..)) where
+
+import qualified Control.Exception as E
+import Data.ByteString.Char8 (ByteString)
+import qualified Data.ByteString.Char8 as B
+import Network.Socket (Socket)
+import Network.WebSockets
+import Network.WebSockets.Stream (Stream)
+import qualified Network.WebSockets.Stream as S
+import Simplex.Messaging.Transport (TProxy, Transport (..), TransportError (..), trimCR)
+
+data WS = WS {wsStream :: Stream, wsConnection :: Connection}
+
+websocketsOpts :: ConnectionOptions
+websocketsOpts =
+  defaultConnectionOptions
+    { connectionCompressionOptions = NoCompression,
+      connectionFramePayloadSizeLimit = SizeLimit 8192,
+      connectionMessageDataSizeLimit = SizeLimit 65536
+    }
+
+instance Transport WS where
+  transportName :: TProxy WS -> String
+  transportName _ = "WebSockets"
+
+  getServerConnection :: Socket -> IO WS
+  getServerConnection sock = do
+    s <- S.makeSocketStream sock
+    WS s <$> acceptClientRequest s
+    where
+      acceptClientRequest :: Stream -> IO Connection
+      acceptClientRequest s = makePendingConnectionFromStream s websocketsOpts >>= acceptRequest
+
+  getClientConnection :: Socket -> IO WS
+  getClientConnection sock = do
+    s <- S.makeSocketStream sock
+    WS s <$> sendClientRequest s
+    where
+      sendClientRequest :: Stream -> IO Connection
+      sendClientRequest s = newClientConnection s "" "/" websocketsOpts []
+
+  closeConnection :: WS -> IO ()
+  closeConnection = S.close . wsStream
+
+  cGet :: WS -> Int -> IO ByteString
+  cGet c n = do
+    s <- receiveData (wsConnection c)
+    if B.length s == n
+      then pure s
+      else E.throwIO TEBadBlock
+
+  cPut :: WS -> ByteString -> IO ()
+  cPut = sendBinaryData . wsConnection
+
+  getLn :: WS -> IO ByteString
+  getLn c = do
+    s <- trimCR <$> receiveData (wsConnection c)
+    if B.null s || B.last s /= '\n'
+      then E.throwIO TEBadBlock
+      else pure $ B.init s

tests/AgentTests.hs

@@ -5,6 +5,7 @@
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE PatternSynonyms #-}
+{-# LANGUAGE ScopedTypeVariables #-}
 
 module AgentTests where
 
@@ -15,37 +16,29 @@ import qualified Data.ByteString.Char8 as B
 import SMPAgentClient
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Protocol (ErrorType (..), MsgBody)
-import System.IO (Handle)
+import Simplex.Messaging.Transport (ATransport (..), TProxy (..), Transport (..))
 import System.Timeout
 import Test.Hspec
 
-agentTests :: Spec
-agentTests = do
+agentTests :: ATransport -> Spec
+agentTests (ATransport t) = do
   describe "SQLite store" storeTests
-  describe "SMP agent protocol syntax" syntaxTests
+  describe "SMP agent protocol syntax" $ syntaxTests t
   describe "Establishing duplex connection" do
     it "should connect via one server and one agent" $
-      smpAgentTest2_1_1 testDuplexConnection
+      smpAgentTest2_1_1 $ testDuplexConnection t
     it "should connect via one server and 2 agents" $
-      smpAgentTest2_2_1 testDuplexConnection
+      smpAgentTest2_2_1 $ testDuplexConnection t
     it "should connect via 2 servers and 2 agents" $
-      smpAgentTest2_2_2 testDuplexConnection
+      smpAgentTest2_2_2 $ testDuplexConnection t
   describe "Connection subscriptions" do
     it "should connect via one server and one agent" $
-      smpAgentTest3_1_1 testSubscription
+      smpAgentTest3_1_1 $ testSubscription t
     it "should send notifications to client when server disconnects" $
-      smpAgentServerTest testSubscrNotification
-
--- | simple test for one command with the expected response
-(>#>) :: ARawTransmission -> ARawTransmission -> Expectation
-command >#> response = smpAgentTest command `shouldReturn` response
-
--- | simple test for one command with a predicate for the expected response
-(>#>=) :: ARawTransmission -> ((ByteString, ByteString, [ByteString]) -> Bool) -> Expectation
-command >#>= p = smpAgentTest command >>= (`shouldSatisfy` p . \(cId, cAlias, cmd) -> (cId, cAlias, B.words cmd))
+      smpAgentServerTest $ testSubscrNotification t
 
 -- | send transmission `t` to handle `h` and get response
-(#:) :: Handle -> (ByteString, ByteString, ByteString) -> IO (ATransmissionOrError 'Agent)
+(#:) :: Transport c => c -> (ByteString, ByteString, ByteString) -> IO (ATransmissionOrError 'Agent)
 h #: t = tPutRaw h t >> tGet SAgent h
 
 -- | action and expected response
@@ -64,15 +57,15 @@ correctTransmission (corrId, cAlias, cmdOrErr) = case cmdOrErr of
   Left e -> error $ show e
 
 -- | receive message to handle `h` and validate that it is the expected one
-(<#) :: Handle -> ATransmission 'Agent -> Expectation
+(<#) :: Transport c => c -> ATransmission 'Agent -> Expectation
 h <# (corrId, cAlias, cmd) = tGet SAgent h `shouldReturn` (corrId, cAlias, Right cmd)
 
 -- | receive message to handle `h` and validate it using predicate `p`
-(<#=) :: Handle -> (ATransmission 'Agent -> Bool) -> Expectation
+(<#=) :: Transport c => c -> (ATransmission 'Agent -> Bool) -> Expectation
 h <#= p = tGet SAgent h >>= (`shouldSatisfy` p . correctTransmission)
 
 -- | test that nothing is delivered to handle `h` during 10ms
-(#:#) :: Handle -> String -> Expectation
+(#:#) :: Transport c => c -> String -> Expectation
 h #:# err = tryGet `shouldReturn` ()
   where
     tryGet =
@@ -83,8 +76,8 @@ h #:# err = tryGet `shouldReturn` ()
 pattern Msg :: MsgBody -> ACommand 'Agent
 pattern Msg msgBody <- MSG {msgBody, msgIntegrity = MsgOk}
 
-testDuplexConnection :: Handle -> Handle -> IO ()
-testDuplexConnection alice bob = do
+testDuplexConnection :: Transport c => TProxy c -> c -> c -> IO ()
+testDuplexConnection _ alice bob = do
   ("1", "bob", Right (INV qInfo)) <- alice #: ("1", "bob", "NEW")
   let qInfo' = serializeSmpQueueInfo qInfo
   bob #: ("11", "alice", "JOIN " <> qInfo') #> ("", "alice", CON)
@@ -102,8 +95,8 @@ testDuplexConnection alice bob = do
   alice #: ("6", "bob", "DEL") #> ("6", "bob", OK)
   alice #:# "nothing else should be delivered to alice"
 
-testSubscription :: Handle -> Handle -> Handle -> IO ()
-testSubscription alice1 alice2 bob = do
+testSubscription :: Transport c => TProxy c -> c -> c -> c -> IO ()
+testSubscription _ alice1 alice2 bob = do
   ("1", "bob", Right (INV qInfo)) <- alice1 #: ("1", "bob", "NEW")
   let qInfo' = serializeSmpQueueInfo qInfo
   bob #: ("11", "alice", "JOIN " <> qInfo') #> ("", "alice", CON)
@@ -118,8 +111,8 @@ testSubscription alice1 alice2 bob = do
   alice2 <#= \case ("", "bob", Msg "hi") -> True; _ -> False
   alice1 #:# "nothing else should be delivered to alice1"
 
-testSubscrNotification :: (ThreadId, ThreadId) -> Handle -> IO ()
-testSubscrNotification (server, _) client = do
+testSubscrNotification :: Transport c => TProxy c -> (ThreadId, ThreadId) -> c -> IO ()
+testSubscrNotification _ (server, _) client = do
   client #: ("1", "conn1", "NEW") =#> \case ("1", "conn1", INV _) -> True; _ -> False
   client #:# "nothing should be delivered to client before the server is killed"
   killThread server
@@ -128,8 +121,8 @@ testSubscrNotification (server, _) client = do
 samplePublicKey :: ByteString
 samplePublicKey = "rsa:MIIBoDANBgkqhkiG9w0BAQEFAAOCAY0AMIIBiAKCAQEAtn1NI2tPoOGSGfad0aUg0tJ0kG2nzrIPGLiz8wb3dQSJC9xkRHyzHhEE8Kmy2cM4q7rNZIlLcm4M7oXOTe7SC4x59bLQG9bteZPKqXu9wk41hNamV25PWQ4zIcIRmZKETVGbwN7jFMpH7wxLdI1zzMArAPKXCDCJ5ctWh4OWDI6OR6AcCtEj+toCI6N6pjxxn5VigJtwiKhxYpoUJSdNM60wVEDCSUrZYBAuDH8pOxPfP+Tm4sokaFDTIG3QJFzOjC+/9nW4MUjAOFll9PCp9kaEFHJ/YmOYKMWNOCCPvLS6lxA83i0UaardkNLNoFS5paWfTlroxRwOC2T6PwO2ywKBgDjtXcSED61zK1seocQMyGRINnlWdhceD669kIHju/f6kAayvYKW3/lbJNXCmyinAccBosO08/0sUxvtuniIo18kfYJE0UmP1ReCjhMP+O+yOmwZJini/QelJk/Pez8IIDDWnY1qYQsN/q7ocjakOYrpGG7mig6JMFpDJtD6istR"
 
-syntaxTests :: Spec
-syntaxTests = do
+syntaxTests :: forall c. Transport c => TProxy c -> Spec
+syntaxTests t = do
   it "unknown command" $ ("1", "5678", "HELLO") >#> ("1", "5678", "ERR CMD SYNTAX")
   describe "NEW" do
     describe "valid" do
@@ -149,3 +142,11 @@ syntaxTests = do
     describe "invalid" do
       -- TODO: JOIN is not merged yet - to be added
       it "no parameters" $ ("321", "", "JOIN") >#> ("321", "", "ERR CMD SYNTAX")
+  where
+    -- simple test for one command with the expected response
+    (>#>) :: ARawTransmission -> ARawTransmission -> Expectation
+    command >#> response = smpAgentTest t command `shouldReturn` response
+
+    -- simple test for one command with a predicate for the expected response
+    (>#>=) :: ARawTransmission -> ((ByteString, ByteString, [ByteString]) -> Bool) -> Expectation
+    command >#>= p = smpAgentTest t command >>= (`shouldSatisfy` p . \(cId, cAlias, cmd) -> (cId, cAlias, B.words cmd))

tests/SMPAgentClient.hs

@@ -3,6 +3,7 @@
 {-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeApplications #-}
 
 module SMPAgentClient where
 
@@ -28,7 +29,6 @@ import Simplex.Messaging.Transport
 import Test.Hspec
 import UnliftIO.Concurrent
 import UnliftIO.Directory
-import UnliftIO.IO
 
 agentTestHost :: HostName
 agentTestHost = "localhost"
@@ -51,44 +51,50 @@ testDB2 = "tests/tmp/smp-agent2.test.protocol.db"
 testDB3 :: String
 testDB3 = "tests/tmp/smp-agent3.test.protocol.db"
 
-smpAgentTest :: ARawTransmission -> IO ARawTransmission
-smpAgentTest cmd = runSmpAgentTest $ \h -> tPutRaw h cmd >> tGetRaw h
+smpAgentTest :: forall c. Transport c => TProxy c -> ARawTransmission -> IO ARawTransmission
+smpAgentTest _ cmd = runSmpAgentTest $ \(h :: c) -> tPutRaw h cmd >> tGetRaw h
 
-runSmpAgentTest :: (MonadUnliftIO m, MonadRandom m) => (Handle -> m a) -> m a
-runSmpAgentTest test = withSmpServer . withSmpAgent $ testSMPAgentClient test
+runSmpAgentTest :: forall c m a. (Transport c, MonadUnliftIO m, MonadRandom m) => (c -> m a) -> m a
+runSmpAgentTest test = withSmpServer t . withSmpAgent t $ testSMPAgentClient test
+  where
+    t = transport @c
 
-runSmpAgentServerTest :: (MonadUnliftIO m, MonadRandom m) => ((ThreadId, ThreadId) -> Handle -> m a) -> m a
+runSmpAgentServerTest :: forall c m a. (Transport c, MonadUnliftIO m, MonadRandom m) => ((ThreadId, ThreadId) -> c -> m a) -> m a
 runSmpAgentServerTest test =
-  withSmpServerThreadOn testPort $
-    \server -> withSmpAgentThreadOn (agentTestPort, testPort, testDB) $
+  withSmpServerThreadOn t testPort $
+    \server -> withSmpAgentThreadOn t (agentTestPort, testPort, testDB) $
       \agent -> testSMPAgentClient $ test (server, agent)
+  where
+    t = transport @c
 
-smpAgentServerTest :: ((ThreadId, ThreadId) -> Handle -> IO ()) -> Expectation
+smpAgentServerTest :: Transport c => ((ThreadId, ThreadId) -> c -> IO ()) -> Expectation
 smpAgentServerTest test' = runSmpAgentServerTest test' `shouldReturn` ()
 
-runSmpAgentTestN :: forall m a. (MonadUnliftIO m, MonadRandom m) => [(ServiceName, ServiceName, String)] -> ([Handle] -> m a) -> m a
-runSmpAgentTestN agents test = withSmpServer $ run agents []
+runSmpAgentTestN :: forall c m a. (Transport c, MonadUnliftIO m, MonadRandom m) => [(ServiceName, ServiceName, String)] -> ([c] -> m a) -> m a
+runSmpAgentTestN agents test = withSmpServer t $ run agents []
   where
-    run :: [(ServiceName, ServiceName, String)] -> [Handle] -> m a
+    run :: [(ServiceName, ServiceName, String)] -> [c] -> m a
     run [] hs = test hs
-    run (a@(p, _, _) : as) hs = withSmpAgentOn a $ testSMPAgentClientOn p $ \h -> run as (h : hs)
+    run (a@(p, _, _) : as) hs = withSmpAgentOn t a $ testSMPAgentClientOn p $ \h -> run as (h : hs)
+    t = transport @c
 
-runSmpAgentTestN_1 :: forall m a. (MonadUnliftIO m, MonadRandom m) => Int -> ([Handle] -> m a) -> m a
-runSmpAgentTestN_1 nClients test = withSmpServer . withSmpAgent $ run nClients []
+runSmpAgentTestN_1 :: forall c m a. (Transport c, MonadUnliftIO m, MonadRandom m) => Int -> ([c] -> m a) -> m a
+runSmpAgentTestN_1 nClients test = withSmpServer t . withSmpAgent t $ run nClients []
   where
-    run :: Int -> [Handle] -> m a
+    run :: Int -> [c] -> m a
     run 0 hs = test hs
     run n hs = testSMPAgentClient $ \h -> run (n - 1) (h : hs)
+    t = transport @c
 
-smpAgentTestN :: [(ServiceName, ServiceName, String)] -> ([Handle] -> IO ()) -> Expectation
+smpAgentTestN :: Transport c => [(ServiceName, ServiceName, String)] -> ([c] -> IO ()) -> Expectation
 smpAgentTestN agents test' = runSmpAgentTestN agents test' `shouldReturn` ()
 
-smpAgentTestN_1 :: Int -> ([Handle] -> IO ()) -> Expectation
+smpAgentTestN_1 :: Transport c => Int -> ([c] -> IO ()) -> Expectation
 smpAgentTestN_1 n test' = runSmpAgentTestN_1 n test' `shouldReturn` ()
 
-smpAgentTest2_2_2 :: (Handle -> Handle -> IO ()) -> Expectation
+smpAgentTest2_2_2 :: forall c. Transport c => (c -> c -> IO ()) -> Expectation
 smpAgentTest2_2_2 test' =
-  withSmpServerOn testPort2 $
+  withSmpServerOn (transport @c) testPort2 $
     smpAgentTestN
       [ (agentTestPort, testPort, testDB),
         (agentTestPort2, testPort2, testDB2)
@@ -98,7 +104,7 @@ smpAgentTest2_2_2 test' =
     _test [h1, h2] = test' h1 h2
     _test _ = error "expected 2 handles"
 
-smpAgentTest2_2_1 :: (Handle -> Handle -> IO ()) -> Expectation
+smpAgentTest2_2_1 :: Transport c => (c -> c -> IO ()) -> Expectation
 smpAgentTest2_2_1 test' =
   smpAgentTestN
     [ (agentTestPort, testPort, testDB),
@@ -109,13 +115,13 @@ smpAgentTest2_2_1 test' =
     _test [h1, h2] = test' h1 h2
     _test _ = error "expected 2 handles"
 
-smpAgentTest2_1_1 :: (Handle -> Handle -> IO ()) -> Expectation
+smpAgentTest2_1_1 :: Transport c => (c -> c -> IO ()) -> Expectation
 smpAgentTest2_1_1 test' = smpAgentTestN_1 2 _test
   where
     _test [h1, h2] = test' h1 h2
     _test _ = error "expected 2 handles"
 
-smpAgentTest3 :: (Handle -> Handle -> Handle -> IO ()) -> Expectation
+smpAgentTest3 :: Transport c => (c -> c -> c -> IO ()) -> Expectation
 smpAgentTest3 test' =
   smpAgentTestN
     [ (agentTestPort, testPort, testDB),
@@ -127,7 +133,7 @@ smpAgentTest3 test' =
     _test [h1, h2, h3] = test' h1 h2 h3
     _test _ = error "expected 3 handles"
 
-smpAgentTest3_1_1 :: (Handle -> Handle -> Handle -> IO ()) -> Expectation
+smpAgentTest3_1_1 :: Transport c => (c -> c -> c -> IO ()) -> Expectation
 smpAgentTest3_1_1 test' = smpAgentTestN_1 3 _test
   where
     _test [h1, h2, h3] = test' h1 h2 h3
@@ -145,31 +151,31 @@ cfg =
       smpCfg =
         smpDefaultConfig
           { qSize = 1,
-            defaultPort = testPort,
+            defaultTransport = (testPort, transport @TCP),
             tcpTimeout = 500_000
           }
     }
 
-withSmpAgentThreadOn :: (MonadUnliftIO m, MonadRandom m) => (ServiceName, ServiceName, String) -> (ThreadId -> m a) -> m a
-withSmpAgentThreadOn (port', smpPort', db') =
+withSmpAgentThreadOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> (ThreadId -> m a) -> m a
+withSmpAgentThreadOn t (port', smpPort', db') =
   let cfg' = cfg {tcpPort = port', dbFile = db', smpServers = L.fromList [SMPServer "localhost" (Just smpPort') testKeyHash]}
    in serverBracket
-        (`runSMPAgentBlocking` cfg')
+        (\started -> runSMPAgentBlocking t started cfg')
         (removeFile db')
 
-withSmpAgentOn :: (MonadUnliftIO m, MonadRandom m) => (ServiceName, ServiceName, String) -> m a -> m a
-withSmpAgentOn (port', smpPort', db') = withSmpAgentThreadOn (port', smpPort', db') . const
+withSmpAgentOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> m a -> m a
+withSmpAgentOn t (port', smpPort', db') = withSmpAgentThreadOn t (port', smpPort', db') . const
 
-withSmpAgent :: (MonadUnliftIO m, MonadRandom m) => m a -> m a
-withSmpAgent = withSmpAgentOn (agentTestPort, testPort, testDB)
+withSmpAgent :: (MonadUnliftIO m, MonadRandom m) => ATransport -> m a -> m a
+withSmpAgent t = withSmpAgentOn t (agentTestPort, testPort, testDB)
 
-testSMPAgentClientOn :: MonadUnliftIO m => ServiceName -> (Handle -> m a) -> m a
+testSMPAgentClientOn :: (Transport c, MonadUnliftIO m) => ServiceName -> (c -> m a) -> m a
 testSMPAgentClientOn port' client = do
-  runTCPClient agentTestHost port' $ \h -> do
+  runTransportClient agentTestHost port' $ \h -> do
     line <- liftIO $ getLn h
     if line == "Welcome to SMP v0.3.1 agent"
       then client h
       else error $ "wrong welcome message: " <> B.unpack line
 
-testSMPAgentClient :: MonadUnliftIO m => (Handle -> m a) -> m a
+testSMPAgentClient :: (Transport c, MonadUnliftIO m) => (c -> m a) -> m a
 testSMPAgentClient = testSMPAgentClientOn agentTestPort

tests/SMPClient.hs

@@ -4,6 +4,7 @@
 {-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeApplications #-}
 
 module SMPClient where
 
@@ -44,9 +45,9 @@ testKeyHash = Just "KXNE1m2E1m0lm92WGKet9CL6+lO742Vy5G6nsrkvgs8="
 testStoreLogFile :: FilePath
 testStoreLogFile = "tests/tmp/smp-server-store.log"
 
-testSMPClient :: MonadUnliftIO m => (THandle -> m a) -> m a
+testSMPClient :: (Transport c, MonadUnliftIO m) => (THandle c -> m a) -> m a
 testSMPClient client =
-  runTCPClient testHost testPort $ \h ->
+  runTransportClient testHost testPort $ \h ->
     liftIO (runExceptT $ clientHandshake h testKeyHash) >>= \case
       Right th -> client th
       Left e -> error $ show e
@@ -54,7 +55,7 @@ testSMPClient client =
 cfg :: ServerConfig
 cfg =
   ServerConfig
-    { tcpPort = testPort,
+    { transports = undefined,
       tbqSize = 1,
       queueIdBytes = 12,
       msgIdBytes = 6,
@@ -91,18 +92,18 @@ cfg =
         \TmKzSAw7iVWwEUZR/PeiEKazqrpp9VU="
     }
 
-withSmpServerStoreLogOn :: (MonadUnliftIO m, MonadRandom m) => ServiceName -> (ThreadId -> m a) -> m a
-withSmpServerStoreLogOn port client = do
+withSmpServerStoreLogOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> ServiceName -> (ThreadId -> m a) -> m a
+withSmpServerStoreLogOn t port client = do
   s <- liftIO $ openReadStoreLog testStoreLogFile
   serverBracket
-    (\started -> runSMPServerBlocking started cfg {tcpPort = port, storeLog = Just s})
+    (\started -> runSMPServerBlocking started cfg {transports = [(port, t)], storeLog = Just s})
     (pure ())
     client
 
-withSmpServerThreadOn :: (MonadUnliftIO m, MonadRandom m) => ServiceName -> (ThreadId -> m a) -> m a
-withSmpServerThreadOn port =
+withSmpServerThreadOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> ServiceName -> (ThreadId -> m a) -> m a
+withSmpServerThreadOn t port =
   serverBracket
-    (\started -> runSMPServerBlocking started cfg {tcpPort = port})
+    (\started -> runSMPServerBlocking started cfg {transports = [(port, t)]})
     (pure ())
 
 serverBracket :: MonadUnliftIO m => (TMVar Bool -> m ()) -> m () -> (ThreadId -> m a) -> m a
@@ -118,49 +119,49 @@ serverBracket process afterProcess f = do
         Nothing -> error $ "server did not " <> s
         _ -> pure ()
 
-withSmpServerOn :: (MonadUnliftIO m, MonadRandom m) => ServiceName -> m a -> m a
-withSmpServerOn port = withSmpServerThreadOn port . const
+withSmpServerOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> ServiceName -> m a -> m a
+withSmpServerOn t port = withSmpServerThreadOn t port . const
 
-withSmpServer :: (MonadUnliftIO m, MonadRandom m) => m a -> m a
-withSmpServer = withSmpServerOn testPort
+withSmpServer :: (MonadUnliftIO m, MonadRandom m) => ATransport -> m a -> m a
+withSmpServer t = withSmpServerOn t testPort
 
-runSmpTest :: (MonadUnliftIO m, MonadRandom m) => (THandle -> m a) -> m a
-runSmpTest test = withSmpServer $ testSMPClient test
+runSmpTest :: forall c m a. (Transport c, MonadUnliftIO m, MonadRandom m) => (THandle c -> m a) -> m a
+runSmpTest test = withSmpServer (transport @c) $ testSMPClient test
 
-runSmpTestN :: forall m a. (MonadUnliftIO m, MonadRandom m) => Int -> ([THandle] -> m a) -> m a
-runSmpTestN nClients test = withSmpServer $ run nClients []
+runSmpTestN :: forall c m a. (Transport c, MonadUnliftIO m, MonadRandom m) => Int -> ([THandle c] -> m a) -> m a
+runSmpTestN nClients test = withSmpServer (transport @c) $ run nClients []
   where
-    run :: Int -> [THandle] -> m a
+    run :: Int -> [THandle c] -> m a
     run 0 hs = test hs
     run n hs = testSMPClient $ \h -> run (n - 1) (h : hs)
 
-smpServerTest :: RawTransmission -> IO RawTransmission
-smpServerTest cmd = runSmpTest $ \h -> tPutRaw h cmd >> tGetRaw h
+smpServerTest :: forall c. Transport c => TProxy c -> RawTransmission -> IO RawTransmission
+smpServerTest _ cmd = runSmpTest $ \(h :: THandle c) -> tPutRaw h cmd >> tGetRaw h
 
-smpTest :: (THandle -> IO ()) -> Expectation
-smpTest test' = runSmpTest test' `shouldReturn` ()
+smpTest :: Transport c => TProxy c -> (THandle c -> IO ()) -> Expectation
+smpTest _ test' = runSmpTest test' `shouldReturn` ()
 
-smpTestN :: Int -> ([THandle] -> IO ()) -> Expectation
+smpTestN :: Transport c => Int -> ([THandle c] -> IO ()) -> Expectation
 smpTestN n test' = runSmpTestN n test' `shouldReturn` ()
 
-smpTest2 :: (THandle -> THandle -> IO ()) -> Expectation
-smpTest2 test' = smpTestN 2 _test
+smpTest2 :: Transport c => TProxy c -> (THandle c -> THandle c -> IO ()) -> Expectation
+smpTest2 _ test' = smpTestN 2 _test
   where
     _test [h1, h2] = test' h1 h2
     _test _ = error "expected 2 handles"
 
-smpTest3 :: (THandle -> THandle -> THandle -> IO ()) -> Expectation
-smpTest3 test' = smpTestN 3 _test
+smpTest3 :: Transport c => TProxy c -> (THandle c -> THandle c -> THandle c -> IO ()) -> Expectation
+smpTest3 _ test' = smpTestN 3 _test
   where
     _test [h1, h2, h3] = test' h1 h2 h3
     _test _ = error "expected 3 handles"
 
-tPutRaw :: THandle -> RawTransmission -> IO ()
+tPutRaw :: Transport c => THandle c -> RawTransmission -> IO ()
 tPutRaw h (sig, corrId, queueId, command) = do
   let t = B.intercalate " " [corrId, queueId, command]
   void $ tPut h (C.Signature sig, t)
 
-tGetRaw :: THandle -> IO RawTransmission
+tGetRaw :: Transport c => THandle c -> IO RawTransmission
 tGetRaw h = do
   ("", (CorrId corrId, qId, Right cmd)) <- tGet fromServer h
   pure ("", corrId, encode qId, serializeCommand cmd)

tests/ServerTests.hs

@@ -28,25 +28,25 @@ import Test.Hspec
 rsaKeySize :: Int
 rsaKeySize = 2048 `div` 8
 
-serverTests :: Spec
-serverTests = do
-  describe "SMP syntax" syntaxTests
+serverTests :: ATransport -> Spec
+serverTests t = do
+  describe "SMP syntax" $ syntaxTests t
   describe "SMP queues" do
-    describe "NEW and KEY commands, SEND messages" testCreateSecure
-    describe "NEW, OFF and DEL commands, SEND messages" testCreateDelete
+    describe "NEW and KEY commands, SEND messages" $ testCreateSecure t
+    describe "NEW, OFF and DEL commands, SEND messages" $ testCreateDelete t
   describe "SMP messages" do
-    describe "duplex communication over 2 SMP connections" testDuplex
-    describe "switch subscription to another SMP queue" testSwitchSub
-  describe "Store log" testWithStoreLog
-  describe "Timing of AUTH error" testTiming
+    describe "duplex communication over 2 SMP connections" $ testDuplex t
+    describe "switch subscription to another SMP queue" $ testSwitchSub t
+  describe "Store log" $ testWithStoreLog t
+  describe "Timing of AUTH error" $ testTiming t
 
 pattern Resp :: CorrId -> QueueId -> Command 'Broker -> SignedTransmissionOrError
 pattern Resp corrId queueId command <- ("", (corrId, queueId, Right (Cmd SBroker command)))
 
-sendRecv :: THandle -> (ByteString, ByteString, ByteString, ByteString) -> IO SignedTransmissionOrError
+sendRecv :: Transport c => THandle c -> (ByteString, ByteString, ByteString, ByteString) -> IO SignedTransmissionOrError
 sendRecv h (sgn, corrId, qId, cmd) = tPutRaw h (sgn, corrId, encode qId, cmd) >> tGet fromServer h
 
-signSendRecv :: THandle -> C.SafePrivateKey -> (ByteString, ByteString, ByteString) -> IO SignedTransmissionOrError
+signSendRecv :: Transport c => THandle c -> C.SafePrivateKey -> (ByteString, ByteString, ByteString) -> IO SignedTransmissionOrError
 signSendRecv h pk (corrId, qId, cmd) = do
   let t = B.intercalate " " [corrId, encode qId, cmd]
   Right sig <- runExceptT $ C.sign pk t
@@ -56,16 +56,13 @@ signSendRecv h pk (corrId, qId, cmd) = do
 cmdSEND :: ByteString -> ByteString
 cmdSEND msg = serializeCommand (Cmd SSender . SEND $ msg)
 
-(>#>) :: RawTransmission -> RawTransmission -> Expectation
-command >#> response = smpServerTest command `shouldReturn` response
-
 (#==) :: (HasCallStack, Eq a, Show a) => (a, a) -> String -> Assertion
 (actual, expected) #== message = assertEqual message expected actual
 
-testCreateSecure :: Spec
-testCreateSecure =
+testCreateSecure :: ATransport -> Spec
+testCreateSecure (ATransport t) =
   it "should create (NEW) and secure (KEY) queue" $
-    smpTest \h -> do
+    smpTest t $ \h -> do
       (rPub, rKey) <- C.generateKeyPair rsaKeySize
       Resp "abcd" rId1 (IDS rId sId) <- signSendRecv h rKey ("abcd", "", "NEW " <> C.serializePubKey rPub)
       (rId1, "") #== "creates queue"
@@ -114,10 +111,10 @@ testCreateSecure =
       Resp "dabc" _ err5 <- sendRecv h ("", "dabc", sId, "SEND 5 hello ")
       (err5, ERR AUTH) #== "rejects unsigned SEND"
 
-testCreateDelete :: Spec
-testCreateDelete =
+testCreateDelete :: ATransport -> Spec
+testCreateDelete (ATransport t) =
   it "should create (NEW), suspend (OFF) and delete (DEL) queue" $
-    smpTest2 \rh sh -> do
+    smpTest2 t $ \rh sh -> do
       (rPub, rKey) <- C.generateKeyPair rsaKeySize
       Resp "abcd" rId1 (IDS rId sId) <- signSendRecv rh rKey ("abcd", "", "NEW " <> C.serializePubKey rPub)
       (rId1, "") #== "creates queue"
@@ -182,10 +179,10 @@ testCreateDelete =
       Resp "cdab" _ err10 <- signSendRecv rh rKey ("cdab", rId, "SUB")
       (err10, ERR AUTH) #== "rejects SUB when deleted"
 
-testDuplex :: Spec
-testDuplex =
+testDuplex :: ATransport -> Spec
+testDuplex (ATransport t) =
   it "should create 2 simplex connections and exchange messages" $
-    smpTest2 \alice bob -> do
+    smpTest2 t $ \alice bob -> do
       (arPub, arKey) <- C.generateKeyPair rsaKeySize
       Resp "abcd" _ (IDS aRcv aSnd) <- signSendRecv alice arKey ("abcd", "", "NEW " <> C.serializePubKey arPub)
       -- aSnd ID is passed to Bob out-of-band
@@ -232,10 +229,10 @@ testDuplex =
       Resp "bcda" _ OK <- signSendRecv bob brKey ("bcda", bRcv, "ACK")
       (msg5, "how are you bob") #== "message received from alice"
 
-testSwitchSub :: Spec
-testSwitchSub =
+testSwitchSub :: ATransport -> Spec
+testSwitchSub (ATransport t) =
   it "should create simplex connections and switch subscription to another TCP connection" $
-    smpTest3 \rh1 rh2 sh -> do
+    smpTest3 t $ \rh1 rh2 sh -> do
       (rPub, rKey) <- C.generateKeyPair rsaKeySize
       Resp "abcd" _ (IDS rId sId) <- signSendRecv rh1 rKey ("abcd", "", "NEW " <> C.serializePubKey rPub)
       Resp "bcda" _ ok1 <- sendRecv sh ("", "bcda", sId, "SEND 5 test1 ")
@@ -270,15 +267,15 @@ testSwitchSub =
         Nothing -> return ()
         Just _ -> error "nothing else is delivered to the 1st TCP connection"
 
-testWithStoreLog :: Spec
-testWithStoreLog =
+testWithStoreLog :: ATransport -> Spec
+testWithStoreLog at@(ATransport t) =
   it "should store simplex queues to log and restore them after server restart" $ do
     (sPub1, sKey1) <- C.generateKeyPair rsaKeySize
     (sPub2, sKey2) <- C.generateKeyPair rsaKeySize
     senderId1 <- newTVarIO ""
     senderId2 <- newTVarIO ""
 
-    withSmpServerStoreLogOn testPort . runTest $ \h -> do
+    withSmpServerStoreLogOn at testPort . runTest t $ \h -> do
       (sId1, _, _) <- createAndSecureQueue h sPub1
       atomically $ writeTVar senderId1 sId1
       Resp "bcda" _ OK <- signSendRecv h sKey1 ("bcda", sId1, "SEND 5 hello ")
@@ -294,13 +291,13 @@ testWithStoreLog =
 
     logSize `shouldReturn` 5
 
-    withSmpServerThreadOn testPort . runTest $ \h -> do
+    withSmpServerThreadOn at testPort . runTest t $ \h -> do
       sId1 <- readTVarIO senderId1
       -- fails if store log is disabled
       Resp "bcda" _ (ERR AUTH) <- signSendRecv h sKey1 ("bcda", sId1, "SEND 5 hello ")
       pure ()
 
-    withSmpServerStoreLogOn testPort . runTest $ \h -> do
+    withSmpServerStoreLogOn at testPort . runTest t $ \h -> do
       -- this queue is restored
       sId1 <- readTVarIO senderId1
       Resp "bcda" _ OK <- signSendRecv h sKey1 ("bcda", sId1, "SEND 5 hello ")
@@ -312,7 +309,7 @@ testWithStoreLog =
     logSize `shouldReturn` 1
     removeFile testStoreLogFile
   where
-    createAndSecureQueue :: THandle -> SenderPublicKey -> IO (SenderId, RecipientId, C.SafePrivateKey)
+    createAndSecureQueue :: Transport c => THandle c -> SenderPublicKey -> IO (SenderId, RecipientId, C.SafePrivateKey)
     createAndSecureQueue h sPub = do
       (rPub, rKey) <- C.generateKeyPair rsaKeySize
       Resp "abcd" "" (IDS rId sId) <- signSendRecv h rKey ("abcd", "", "NEW " <> C.serializePubKey rPub)
@@ -321,8 +318,8 @@ testWithStoreLog =
       (rId', rId) #== "same queue ID"
       pure (sId, rId, rKey)
 
-    runTest :: (THandle -> IO ()) -> ThreadId -> Expectation
-    runTest test' server = do
+    runTest :: Transport c => TProxy c -> (THandle c -> IO ()) -> ThreadId -> Expectation
+    runTest _ test' server = do
       testSMPClient test' `shouldReturn` ()
       killThread server
 
@@ -332,10 +329,10 @@ testWithStoreLog =
         Right l -> pure l
         Left (_ :: SomeException) -> logSize
 
-testTiming :: Spec
-testTiming =
+testTiming :: ATransport -> Spec
+testTiming (ATransport t) =
   it "should have similar time for auth error, whether queue exists or not, for all key sizes" $
-    smpTest2 \rh sh ->
+    smpTest2 t $ \rh sh ->
       mapM_
         (testSameTiming rh sh)
         [ (128, 128, 100),
@@ -351,7 +348,7 @@ testTiming =
   where
     timeRepeat n = fmap fst . timeItT . forM_ (replicate n ()) . const
     similarTime t1 t2 = abs (t1 - t2) / t1 < 0.2 `shouldBe` True
-    testSameTiming :: THandle -> THandle -> (Int, Int, Int) -> Expectation
+    testSameTiming :: Transport c => THandle c -> THandle c -> (Int, Int, Int) -> Expectation
     testSameTiming rh sh (senderKeySize, badKeySize, n) = do
       (rPub, rKey) <- C.generateKeyPair rsaKeySize
       Resp "abcd" "" (IDS rId sId) <- signSendRecv rh rKey ("abcd", "", "NEW " <> C.serializePubKey rPub)
@@ -377,8 +374,8 @@ samplePubKey = "rsa:MIIBoDANBgkqhkiG9w0BAQEFAAOCAY0AMIIBiAKCAQEAtn1NI2tPoOGSGfad
 sampleSig :: ByteString
 sampleSig = "\128\207*\159eq\220i!\"\157\161\130\184\226\246\232_\\\170`\180\160\230sI\154\197\211\252\SUB\246\206ELL\t9K\ESC\196?\128\215%\222\148\NAK;9\155f\164\217e\242\156\CAN9\253\r\170\174'w\211\228?\205)\215\150\255\247z\DC115\DC1{\bn\145\rKD,K\230\202d8\233\167|7y\t_S\EM\248\EOT\216\172\167d\181\224)\137\ACKo\197j#c\217\243\228.\167\228\205\144\vr\134"
 
-syntaxTests :: Spec
-syntaxTests = do
+syntaxTests :: ATransport -> Spec
+syntaxTests (ATransport t) = do
   it "unknown command" $ ("", "abcd", "1234", "HELLO") >#> ("", "abcd", "1234", "ERR CMD SYNTAX")
   describe "NEW" do
     it "no parameters" $ (sampleSig, "bcda", "", "NEW") >#> ("", "bcda", "", "ERR CMD SYNTAX")
@@ -414,3 +411,5 @@ syntaxTests = do
       it "wrong terminator" $ (sampleSig, "bcda", "12345678", cmd <> "=") >#> ("", "bcda", "12345678", "ERR CMD SYNTAX")
       it "no signature" $ ("", "cdab", "12345678", cmd) >#> ("", "cdab", "12345678", "ERR CMD NO_AUTH")
       it "no queue ID" $ (sampleSig, "dabc", "", cmd) >#> ("", "dabc", "", "ERR CMD NO_AUTH")
+    (>#>) :: RawTransmission -> RawTransmission -> Expectation
+    command >#> response = smpServerTest t command `shouldReturn` response

tests/Test.hs

@@ -1,6 +1,10 @@
+{-# LANGUAGE TypeApplications #-}
+
 import AgentTests
 import ProtocolErrorTests
 import ServerTests
+import Simplex.Messaging.Transport (TCP, Transport (..))
+import Simplex.Messaging.Transport.WebSockets (WS)
 import System.Directory (createDirectoryIfMissing, removeDirectoryRecursive)
 import Test.Hspec
 
@@ -9,6 +13,7 @@ main = do
   createDirectoryIfMissing False "tests/tmp"
   hspec $ do
     describe "Protocol errors" protocolErrorTests
-    describe "SMP server" serverTests
-    describe "SMP client agent" agentTests
+    describe "SMP server via TCP" $ serverTests (transport @TCP)
+    describe "SMP server via WebSockets" $ serverTests (transport @WS)
+    describe "SMP client agent" $ agentTests (transport @TCP)
   removeDirectoryRecursive "tests/tmp"