From 1f61267308a11bb00458e0da21376a2d750c8c49 Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin <2769109+epoberezkin@users.noreply.github.com> Date: Sun, 31 Jan 2021 17:29:16 +0000 Subject: [PATCH] chat prototype (#35) * chat prototype * chat prototype now compiles * chat prototype works * agent: respond SENT mId to SEND (instead of OK), ne repsonse to chat message in terminal * chat prototype help, update commands * chat CLI options * add active contact to ChatClient (not used yet) * refactor agentTransmission * InviteContact -> AddContact * automatically insert active contact * highlight contact in chat * name for invitations * do not ask name on start * change default server to smp.simplex.im --- .gitignore | 2 + apps/dog-food/ChatOptions.hs | 56 ++++ apps/dog-food/Main.hs | 257 ++++++++++++++++++ package.yaml | 10 + src/Simplex/Messaging/Agent.hs | 30 +- src/Simplex/Messaging/Agent/Env/SQLite.hs | 4 +- .../Messaging/Agent/Store/SQLite/Util.hs | 8 +- src/Simplex/Messaging/Agent/Transmission.hs | 4 + tests/AgentTests.hs | 33 ++- tests/SMPAgentClient.hs | 2 +- 10 files changed, 374 insertions(+), 32 deletions(-) create mode 100644 apps/dog-food/ChatOptions.hs create mode 100644 apps/dog-food/Main.hs diff --git a/.gitignore b/.gitignore index 18235a6b5..a4d295219 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,5 @@ *.lock *.cabal smp-agent.db +smp-chat.db +smp-chat1.db diff --git a/apps/dog-food/ChatOptions.hs b/apps/dog-food/ChatOptions.hs new file mode 100644 index 000000000..0f1a36652 --- /dev/null +++ b/apps/dog-food/ChatOptions.hs @@ -0,0 +1,56 @@ +module ChatOptions (getChatOpts, ChatOpts (..)) where + +import qualified Data.Attoparsec.ByteString.Char8 as A +import qualified Data.ByteString.Char8 as B +import Options.Applicative +import Simplex.Messaging.Agent.Transmission (SMPServer (..), smpServerP) + +data ChatOpts = ChatOpts + { name :: Maybe B.ByteString, + dbFileName :: String, + smpServer :: SMPServer + } + +chatOpts :: Parser ChatOpts +chatOpts = + ChatOpts + <$> option + parseName + ( long "name" + <> short 'n' + <> metavar "NAME" + <> help "optional name to use for invitations" + <> value Nothing + ) + <*> strOption + ( long "database" + <> short 'd' + <> metavar "DB_FILE" + <> help "sqlite database filename (smp-chat.db)" + <> value "smp-chat.db" + ) + <*> option + parseSMPServer + ( long "server" + <> short 's' + <> metavar "SERVER" + <> help "SMP server to use (localhost:5223)" + <> value (SMPServer "smp.simplex.im" (Just "5223") Nothing) + ) + +parseName :: ReadM (Maybe B.ByteString) +parseName = maybeReader $ Just . Just . B.pack + +parseSMPServer :: ReadM SMPServer +parseSMPServer = eitherReader $ A.parseOnly (smpServerP <* A.endOfInput) . B.pack + +getChatOpts :: IO ChatOpts +getChatOpts = execParser opts + where + opts = + info + (chatOpts <**> helper) + ( fullDesc + <> header "Chat prototype using Simplex Messaging Protocol (SMP)" + <> progDesc "Start chat with DB_FILE file and use SERVER as SMP server" + ) diff --git a/apps/dog-food/Main.hs b/apps/dog-food/Main.hs new file mode 100644 index 000000000..61bf9d81c --- /dev/null +++ b/apps/dog-food/Main.hs @@ -0,0 +1,257 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DuplicateRecordFields #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module Main where + +import ChatOptions +import Control.Applicative ((<|>)) +import Control.Concurrent.STM +import Control.Logger.Simple +import Control.Monad.Reader +import Data.Attoparsec.ByteString.Char8 (Parser) +import qualified Data.Attoparsec.ByteString.Char8 as A +import Data.ByteString.Char8 (ByteString) +import qualified Data.ByteString.Char8 as B +import Data.Functor (($>)) +import qualified Data.Text as T +import Data.Text.Encoding +import Numeric.Natural +import Simplex.Messaging.Agent (getSMPAgentClient, runSMPAgentClient) +import Simplex.Messaging.Agent.Client (AgentClient (..)) +import Simplex.Messaging.Agent.Env.SQLite +import Simplex.Messaging.Agent.Transmission +import Simplex.Messaging.Client (smpDefaultConfig) +import Simplex.Messaging.Transport (getLn, putLn) +import Simplex.Messaging.Util (bshow, raceAny_) +import qualified System.Console.ANSI as C +import System.IO + +cfg :: AgentConfig +cfg = + AgentConfig + { tcpPort = undefined, -- TODO maybe take it out of config + tbqSize = 16, + connIdBytes = 12, + dbFile = "smp-chat.db", + smpCfg = smpDefaultConfig + } + +logCfg :: LogConfig +logCfg = LogConfig {lc_file = Nothing, lc_stderr = True} + +data ChatClient = ChatClient + { inQ :: TBQueue ChatCommand, + outQ :: TBQueue ChatResponse, + smpServer :: SMPServer, + activeContact :: TVar (Maybe Contact), + username :: TVar (Maybe Contact) + } + +newtype Contact = Contact {toBs :: ByteString} + +-- | GroupMessage ChatGroup ByteString +-- | AddToGroup Contact +data ChatCommand + = ChatHelp + | AddContact Contact + | AcceptContact Contact SMPQueueInfo + | ChatWith Contact + | SetName Contact + | SendMessage Contact ByteString + +chatCommandP :: Parser ChatCommand +chatCommandP = + "/help" $> ChatHelp + <|> "/add " *> (AddContact <$> contact) + <|> "/accept " *> acceptContact + <|> "/chat " *> chatWith + <|> "/name " *> setName + <|> "@" *> sendMessage + where + acceptContact = AcceptContact <$> contact <* A.space <*> smpQueueInfoP + chatWith = ChatWith <$> contact + setName = SetName <$> contact + sendMessage = SendMessage <$> contact <* A.space <*> A.takeByteString + contact = Contact <$> A.takeTill (== ' ') + +data ChatResponse + = ChatHelpInfo + | Invitation SMPQueueInfo + | Connected Contact + | ReceivedMessage Contact ByteString + | Disconnected Contact + | YesYes + | ErrorInput ByteString + | ChatError AgentErrorType + | NoChatResponse + +serializeChatResponse :: Maybe Contact -> ChatResponse -> ByteString +serializeChatResponse name = \case + ChatHelpInfo -> chatHelpInfo + Invitation qInfo -> "ask your contact to enter: /accept " <> showName name <> " " <> serializeSmpQueueInfo qInfo + Connected c -> ttyContact c <> " connected" + ReceivedMessage c t -> ttyContact c <> ": " <> t + Disconnected c -> "disconnected from " <> ttyContact c <> " - try \"/chat " <> toBs c <> "\"" + YesYes -> "you got it!" + ErrorInput t -> "invalid input: " <> t + ChatError e -> "chat error: " <> bshow e + NoChatResponse -> "" + where + showName Nothing = "" + showName (Just (Contact a)) = a + +chatHelpInfo :: ByteString +chatHelpInfo = + "Using chat:\n\ + \/add - create invitation to send out-of-band\n\ + \ to your contact \n\ + \ (any unique string without spaces)\n\ + \/accept - accept \n\ + \ (a string that starts from \"smp::\")\n\ + \ from your contact \n\ + \/chat - resume chat with \n\ + \/name - set to use in invitations\n\ + \@ - send (any string) to contact \n\ + \ @ can be omitted to send to previous" + +main :: IO () +main = do + ChatOpts {dbFileName, smpServer, name} <- getChatOpts + putStrLn "simpleX chat prototype (no encryption), \"/help\" for usage information" + t <- getChatClient smpServer (Contact <$> name) + -- setLogLevel LogInfo -- LogError + -- withGlobalLogging logCfg $ + env <- newSMPAgentEnv cfg {dbFile = dbFileName} + dogFoodChat t env + +dogFoodChat :: ChatClient -> Env -> IO () +dogFoodChat t env = do + c <- runReaderT getSMPAgentClient env + raceAny_ + [ runReaderT (runSMPAgentClient c) env, + sendToAgent t c, + sendToTTY t, + receiveFromAgent t c, + receiveFromTTY t + ] + +getChatClient :: SMPServer -> Maybe Contact -> IO ChatClient +getChatClient srv name = atomically $ newChatClient (tbqSize cfg) srv name + +newChatClient :: Natural -> SMPServer -> Maybe Contact -> STM ChatClient +newChatClient qSize smpServer name = do + inQ <- newTBQueue qSize + outQ <- newTBQueue qSize + activeContact <- newTVar Nothing + username <- newTVar name + return ChatClient {inQ, outQ, smpServer, activeContact, username} + +receiveFromTTY :: ChatClient -> IO () +receiveFromTTY t = + forever $ getChatLn t >>= processOrError . A.parseOnly (chatCommandP <* A.endOfInput) + where + processOrError = \case + Left err -> atomically . writeTBQueue (outQ t) . ErrorInput $ B.pack err + Right ChatHelp -> atomically . writeTBQueue (outQ t) $ ChatHelpInfo + Right (SetName a) -> atomically $ do + writeTVar (username t) $ Just a + writeTBQueue (outQ t) YesYes + Right cmd -> atomically $ writeTBQueue (inQ t) cmd + +sendToTTY :: ChatClient -> IO () +sendToTTY ChatClient {outQ, username} = forever $ do + atomically (readTBQueue outQ) >>= \case + NoChatResponse -> return () + resp -> do + name <- readTVarIO username + putLn stdout $ serializeChatResponse name resp + +sendToAgent :: ChatClient -> AgentClient -> IO () +sendToAgent ChatClient {inQ, smpServer, activeContact} AgentClient {rcvQ} = + forever . atomically $ do + cmd <- readTBQueue inQ + writeTBQueue rcvQ `mapM_` agentTransmission cmd + setActiveContact cmd + where + setActiveContact :: ChatCommand -> STM () + setActiveContact cmd = + writeTVar activeContact $ case cmd of + ChatWith a -> Just a + SendMessage a _ -> Just a + _ -> Nothing + agentTransmission :: ChatCommand -> Maybe (ATransmission 'Client) + agentTransmission = \case + AddContact a -> transmission a $ NEW smpServer + AcceptContact a qInfo -> transmission a $ JOIN qInfo $ ReplyVia smpServer + ChatWith a -> transmission a SUB + SendMessage a msg -> transmission a $ SEND msg + ChatHelp -> Nothing + SetName _ -> Nothing + transmission :: Contact -> ACommand 'Client -> Maybe (ATransmission 'Client) + transmission (Contact a) cmd = Just ("1", a, cmd) + +receiveFromAgent :: ChatClient -> AgentClient -> IO () +receiveFromAgent t c = forever . atomically $ do + resp <- chatResponse <$> readTBQueue (sndQ c) + writeTBQueue (outQ t) resp + setActiveContact resp + where + chatResponse :: ATransmission 'Agent -> ChatResponse + chatResponse (_, a, resp) = case resp of + INV qInfo -> Invitation qInfo + CON -> Connected $ Contact a + END -> Disconnected $ Contact a + MSG {m_body} -> ReceivedMessage (Contact a) m_body + SENT _ -> NoChatResponse + OK -> YesYes + ERR e -> ChatError e + setActiveContact :: ChatResponse -> STM () + setActiveContact = \case + Connected a -> set $ Just a + ReceivedMessage a _ -> set $ Just a + Disconnected _ -> set Nothing + _ -> return () + where + set a = writeTVar (activeContact t) a + +getChatLn :: ChatClient -> IO ByteString +getChatLn t = do + setTTY NoBuffering + getChar >>= \case + '/' -> getRest "/" + '@' -> getRest "@" + ch -> do + let s = encodeUtf8 $ T.singleton ch + readTVarIO (activeContact t) >>= \case + Nothing -> getRest s + Just a -> getWithContact a s + where + getWithContact :: Contact -> ByteString -> IO ByteString + getWithContact a s = do + C.cursorBackward 1 + B.hPut stdout $ " " <> ttyContact a <> " " <> s + getRest $ "@" <> toBs a <> " " <> s + getRest :: ByteString -> IO ByteString + getRest s = do + setTTY LineBuffering + (s <>) <$> getLn stdin + +setTTY :: BufferMode -> IO () +setTTY mode = do + hSetBuffering stdin mode + hSetBuffering stdout mode + +ttyContact :: Contact -> ByteString +ttyContact (Contact a) = withSGR contactSGR $ "@" <> a + +contactSGR :: [C.SGR] +contactSGR = [C.SetColor C.Foreground C.Vivid C.Cyan] + +withSGR :: [C.SGR] -> ByteString -> ByteString +withSGR sgr s = B.pack (C.setSGRCode sgr) <> s <> B.pack (C.setSGRCode [C.Reset]) diff --git a/package.yaml b/package.yaml index 12fc0c43b..d4b74b400 100644 --- a/package.yaml +++ b/package.yaml @@ -52,6 +52,16 @@ executables: ghc-options: - -threaded + dog-food: + source-dirs: apps/dog-food + main: Main.hs + dependencies: + - ansi-terminal == 0.10.* + - optparse-applicative == 0.15.* + - simplex-messaging + ghc-options: + - -threaded + tests: smp-server-test: source-dirs: tests diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 3dc902285..bd4156fd1 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -8,7 +8,12 @@ {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} -module Simplex.Messaging.Agent (runSMPAgent) where +module Simplex.Messaging.Agent + ( runSMPAgent, + getSMPAgentClient, + runSMPAgentClient, + ) +where import Control.Logger.Simple import Control.Monad.Except @@ -38,20 +43,22 @@ import UnliftIO.IO import UnliftIO.STM runSMPAgent :: (MonadRandom m, MonadUnliftIO m) => AgentConfig -> m () -runSMPAgent cfg@AgentConfig {tcpPort} = do - env <- newEnv cfg - runReaderT smpAgent env +runSMPAgent cfg@AgentConfig {tcpPort} = runReaderT smpAgent =<< newSMPAgentEnv cfg where smpAgent :: (MonadUnliftIO m', MonadReader Env m') => m' () smpAgent = runTCPServer tcpPort $ \h -> do liftIO $ putLn h "Welcome to SMP v0.2.0 agent" - q <- asks $ tbqSize . config - n <- asks clientCounter - c <- atomically $ newAgentClient n q + c <- getSMPAgentClient logConnection c True - race_ (connectClient h c) (runClient c) + race_ (connectClient h c) (runSMPAgentClient c) `E.finally` (closeSMPServerClients c >> logConnection c False) +getSMPAgentClient :: (MonadUnliftIO m, MonadReader Env m) => m AgentClient +getSMPAgentClient = do + q <- asks $ tbqSize . config + n <- asks clientCounter + atomically $ newAgentClient n q + connectClient :: MonadUnliftIO m => Handle -> AgentClient -> m () connectClient h c = race_ (send h c) (receive h c) @@ -60,8 +67,8 @@ logConnection c connected = let event = if connected then "connected to" else "disconnected from" in logInfo $ T.unwords ["client", showText (clientId c), event, "Agent"] -runClient :: (MonadUnliftIO m, MonadReader Env m) => AgentClient -> m () -runClient c = race_ (subscriber c) (client c) +runSMPAgentClient :: (MonadUnliftIO m, MonadReader Env m) => AgentClient -> m () +runSMPAgentClient c = race_ (subscriber c) (client c) receive :: forall m. MonadUnliftIO m => Handle -> AgentClient -> m () receive h c@AgentClient {rcvQ, sndQ} = forever $ do @@ -157,7 +164,8 @@ processCommand c@AgentClient {sndQ} (corrId, connAlias, cmd) = sendMsg sq = do sendAgentMessage c sq $ A_MSG msgBody -- TODO respond $ SENT aMsgId - respond OK + -- TODO send message to DB + respond $ SENT 0 suspendConnection :: m () suspendConnection = diff --git a/src/Simplex/Messaging/Agent/Env/SQLite.hs b/src/Simplex/Messaging/Agent/Env/SQLite.hs index 79e1615b3..22a4d0120 100644 --- a/src/Simplex/Messaging/Agent/Env/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Env/SQLite.hs @@ -29,8 +29,8 @@ data Env = Env clientCounter :: TVar Int } -newEnv :: (MonadUnliftIO m, MonadRandom m) => AgentConfig -> m Env -newEnv config = do +newSMPAgentEnv :: (MonadUnliftIO m, MonadRandom m) => AgentConfig -> m Env +newSMPAgentEnv config = do idsDrg <- drgNew >>= newTVarIO db <- newSQLiteStore $ dbFile config clientCounter <- newTVarIO 0 diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Util.hs b/src/Simplex/Messaging/Agent/Store/SQLite/Util.hs index 1a2415df1..9db84ffca 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Util.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Util.hs @@ -80,15 +80,15 @@ withLock st tableLock f = do (f $ conn st) insertWithLock :: (MonadUnliftIO m, ToRow q) => SQLiteStore -> (SQLiteStore -> TMVar ()) -> DB.Query -> q -> m Int64 -insertWithLock st tableLock queryStr q = do +insertWithLock st tableLock queryStr q = withLock st tableLock $ \c -> liftIO $ do DB.execute c queryStr q DB.lastInsertRowId c executeWithLock :: (MonadUnliftIO m, ToRow q) => SQLiteStore -> (SQLiteStore -> TMVar ()) -> DB.Query -> q -> m () -executeWithLock st tableLock queryStr q = do - withLock st tableLock $ \c -> liftIO $ do - DB.execute c queryStr q +executeWithLock st tableLock queryStr q = + withLock st tableLock $ \c -> + liftIO $ DB.execute c queryStr q instance ToRow SMPServer where toRow SMPServer {host, port, keyHash} = toRow (host, port, keyHash) diff --git a/src/Simplex/Messaging/Agent/Transmission.hs b/src/Simplex/Messaging/Agent/Transmission.hs index ea117ea76..3645764d6 100644 --- a/src/Simplex/Messaging/Agent/Transmission.hs +++ b/src/Simplex/Messaging/Agent/Transmission.hs @@ -79,6 +79,7 @@ data ACommand (p :: AParty) where -- QST :: QueueDirection -> ACommand Client -- STAT :: QueueDirection -> Maybe QueueStatus -> Maybe SubMode -> ACommand Agent SEND :: MsgBody -> ACommand Client + SENT :: AgentMsgId -> ACommand Agent MSG :: { m_recipient :: (AgentMsgId, UTCTime), m_broker :: (ST.MsgId, UTCTime), @@ -292,6 +293,7 @@ parseCommandP = <|> "SUB" $> ACmd SClient SUB <|> "END" $> ACmd SAgent END <|> "SEND " *> sendCmd + <|> "SENT " *> sentResp <|> "MSG " *> message <|> "OFF" $> ACmd SClient OFF <|> "DEL" $> ACmd SClient DEL @@ -303,6 +305,7 @@ parseCommandP = invResp = ACmd SAgent . INV <$> smpQueueInfoP joinCmd = ACmd SClient <$> (JOIN <$> smpQueueInfoP <*> replyMode) sendCmd = ACmd SClient <$> (SEND <$> A.takeByteString) + sentResp = ACmd SAgent <$> (SENT <$> A.decimal) message = do m_status <- status <* A.space m_recipient <- "R=" *> partyMeta A.decimal @@ -335,6 +338,7 @@ serializeCommand = \case SUB -> "SUB" END -> "END" SEND msgBody -> "SEND " <> serializeMsg msgBody + SENT mId -> "SENT " <> bshow mId MSG {m_recipient = (rmId, rTs), m_broker = (bmId, bTs), m_sender = (smId, sTs), m_status, m_body} -> B.unwords [ "MSG", diff --git a/tests/AgentTests.hs b/tests/AgentTests.hs index 74049b258..adedb24ea 100644 --- a/tests/AgentTests.hs +++ b/tests/AgentTests.hs @@ -53,16 +53,21 @@ action #> (corrId, cAlias, cmd) = action `shouldReturn` (corrId, cAlias, Right c -- | action and predicate for the response -- `h #:t =#> p` is the test that sends `t` to `h` and validates the response using `p` -(=#>) :: IO (ATransmissionOrError 'Agent) -> (ATransmissionOrError 'Agent -> Bool) -> Expectation -action =#> p = action >>= (`shouldSatisfy` p) +(=#>) :: IO (ATransmissionOrError 'Agent) -> (ATransmission 'Agent -> Bool) -> Expectation +action =#> p = action >>= (`shouldSatisfy` p . correctTransmission) + +correctTransmission :: ATransmissionOrError a -> ATransmission a +correctTransmission (corrId, cAlias, cmdOrErr) = case cmdOrErr of + Right cmd -> (corrId, cAlias, cmd) + Left e -> error $ show e -- | receive message to handle `h` and validate that it is the expected one (<#) :: Handle -> 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 -> (ATransmissionOrError 'Agent -> Bool) -> Expectation -h <#= p = tGet SAgent h >>= (`shouldSatisfy` p) +(<#=) :: Handle -> (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 @@ -73,8 +78,8 @@ h #:# err = tryGet `shouldReturn` () Just _ -> error err _ -> return () -pattern Msg :: MsgBody -> Either AgentErrorType (ACommand 'Agent) -pattern Msg m_body <- Right MSG {m_body} +pattern Msg :: MsgBody -> ACommand 'Agent +pattern Msg m_body <- MSG {m_body} testDuplexConnection :: Handle -> Handle -> IO () testDuplexConnection alice bob = do @@ -82,13 +87,13 @@ testDuplexConnection alice bob = do let qInfo' = serializeSmpQueueInfo qInfo bob #: ("11", "alice", "JOIN " <> qInfo') #> ("11", "alice", CON) alice <# ("", "bob", CON) - alice #: ("2", "bob", "SEND :hello") #> ("2", "bob", OK) - alice #: ("3", "bob", "SEND :how are you?") #> ("3", "bob", OK) + alice #: ("2", "bob", "SEND :hello") =#> \case ("2", "bob", SENT _) -> True; _ -> False + alice #: ("3", "bob", "SEND :how are you?") =#> \case ("3", "bob", SENT _) -> True; _ -> False bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False bob <#= \case ("", "alice", Msg "how are you?") -> True; _ -> False - bob #: ("14", "alice", "SEND 9\nhello too") #> ("14", "alice", OK) + bob #: ("14", "alice", "SEND 9\nhello too") =#> \case ("14", "alice", SENT _) -> True; _ -> False alice <#= \case ("", "bob", Msg "hello too") -> True; _ -> False - bob #: ("15", "alice", "SEND 9\nmessage 1") #> ("15", "alice", OK) + bob #: ("15", "alice", "SEND 9\nmessage 1") =#> \case ("15", "alice", SENT _) -> True; _ -> False alice <#= \case ("", "bob", Msg "message 1") -> True; _ -> False alice #: ("5", "bob", "OFF") #> ("5", "bob", OK) bob #: ("17", "alice", "SEND 9\nmessage 3") #> ("17", "alice", ERR (SMP AUTH)) @@ -100,20 +105,20 @@ testSubscription alice1 alice2 bob = do ("1", "bob", Right (INV qInfo)) <- alice1 #: ("1", "bob", "NEW localhost:5000") let qInfo' = serializeSmpQueueInfo qInfo bob #: ("11", "alice", "JOIN " <> qInfo') #> ("11", "alice", CON) - bob #: ("12", "alice", "SEND 5\nhello") #> ("12", "alice", OK) - bob #: ("13", "alice", "SEND 11\nhello again") #> ("13", "alice", OK) + bob #: ("12", "alice", "SEND 5\nhello") =#> \case ("12", "alice", SENT _) -> True; _ -> False + bob #: ("13", "alice", "SEND 11\nhello again") =#> \case ("13", "alice", SENT _) -> True; _ -> False alice1 <# ("", "bob", CON) alice1 <#= \case ("", "bob", Msg "hello") -> True; _ -> False alice1 <#= \case ("", "bob", Msg "hello again") -> True; _ -> False alice2 #: ("21", "bob", "SUB") #> ("21", "bob", OK) alice1 <# ("", "bob", END) - bob #: ("14", "alice", "SEND 2\nhi") #> ("14", "alice", OK) + bob #: ("14", "alice", "SEND 2\nhi") =#> \case ("14", "alice", SENT _) -> True; _ -> False alice2 <#= \case ("", "bob", Msg "hi") -> True; _ -> False alice1 #:# "nothing else should be delivered to alice1" testSubscrNotification :: (ThreadId, ThreadId) -> Handle -> IO () testSubscrNotification (server, _) client = do - client #: ("1", "conn1", "NEW localhost:5000") =#> \case ("1", "conn1", Right (INV _)) -> True; _ -> False + client #: ("1", "conn1", "NEW localhost:5000") =#> \case ("1", "conn1", INV _) -> True; _ -> False client #:# "nothing should be delivered to client before the server is killed" killThread server client <# ("", "conn1", END) diff --git a/tests/SMPAgentClient.hs b/tests/SMPAgentClient.hs index ee67ef8f9..49ba0df47 100644 --- a/tests/SMPAgentClient.hs +++ b/tests/SMPAgentClient.hs @@ -134,7 +134,7 @@ withSmpAgent = withSmpAgentOn (agentTestPort, testDB) testSMPAgentClientOn :: MonadUnliftIO m => ServiceName -> (Handle -> m a) -> m a testSMPAgentClientOn port' client = do - threadDelay 100_000 -- TODO hack: thread delay for SMP agent to start + threadDelay 200_000 -- TODO hack: thread delay for SMP agent to start runTCPClient agentTestHost port' $ \h -> do line <- liftIO $ getLn h if line == "Welcome to SMP v0.2.0 agent"