mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-10-23 10:52:28 +00:00
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
This commit is contained in:
1 parent
3192092349
commit
1f61267308
10 files changed
+374
-32
No files matched your search
@@ -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 =
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in new issue
Block a user