mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-10-23 00:02:44 +00:00
move modules to folder Simplex.Messaging.Server
This commit is contained in:
Evgeny Poberezkin committed
1 parent
64362fe013
commit
3cb2421373
11 files changed
+34
-34
No files matched your search
@@ -0,0 +1,279 @@
|
||||
{-# LANGUAGE DataKinds #-}
|
||||
{-# LANGUAGE DuplicateRecordFields #-}
|
||||
{-# LANGUAGE FlexibleContexts #-}
|
||||
{-# LANGUAGE GADTs #-}
|
||||
{-# LANGUAGE KindSignatures #-}
|
||||
{-# LANGUAGE LambdaCase #-}
|
||||
{-# LANGUAGE NamedFieldPuns #-}
|
||||
{-# LANGUAGE OverloadedStrings #-}
|
||||
{-# LANGUAGE RankNTypes #-}
|
||||
{-# LANGUAGE ScopedTypeVariables #-}
|
||||
{-# LANGUAGE TupleSections #-}
|
||||
|
||||
module Simplex.Messaging.Server (runSMPServer) where
|
||||
|
||||
import Control.Concurrent.STM (stateTVar)
|
||||
import Control.Monad
|
||||
import Control.Monad.IO.Unlift
|
||||
import Control.Monad.Reader
|
||||
import Crypto.Random
|
||||
import Data.ByteString.Char8 (ByteString)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
import Data.Functor (($>))
|
||||
import qualified Data.Map.Strict as M
|
||||
import Data.Time.Clock
|
||||
import Simplex.Messaging.Server.Env.STM
|
||||
import Simplex.Messaging.Server.MsgStore
|
||||
import Simplex.Messaging.Server.MsgStore.STM (MsgQueue)
|
||||
import Simplex.Messaging.Server.QueueStore
|
||||
import Simplex.Messaging.Server.QueueStore.STM (QueueStore)
|
||||
import Simplex.Messaging.Server.Transmission
|
||||
import Simplex.Messaging.Transport
|
||||
import UnliftIO.Async
|
||||
import UnliftIO.Concurrent
|
||||
import UnliftIO.Exception
|
||||
import UnliftIO.IO
|
||||
import UnliftIO.STM
|
||||
|
||||
runSMPServer :: (MonadRandom m, MonadUnliftIO m) => Config -> m ()
|
||||
runSMPServer cfg@Config {tcpPort} = do
|
||||
env <- newEnv cfg
|
||||
runReaderT smpServer env
|
||||
where
|
||||
smpServer :: (MonadUnliftIO m, MonadReader Env m) => m ()
|
||||
smpServer = do
|
||||
s <- asks server
|
||||
race_ (runTCPServer tcpPort runClient) (serverThread s)
|
||||
|
||||
serverThread :: MonadUnliftIO m => Server -> m ()
|
||||
serverThread Server {subscribedQ, subscribers} = forever . atomically $ do
|
||||
(rId, clnt) <- readTBQueue subscribedQ
|
||||
cs <- readTVar subscribers
|
||||
case M.lookup rId cs of
|
||||
Just Client {rcvQ} -> writeTBQueue rcvQ (rId, Cmd SBroker END)
|
||||
Nothing -> return ()
|
||||
writeTVar subscribers $ M.insert rId clnt cs
|
||||
|
||||
runClient :: (MonadUnliftIO m, MonadReader Env m) => Handle -> m ()
|
||||
runClient h = do
|
||||
putLn h "Welcome to SMP"
|
||||
q <- asks $ tbqSize . config
|
||||
c <- atomically $ newClient q
|
||||
s <- asks server
|
||||
raceAny_ [send h c, client c s, receive h c]
|
||||
`finally` cancelSubscribers c
|
||||
|
||||
cancelSubscribers :: (MonadUnliftIO m) => Client -> m ()
|
||||
cancelSubscribers Client {subscriptions} =
|
||||
readTVarIO subscriptions >>= mapM_ cancelSub
|
||||
|
||||
cancelSub :: (MonadUnliftIO m) => Sub -> m ()
|
||||
cancelSub = \case
|
||||
Sub {subThread = SubThread t} -> killThread t
|
||||
_ -> return ()
|
||||
|
||||
raceAny_ :: MonadUnliftIO m => [m a] -> m ()
|
||||
raceAny_ = r []
|
||||
where
|
||||
r as (m : ms) = withAsync m $ \a -> r (a : as) ms
|
||||
r as [] = void $ waitAnyCancel as
|
||||
|
||||
receive :: (MonadUnliftIO m, MonadReader Env m) => Handle -> Client -> m ()
|
||||
receive h Client {rcvQ} = forever $ do
|
||||
(signature, (queueId, cmdOrError)) <- tGet fromClient h
|
||||
signed <- case cmdOrError of
|
||||
Left e -> return . mkResp queueId $ ERR e
|
||||
Right cmd -> verifyTransmission signature queueId cmd
|
||||
atomically $ writeTBQueue rcvQ signed
|
||||
|
||||
send :: MonadUnliftIO m => Handle -> Client -> m ()
|
||||
send h Client {sndQ} = forever $ do
|
||||
signed <- atomically $ readTBQueue sndQ
|
||||
tPut h (B.empty, signed)
|
||||
|
||||
mkResp :: QueueId -> Command 'Broker -> Signed
|
||||
mkResp queueId command = (queueId, Cmd SBroker command)
|
||||
|
||||
verifyTransmission :: forall m. (MonadUnliftIO m, MonadReader Env m) => Signature -> QueueId -> Cmd -> m Signed
|
||||
verifyTransmission signature queueId cmd = do
|
||||
(queueId,) <$> case cmd of
|
||||
Cmd SBroker _ -> return $ smpErr INTERNAL -- it can only be client command, because `fromClient` was used
|
||||
Cmd SRecipient (NEW _) -> return cmd
|
||||
Cmd SRecipient _ -> withQueueRec SRecipient $ verifySignature . recipientKey
|
||||
Cmd SSender (SEND _) -> withQueueRec SSender $ verifySend . senderKey
|
||||
where
|
||||
withQueueRec :: SParty (p :: Party) -> (QueueRec -> m Cmd) -> m Cmd
|
||||
withQueueRec party f = do
|
||||
st <- asks queueStore
|
||||
qr <- atomically $ getQueue st party queueId
|
||||
either (return . smpErr) f qr
|
||||
verifySend :: Maybe PublicKey -> m Cmd
|
||||
verifySend
|
||||
| B.null signature = return . maybe cmd (const authErr)
|
||||
| otherwise = maybe (return authErr) verifySignature
|
||||
-- TODO stub
|
||||
verifySignature :: PublicKey -> m Cmd
|
||||
verifySignature key = return $ if signature == key then cmd else authErr
|
||||
|
||||
smpErr e = Cmd SBroker $ ERR e
|
||||
authErr = smpErr AUTH
|
||||
|
||||
client :: forall m. (MonadUnliftIO m, MonadReader Env m) => Client -> Server -> m ()
|
||||
client clnt@Client {subscriptions, rcvQ, sndQ} Server {subscribedQ} =
|
||||
forever $
|
||||
atomically (readTBQueue rcvQ)
|
||||
>>= processCommand
|
||||
>>= atomically . writeTBQueue sndQ
|
||||
where
|
||||
processCommand :: Signed -> m Signed
|
||||
processCommand (queueId, cmd) = do
|
||||
st <- asks queueStore
|
||||
case cmd of
|
||||
Cmd SBroker END -> unsubscribeQueue $> (queueId, cmd)
|
||||
Cmd SBroker _ -> return (queueId, cmd)
|
||||
Cmd SSender (SEND msgBody) -> sendMessage st msgBody
|
||||
Cmd SRecipient command -> case command of
|
||||
NEW rKey -> createQueue st rKey
|
||||
SUB -> subscribeQueue queueId
|
||||
ACK -> acknowledgeMsg
|
||||
KEY sKey -> okResp <$> atomically (secureQueue st queueId sKey)
|
||||
OFF -> okResp <$> atomically (suspendQueue st queueId)
|
||||
DEL -> delQueueAndMsgs st
|
||||
where
|
||||
createQueue :: QueueStore -> RecipientKey -> m Signed
|
||||
createQueue st rKey = mkResp B.empty <$> addSubscribe
|
||||
where
|
||||
addSubscribe =
|
||||
addQueueRetry 3 >>= \case
|
||||
Left e -> return $ ERR e
|
||||
Right (rId, sId) -> subscribeQueue rId $> IDS rId sId
|
||||
|
||||
addQueueRetry :: Int -> m (Either ErrorType (RecipientId, SenderId))
|
||||
addQueueRetry 0 = return $ Left INTERNAL
|
||||
addQueueRetry n = do
|
||||
ids <- getIds
|
||||
atomically (addQueue st rKey ids) >>= \case
|
||||
Left DUPLICATE -> addQueueRetry $ n - 1
|
||||
Left e -> return $ Left e
|
||||
Right _ -> return $ Right ids
|
||||
|
||||
getIds :: m (RecipientId, SenderId)
|
||||
getIds = do
|
||||
n <- asks $ queueIdBytes . config
|
||||
liftM2 (,) (randomId n) (randomId n)
|
||||
|
||||
subscribeQueue :: RecipientId -> m Signed
|
||||
subscribeQueue rId =
|
||||
atomically (getSubscription rId) >>= deliverMessage tryPeekMsg rId
|
||||
|
||||
getSubscription :: RecipientId -> STM Sub
|
||||
getSubscription rId = do
|
||||
subs <- readTVar subscriptions
|
||||
case M.lookup rId subs of
|
||||
Just s -> tryTakeTMVar (delivered s) $> s
|
||||
Nothing -> do
|
||||
writeTBQueue subscribedQ (rId, clnt)
|
||||
s <- newSubscription
|
||||
writeTVar subscriptions $ M.insert rId s subs
|
||||
return s
|
||||
|
||||
unsubscribeQueue :: m ()
|
||||
unsubscribeQueue = do
|
||||
sub <- atomically . stateTVar subscriptions $
|
||||
\cs -> (M.lookup queueId cs, M.delete queueId cs)
|
||||
mapM_ cancelSub sub
|
||||
|
||||
acknowledgeMsg :: m Signed
|
||||
acknowledgeMsg =
|
||||
atomically (withSub queueId $ \s -> const s <$$> tryTakeTMVar (delivered s))
|
||||
>>= \case
|
||||
Just (Just s) -> deliverMessage tryDelPeekMsg queueId s
|
||||
_ -> return $ err PROHIBITED
|
||||
|
||||
withSub :: RecipientId -> (Sub -> STM a) -> STM (Maybe a)
|
||||
withSub rId f = readTVar subscriptions >>= mapM f . M.lookup rId
|
||||
|
||||
sendMessage :: QueueStore -> MsgBody -> m Signed
|
||||
sendMessage st msgBody = do
|
||||
qr <- atomically $ getQueue st SSender queueId
|
||||
either (return . err) storeMessage qr
|
||||
where
|
||||
mkMessage :: m Message
|
||||
mkMessage = do
|
||||
msgId <- asks (msgIdBytes . config) >>= randomId
|
||||
ts <- liftIO getCurrentTime
|
||||
return $ Message {msgId, ts, msgBody}
|
||||
|
||||
storeMessage :: QueueRec -> m Signed
|
||||
storeMessage qr = case status qr of
|
||||
QueueOff -> return $ err AUTH
|
||||
QueueActive -> do
|
||||
ms <- asks msgStore
|
||||
msg <- mkMessage
|
||||
atomically $ do
|
||||
q <- getMsgQueue ms (recipientId qr)
|
||||
writeMsg q msg
|
||||
return ok
|
||||
|
||||
deliverMessage :: (MsgQueue -> STM (Maybe Message)) -> RecipientId -> Sub -> m Signed
|
||||
deliverMessage tryPeek rId = \case
|
||||
Sub {subThread = NoSub} -> do
|
||||
ms <- asks msgStore
|
||||
q <- atomically $ getMsgQueue ms rId
|
||||
atomically (tryPeek q) >>= \case
|
||||
Nothing -> forkSub q $> ok
|
||||
Just msg -> atomically setDelivered $> msgResp rId msg
|
||||
_ -> return ok
|
||||
where
|
||||
forkSub :: MsgQueue -> m ()
|
||||
forkSub q = do
|
||||
atomically . setSub $ \s -> s {subThread = SubPending}
|
||||
t <- forkIO $ subscriber q
|
||||
atomically . setSub $ \case
|
||||
s@Sub {subThread = SubPending} -> s {subThread = SubThread t}
|
||||
s -> s
|
||||
|
||||
subscriber :: MsgQueue -> m ()
|
||||
subscriber q = atomically $ do
|
||||
msg <- peekMsg q
|
||||
writeTBQueue sndQ $ msgResp rId msg
|
||||
setSub (\s -> s {subThread = NoSub})
|
||||
void setDelivered
|
||||
|
||||
setSub :: (Sub -> Sub) -> STM ()
|
||||
setSub f = modifyTVar subscriptions $ M.adjust f rId
|
||||
|
||||
setDelivered :: STM (Maybe Bool)
|
||||
setDelivered = withSub rId $ \s -> tryPutTMVar (delivered s) ()
|
||||
|
||||
delQueueAndMsgs :: QueueStore -> m Signed
|
||||
delQueueAndMsgs st = do
|
||||
ms <- asks msgStore
|
||||
atomically $
|
||||
deleteQueue st queueId >>= \case
|
||||
Left e -> return $ err e
|
||||
Right _ -> delMsgQueue ms queueId $> ok
|
||||
|
||||
ok :: Signed
|
||||
ok = mkResp queueId OK
|
||||
|
||||
err :: ErrorType -> Signed
|
||||
err = mkResp queueId . ERR
|
||||
|
||||
okResp :: Either ErrorType () -> Signed
|
||||
okResp = either err $ const ok
|
||||
|
||||
msgResp :: RecipientId -> Message -> Signed
|
||||
msgResp rId Message {msgId, ts, msgBody} = mkResp rId $ MSG msgId ts msgBody
|
||||
|
||||
randomId :: (MonadUnliftIO m, MonadReader Env m) => Int -> m Encoded
|
||||
randomId n = do
|
||||
gVar <- asks idsDrg
|
||||
atomically (randomBytes n gVar)
|
||||
|
||||
randomBytes :: Int -> TVar ChaChaDRG -> STM ByteString
|
||||
randomBytes n gVar = do
|
||||
g <- readTVar gVar
|
||||
let (bytes, g') = randomBytesGenerate n g
|
||||
writeTVar gVar g'
|
||||
return bytes
|
||||
@@ -0,0 +1,75 @@
|
||||
{-# LANGUAGE DuplicateRecordFields #-}
|
||||
{-# LANGUAGE NamedFieldPuns #-}
|
||||
|
||||
module Simplex.Messaging.Server.Env.STM where
|
||||
|
||||
import Control.Concurrent (ThreadId)
|
||||
import Control.Monad.IO.Unlift
|
||||
import Crypto.Random
|
||||
import Data.Map.Strict (Map)
|
||||
import qualified Data.Map.Strict as M
|
||||
import Network.Socket (ServiceName)
|
||||
import Numeric.Natural
|
||||
import Simplex.Messaging.Server.MsgStore.STM
|
||||
import Simplex.Messaging.Server.QueueStore.STM
|
||||
import Simplex.Messaging.Server.Transmission
|
||||
import UnliftIO.STM
|
||||
|
||||
data Config = Config
|
||||
{ tcpPort :: ServiceName,
|
||||
tbqSize :: Natural,
|
||||
queueIdBytes :: Int,
|
||||
msgIdBytes :: Int
|
||||
}
|
||||
|
||||
data Env = Env
|
||||
{ config :: Config,
|
||||
server :: Server,
|
||||
queueStore :: QueueStore,
|
||||
msgStore :: STMMsgStore,
|
||||
idsDrg :: TVar ChaChaDRG
|
||||
}
|
||||
|
||||
data Server = Server
|
||||
{ subscribedQ :: TBQueue (RecipientId, Client),
|
||||
subscribers :: TVar (Map RecipientId Client)
|
||||
}
|
||||
|
||||
data Client = Client
|
||||
{ subscriptions :: TVar (Map RecipientId Sub),
|
||||
rcvQ :: TBQueue Signed,
|
||||
sndQ :: TBQueue Signed
|
||||
}
|
||||
|
||||
data SubscriptionThread = NoSub | SubPending | SubThread ThreadId
|
||||
|
||||
data Sub = Sub
|
||||
{ subThread :: SubscriptionThread,
|
||||
delivered :: TMVar ()
|
||||
}
|
||||
|
||||
newServer :: Natural -> STM Server
|
||||
newServer qSize = do
|
||||
subscribedQ <- newTBQueue qSize
|
||||
subscribers <- newTVar M.empty
|
||||
return Server {subscribedQ, subscribers}
|
||||
|
||||
newClient :: Natural -> STM Client
|
||||
newClient qSize = do
|
||||
subscriptions <- newTVar M.empty
|
||||
rcvQ <- newTBQueue qSize
|
||||
sndQ <- newTBQueue qSize
|
||||
return Client {subscriptions, rcvQ, sndQ}
|
||||
|
||||
newSubscription :: STM Sub
|
||||
newSubscription = do
|
||||
delivered <- newEmptyTMVar
|
||||
return Sub {subThread = NoSub, delivered}
|
||||
|
||||
newEnv :: (MonadUnliftIO m, MonadRandom m) => Config -> m Env
|
||||
newEnv config = do
|
||||
server <- atomically $ newServer (tbqSize config)
|
||||
queueStore <- atomically newQueueStore
|
||||
msgStore <- atomically newMsgStore
|
||||
idsDrg <- drgNew >>= newTVarIO
|
||||
return Env {config, server, queueStore, msgStore, idsDrg}
|
||||
@@ -0,0 +1,22 @@
|
||||
{-# LANGUAGE FunctionalDependencies #-}
|
||||
|
||||
module Simplex.Messaging.Server.MsgStore where
|
||||
|
||||
import Data.Time.Clock
|
||||
import Simplex.Messaging.Server.Transmission
|
||||
|
||||
data Message = Message
|
||||
{ msgId :: Encoded,
|
||||
ts :: UTCTime,
|
||||
msgBody :: MsgBody
|
||||
}
|
||||
|
||||
class MonadMsgStore s q m | s -> q where
|
||||
getMsgQueue :: s -> RecipientId -> m q
|
||||
delMsgQueue :: s -> RecipientId -> m ()
|
||||
|
||||
class MonadMsgQueue q m where
|
||||
writeMsg :: q -> Message -> m () -- non blocking
|
||||
tryPeekMsg :: q -> m (Maybe Message) -- non blocking
|
||||
peekMsg :: q -> m Message -- blocking
|
||||
tryDelPeekMsg :: q -> m (Maybe Message) -- atomic delete (== read) last and peek next message, if available
|
||||
@@ -0,0 +1,51 @@
|
||||
{-# LANGUAGE ConstraintKinds #-}
|
||||
{-# LANGUAGE FlexibleContexts #-}
|
||||
{-# LANGUAGE FlexibleInstances #-}
|
||||
{-# LANGUAGE InstanceSigs #-}
|
||||
{-# LANGUAGE MultiParamTypeClasses #-}
|
||||
|
||||
module Simplex.Messaging.Server.MsgStore.STM where
|
||||
|
||||
import Data.Map.Strict (Map)
|
||||
import qualified Data.Map.Strict as M
|
||||
import Simplex.Messaging.Server.MsgStore
|
||||
import Simplex.Messaging.Server.Transmission
|
||||
import UnliftIO.STM
|
||||
|
||||
newtype MsgQueue = MsgQueue {msgQueue :: TQueue Message}
|
||||
|
||||
newtype MsgStoreData = MsgStoreData {messages :: Map RecipientId MsgQueue}
|
||||
|
||||
type STMMsgStore = TVar MsgStoreData
|
||||
|
||||
newMsgStore :: STM STMMsgStore
|
||||
newMsgStore = newTVar $ MsgStoreData M.empty
|
||||
|
||||
instance MonadMsgStore STMMsgStore MsgQueue STM where
|
||||
getMsgQueue :: STMMsgStore -> RecipientId -> STM MsgQueue
|
||||
getMsgQueue store rId = do
|
||||
m <- messages <$> readTVar store
|
||||
maybe (newQ m) return $ M.lookup rId m
|
||||
where
|
||||
newQ m' = do
|
||||
q <- MsgQueue <$> newTQueue
|
||||
writeTVar store . MsgStoreData $ M.insert rId q m'
|
||||
return q
|
||||
|
||||
delMsgQueue :: STMMsgStore -> RecipientId -> STM ()
|
||||
delMsgQueue store rId =
|
||||
modifyTVar store $ MsgStoreData . M.delete rId . messages
|
||||
|
||||
instance MonadMsgQueue MsgQueue STM where
|
||||
writeMsg :: MsgQueue -> Message -> STM ()
|
||||
writeMsg = writeTQueue . msgQueue
|
||||
|
||||
tryPeekMsg :: MsgQueue -> STM (Maybe Message)
|
||||
tryPeekMsg = tryPeekTQueue . msgQueue
|
||||
|
||||
peekMsg :: MsgQueue -> STM Message
|
||||
peekMsg = peekTQueue . msgQueue
|
||||
|
||||
-- atomic delete (== read) last and peek next message if available
|
||||
tryDelPeekMsg :: MsgQueue -> STM (Maybe Message)
|
||||
tryDelPeekMsg (MsgQueue q) = tryReadTQueue q >> tryPeekTQueue q
|
||||
@@ -0,0 +1,35 @@
|
||||
{-# LANGUAGE DataKinds #-}
|
||||
{-# LANGUAGE KindSignatures #-}
|
||||
{-# LANGUAGE MultiParamTypeClasses #-}
|
||||
{-# LANGUAGE NamedFieldPuns #-}
|
||||
|
||||
module Simplex.Messaging.Server.QueueStore where
|
||||
|
||||
import Simplex.Messaging.Server.Transmission
|
||||
|
||||
data QueueRec = QueueRec
|
||||
{ recipientId :: QueueId,
|
||||
senderId :: QueueId,
|
||||
recipientKey :: PublicKey,
|
||||
senderKey :: Maybe PublicKey,
|
||||
status :: QueueStatus
|
||||
}
|
||||
|
||||
data QueueStatus = QueueActive | QueueOff
|
||||
|
||||
class MonadQueueStore s m where
|
||||
addQueue :: s -> RecipientKey -> (RecipientId, SenderId) -> m (Either ErrorType ())
|
||||
getQueue :: s -> SParty (a :: Party) -> QueueId -> m (Either ErrorType QueueRec)
|
||||
secureQueue :: s -> RecipientId -> SenderKey -> m (Either ErrorType ())
|
||||
suspendQueue :: s -> RecipientId -> m (Either ErrorType ())
|
||||
deleteQueue :: s -> RecipientId -> m (Either ErrorType ())
|
||||
|
||||
mkQueueRec :: RecipientKey -> (RecipientId, SenderId) -> QueueRec
|
||||
mkQueueRec recipientKey (recipientId, senderId) =
|
||||
QueueRec
|
||||
{ recipientId,
|
||||
senderId,
|
||||
recipientKey,
|
||||
senderKey = Nothing,
|
||||
status = QueueActive
|
||||
}
|
||||
@@ -0,0 +1,92 @@
|
||||
{-# LANGUAGE DataKinds #-}
|
||||
{-# LANGUAGE FlexibleInstances #-}
|
||||
{-# LANGUAGE GADTs #-}
|
||||
{-# LANGUAGE InstanceSigs #-}
|
||||
{-# LANGUAGE KindSignatures #-}
|
||||
{-# LANGUAGE MultiParamTypeClasses #-}
|
||||
{-# LANGUAGE NamedFieldPuns #-}
|
||||
{-# LANGUAGE RankNTypes #-}
|
||||
{-# LANGUAGE ScopedTypeVariables #-}
|
||||
{-# LANGUAGE UndecidableInstances #-}
|
||||
|
||||
module Simplex.Messaging.Server.QueueStore.STM where
|
||||
|
||||
import Data.Map.Strict (Map)
|
||||
import qualified Data.Map.Strict as M
|
||||
import Simplex.Messaging.Server.QueueStore
|
||||
import Simplex.Messaging.Server.Transmission
|
||||
import UnliftIO.STM
|
||||
|
||||
data QueueStoreData = QueueStoreData
|
||||
{ queues :: Map RecipientId QueueRec,
|
||||
senders :: Map SenderId RecipientId
|
||||
}
|
||||
|
||||
type QueueStore = TVar QueueStoreData
|
||||
|
||||
newQueueStore :: STM QueueStore
|
||||
newQueueStore = newTVar QueueStoreData {queues = M.empty, senders = M.empty}
|
||||
|
||||
instance MonadQueueStore QueueStore STM where
|
||||
addQueue :: QueueStore -> RecipientKey -> (RecipientId, SenderId) -> STM (Either ErrorType ())
|
||||
addQueue store rKey ids@(rId, sId) = do
|
||||
cs@QueueStoreData {queues, senders} <- readTVar store
|
||||
if M.member rId queues || M.member sId senders
|
||||
then return $ Left DUPLICATE
|
||||
else do
|
||||
writeTVar store $
|
||||
cs
|
||||
{ queues = M.insert rId (mkQueueRec rKey ids) queues,
|
||||
senders = M.insert sId rId senders
|
||||
}
|
||||
return $ Right ()
|
||||
|
||||
getQueue :: QueueStore -> SParty (p :: Party) -> QueueId -> STM (Either ErrorType QueueRec)
|
||||
getQueue store SRecipient rId = do
|
||||
cs <- readTVar store
|
||||
return $ getRcpQueue cs rId
|
||||
getQueue store SSender sId = do
|
||||
cs <- readTVar store
|
||||
let rId = M.lookup sId $ senders cs
|
||||
return $ maybe (Left AUTH) (getRcpQueue cs) rId
|
||||
getQueue _ SBroker _ =
|
||||
return $ Left INTERNAL
|
||||
|
||||
secureQueue store rId sKey =
|
||||
updateQueues store rId $ \cs c ->
|
||||
case senderKey c of
|
||||
Just _ -> (Left AUTH, cs)
|
||||
_ -> (Right (), cs {queues = M.insert rId c {senderKey = Just sKey} (queues cs)})
|
||||
|
||||
suspendQueue :: QueueStore -> RecipientId -> STM (Either ErrorType ())
|
||||
suspendQueue store rId =
|
||||
updateQueues store rId $ \cs c ->
|
||||
(Right (), cs {queues = M.insert rId c {status = QueueOff} (queues cs)})
|
||||
|
||||
deleteQueue :: QueueStore -> RecipientId -> STM (Either ErrorType ())
|
||||
deleteQueue store rId =
|
||||
updateQueues store rId $ \cs c ->
|
||||
( Right (),
|
||||
cs
|
||||
{ queues = M.delete rId (queues cs),
|
||||
senders = M.delete (senderId c) (senders cs)
|
||||
}
|
||||
)
|
||||
|
||||
updateQueues ::
|
||||
QueueStore ->
|
||||
RecipientId ->
|
||||
(QueueStoreData -> QueueRec -> (Either ErrorType (), QueueStoreData)) ->
|
||||
STM (Either ErrorType ())
|
||||
updateQueues store rId update = do
|
||||
cs <- readTVar store
|
||||
let conn = getRcpQueue cs rId
|
||||
either (return . Left) (_update cs) conn
|
||||
where
|
||||
_update cs c = do
|
||||
let (res, cs') = update cs c
|
||||
writeTVar store cs'
|
||||
return res
|
||||
|
||||
getRcpQueue :: QueueStoreData -> RecipientId -> Either ErrorType QueueRec
|
||||
getRcpQueue cs rId = maybe (Left AUTH) Right . M.lookup rId $ queues cs
|
||||
@@ -0,0 +1,176 @@
|
||||
{-# LANGUAGE DataKinds #-}
|
||||
{-# LANGUAGE GADTs #-}
|
||||
{-# LANGUAGE LambdaCase #-}
|
||||
{-# LANGUAGE OverloadedStrings #-}
|
||||
{-# LANGUAGE ScopedTypeVariables #-}
|
||||
{-# LANGUAGE StandaloneDeriving #-}
|
||||
{-# LANGUAGE TypeFamilies #-}
|
||||
{-# LANGUAGE UndecidableInstances #-}
|
||||
{-# OPTIONS_GHC -fno-warn-unticked-promoted-constructors #-}
|
||||
|
||||
module Simplex.Messaging.Server.Transmission where
|
||||
|
||||
import Data.ByteString.Base64
|
||||
import Data.ByteString.Char8 (ByteString)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
import Data.Char (ord)
|
||||
import Data.Kind
|
||||
import Data.Time.Clock
|
||||
import Data.Time.ISO8601
|
||||
|
||||
data Party = Broker | Recipient | Sender
|
||||
deriving (Show)
|
||||
|
||||
data SParty :: Party -> Type where
|
||||
SBroker :: SParty Broker
|
||||
SRecipient :: SParty Recipient
|
||||
SSender :: SParty Sender
|
||||
|
||||
deriving instance Show (SParty a)
|
||||
|
||||
data Cmd where
|
||||
Cmd :: SParty a -> Command a -> Cmd
|
||||
|
||||
deriving instance Show Cmd
|
||||
|
||||
type Signed = (QueueId, Cmd)
|
||||
|
||||
type Transmission = (Signature, Signed)
|
||||
|
||||
type SignedOrError = (QueueId, Either ErrorType Cmd)
|
||||
|
||||
type TransmissionOrError = (Signature, SignedOrError)
|
||||
|
||||
type RawTransmission = (ByteString, ByteString, ByteString)
|
||||
|
||||
data Command (a :: Party) where
|
||||
NEW :: RecipientKey -> Command Recipient
|
||||
SUB :: Command Recipient
|
||||
KEY :: SenderKey -> Command Recipient
|
||||
ACK :: Command Recipient
|
||||
OFF :: Command Recipient
|
||||
DEL :: Command Recipient
|
||||
SEND :: MsgBody -> Command Sender
|
||||
IDS :: RecipientId -> SenderId -> Command Broker
|
||||
MSG :: MsgId -> UTCTime -> MsgBody -> Command Broker
|
||||
END :: Command Broker
|
||||
OK :: Command Broker
|
||||
ERR :: ErrorType -> Command Broker
|
||||
|
||||
deriving instance Show (Command a)
|
||||
|
||||
deriving instance Eq (Command a)
|
||||
|
||||
parseCommand :: ByteString -> Either ErrorType Cmd
|
||||
parseCommand command = case B.words command of
|
||||
["NEW", rKeyStr] -> case decode rKeyStr of
|
||||
Right rKey -> rCmd $ NEW rKey
|
||||
_ -> errParams
|
||||
["SUB"] -> rCmd SUB
|
||||
["KEY", sKeyStr] -> case decode sKeyStr of
|
||||
Right sKey -> rCmd $ KEY sKey
|
||||
_ -> errParams
|
||||
["ACK"] -> rCmd ACK
|
||||
["OFF"] -> rCmd OFF
|
||||
["DEL"] -> rCmd DEL
|
||||
["SEND"] -> errParams
|
||||
"SEND" : msgBody -> Right . Cmd SSender . SEND $ B.unwords msgBody
|
||||
["IDS", rIdStr, sIdStr] -> case decode rIdStr of
|
||||
Right rId -> case decode sIdStr of
|
||||
Right sId -> bCmd $ IDS rId sId
|
||||
_ -> errParams
|
||||
_ -> errParams
|
||||
["MSG", msgIdStr, ts, msgBody] -> case decode msgIdStr of
|
||||
Right msgId -> case parseISO8601 $ B.unpack ts of
|
||||
Just utc -> bCmd $ MSG msgId utc msgBody
|
||||
_ -> errParams
|
||||
_ -> errParams
|
||||
["END"] -> bCmd END
|
||||
["OK"] -> bCmd OK
|
||||
"ERR" : err -> case err of
|
||||
["UNKNOWN"] -> bErr UNKNOWN
|
||||
["PROHIBITED"] -> bErr PROHIBITED
|
||||
["SYNTAX", errCode] -> maybe errParams (bErr . SYNTAX) $ digitToInt $ B.unpack errCode
|
||||
["SIZE"] -> bErr SIZE
|
||||
["AUTH"] -> bErr AUTH
|
||||
["INTERNAL"] -> bErr INTERNAL
|
||||
_ -> errParams
|
||||
"NEW" : _ -> errParams
|
||||
"SUB" : _ -> errParams
|
||||
"KEY" : _ -> errParams
|
||||
"ACK" : _ -> errParams
|
||||
"OFF" : _ -> errParams
|
||||
"DEL" : _ -> errParams
|
||||
"MSG" : _ -> errParams
|
||||
"IDS" : _ -> errParams
|
||||
"END" : _ -> errParams
|
||||
"OK" : _ -> errParams
|
||||
_ -> Left UNKNOWN
|
||||
where
|
||||
errParams = Left $ SYNTAX errBadParameters
|
||||
rCmd = Right . Cmd SRecipient
|
||||
bCmd = Right . Cmd SBroker
|
||||
bErr = bCmd . ERR
|
||||
|
||||
digitToInt :: String -> Maybe Int
|
||||
digitToInt [c] =
|
||||
let i = ord c - zero
|
||||
in if i >= 0 && i <= 9 then Just i else Nothing
|
||||
digitToInt _ = Nothing
|
||||
|
||||
zero :: Int
|
||||
zero = ord '0'
|
||||
|
||||
serializeCommand :: Cmd -> ByteString
|
||||
serializeCommand = \case
|
||||
Cmd SRecipient (NEW rKey) -> "NEW " <> encode rKey
|
||||
Cmd SRecipient (KEY sKey) -> "KEY " <> encode sKey
|
||||
Cmd SRecipient cmd -> B.pack $ show cmd
|
||||
Cmd SSender (SEND msgBody) -> "SEND" <> serializeMsg msgBody
|
||||
Cmd SBroker (MSG msgId ts msgBody) ->
|
||||
B.unwords ["MSG", encode msgId, B.pack $ formatISO8601Millis ts] <> serializeMsg msgBody
|
||||
Cmd SBroker (IDS rId sId) -> B.unwords ["IDS", encode rId, encode sId]
|
||||
Cmd SBroker (ERR err) -> "ERR " <> B.pack (show err)
|
||||
Cmd SBroker resp -> B.pack $ show resp
|
||||
where
|
||||
serializeMsg msgBody = " " <> B.pack (show $ B.length msgBody) <> "\n" <> msgBody
|
||||
|
||||
type Encoded = ByteString
|
||||
|
||||
type PublicKey = Encoded
|
||||
|
||||
type Signature = Encoded
|
||||
|
||||
type RecipientKey = PublicKey
|
||||
|
||||
type SenderKey = PublicKey
|
||||
|
||||
type RecipientId = QueueId
|
||||
|
||||
type SenderId = QueueId
|
||||
|
||||
type QueueId = Encoded
|
||||
|
||||
type MsgId = Encoded
|
||||
|
||||
type MsgBody = ByteString
|
||||
|
||||
data ErrorType = UNKNOWN | PROHIBITED | SYNTAX Int | SIZE | AUTH | INTERNAL | DUPLICATE deriving (Show, Eq)
|
||||
|
||||
errBadTransmission :: Int
|
||||
errBadTransmission = 1
|
||||
|
||||
errBadParameters :: Int
|
||||
errBadParameters = 2
|
||||
|
||||
errNoCredentials :: Int
|
||||
errNoCredentials = 3
|
||||
|
||||
errHasCredentials :: Int
|
||||
errHasCredentials = 4
|
||||
|
||||
errNoConnectionId :: Int
|
||||
errNoConnectionId = 5
|
||||
|
||||
errMessageBody :: Int
|
||||
errMessageBody = 6
|
||||
@@ -0,0 +1,167 @@
|
||||
{-# LANGUAGE AllowAmbiguousTypes #-}
|
||||
{-# LANGUAGE BlockArguments #-}
|
||||
{-# LANGUAGE FlexibleContexts #-}
|
||||
{-# LANGUAGE GADTs #-}
|
||||
{-# LANGUAGE LambdaCase #-}
|
||||
{-# LANGUAGE ScopedTypeVariables #-}
|
||||
{-# LANGUAGE TupleSections #-}
|
||||
|
||||
module Simplex.Messaging.Transport where
|
||||
|
||||
import Control.Monad.IO.Class
|
||||
import Control.Monad.IO.Unlift
|
||||
import Control.Monad.Reader
|
||||
import Data.ByteString.Base64
|
||||
import Data.ByteString.Char8 (ByteString)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
import Network.Socket
|
||||
import Simplex.Messaging.Server.Transmission
|
||||
import System.IO
|
||||
import Text.Read
|
||||
import UnliftIO.Concurrent
|
||||
import qualified UnliftIO.Exception as E
|
||||
import qualified UnliftIO.IO as IO
|
||||
|
||||
startTCPServer :: MonadIO m => ServiceName -> m Socket
|
||||
startTCPServer port = liftIO . withSocketsDo $ resolve >>= open
|
||||
where
|
||||
resolve = do
|
||||
let hints = defaultHints {addrFlags = [AI_PASSIVE], addrSocketType = Stream}
|
||||
head <$> getAddrInfo (Just hints) Nothing (Just port)
|
||||
open addr = do
|
||||
sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)
|
||||
setSocketOption sock ReuseAddr 1
|
||||
withFdSocket sock setCloseOnExecIfNeeded
|
||||
bind sock $ addrAddress addr
|
||||
listen sock 1024
|
||||
return sock
|
||||
|
||||
runTCPServer :: MonadUnliftIO m => ServiceName -> (Handle -> m ()) -> m ()
|
||||
runTCPServer port server =
|
||||
E.bracket (startTCPServer port) (liftIO . close) $ \sock -> forever $ do
|
||||
h <- acceptTCPConn sock
|
||||
forkFinally (server h) (const $ IO.hClose h)
|
||||
|
||||
acceptTCPConn :: MonadIO m => Socket -> m Handle
|
||||
acceptTCPConn sock = liftIO $ do
|
||||
(conn, _) <- accept sock
|
||||
-- putStrLn $ "Accepted connection from " ++ show peer
|
||||
getSocketHandle conn
|
||||
|
||||
startTCPClient :: MonadIO m => HostName -> ServiceName -> m Handle
|
||||
startTCPClient host port = liftIO . withSocketsDo $ resolve >>= open
|
||||
where
|
||||
resolve = do
|
||||
let hints = defaultHints {addrSocketType = Stream}
|
||||
head <$> getAddrInfo (Just hints) (Just host) (Just port)
|
||||
open addr = do
|
||||
sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)
|
||||
connect sock $ addrAddress addr
|
||||
getSocketHandle sock
|
||||
|
||||
runTCPClient :: MonadUnliftIO m => HostName -> ServiceName -> (Handle -> m a) -> m a
|
||||
runTCPClient host port = E.bracket (startTCPClient host port) IO.hClose
|
||||
|
||||
getSocketHandle :: MonadIO m => Socket -> m Handle
|
||||
getSocketHandle conn = liftIO $ do
|
||||
h <- socketToHandle conn ReadWriteMode
|
||||
hSetBinaryMode h True
|
||||
hSetNewlineMode h NewlineMode {inputNL = CRLF, outputNL = CRLF}
|
||||
hSetBuffering h LineBuffering
|
||||
return h
|
||||
|
||||
putLn :: MonadIO m => Handle -> ByteString -> m ()
|
||||
putLn h = liftIO . hPutStrLn h . B.unpack
|
||||
|
||||
getLn :: MonadIO m => Handle -> m ByteString
|
||||
getLn h = B.pack <$> liftIO (hGetLine h)
|
||||
|
||||
getBytes :: MonadIO m => Handle -> Int -> m ByteString
|
||||
getBytes h = liftIO . B.hGet h
|
||||
|
||||
tPutRaw :: MonadIO m => Handle -> RawTransmission -> m ()
|
||||
tPutRaw h (signature, queueId, command) = do
|
||||
putLn h (encode signature)
|
||||
putLn h (encode queueId)
|
||||
putLn h command
|
||||
|
||||
tGetRaw :: MonadIO m => Handle -> m (Either String RawTransmission)
|
||||
tGetRaw h = do
|
||||
signature <- decode <$> getLn h
|
||||
queueId <- decode <$> getLn h
|
||||
command <- getLn h
|
||||
return $ liftM2 (,,command) signature queueId
|
||||
|
||||
tPut :: MonadIO m => Handle -> Transmission -> m ()
|
||||
tPut h (signature, (queueId, command)) = tPutRaw h (signature, queueId, serializeCommand command)
|
||||
|
||||
fromClient :: Cmd -> Either ErrorType Cmd
|
||||
fromClient = \case
|
||||
Cmd SBroker _ -> Left PROHIBITED
|
||||
cmd -> Right cmd
|
||||
|
||||
fromServer :: Cmd -> Either ErrorType Cmd
|
||||
fromServer = \case
|
||||
cmd@(Cmd SBroker _) -> Right cmd
|
||||
_ -> Left PROHIBITED
|
||||
|
||||
-- | get client and server transmissions
|
||||
-- `fromParty` is used to limit allowed senders - `fromClient` or `fromServer` should be used
|
||||
tGet :: forall m. MonadIO m => (Cmd -> Either ErrorType Cmd) -> Handle -> m TransmissionOrError
|
||||
tGet fromParty h = tGetRaw h >>= either (const tError) tParseLoadBody
|
||||
where
|
||||
tError :: m TransmissionOrError
|
||||
tError = return (B.empty, (B.empty, Left $ SYNTAX errBadTransmission))
|
||||
|
||||
tParseLoadBody :: RawTransmission -> m TransmissionOrError
|
||||
tParseLoadBody t@(signature, queueId, command) = do
|
||||
let cmd = parseCommand command >>= fromParty >>= tCredentials t
|
||||
fullCmd <- either (return . Left) cmdWithMsgBody cmd
|
||||
return (signature, (queueId, fullCmd))
|
||||
|
||||
tCredentials :: RawTransmission -> Cmd -> Either ErrorType Cmd
|
||||
tCredentials (signature, queueId, _) cmd = case cmd of
|
||||
-- IDS response should not have queue ID
|
||||
Cmd SBroker (IDS _ _) -> Right cmd
|
||||
-- ERROR response does not always have queue ID
|
||||
Cmd SBroker (ERR _) -> Right cmd
|
||||
-- other responses must have queue ID
|
||||
Cmd SBroker _
|
||||
| B.null queueId -> Left $ SYNTAX errNoConnectionId
|
||||
| otherwise -> Right cmd
|
||||
-- CREATE must NOT have signature or queue ID
|
||||
Cmd SRecipient (NEW _)
|
||||
| B.null signature && B.null queueId -> Right cmd
|
||||
| otherwise -> Left $ SYNTAX errHasCredentials
|
||||
-- SEND must have queue ID, signature is not always required
|
||||
Cmd SSender (SEND _)
|
||||
| B.null queueId -> Left $ SYNTAX errNoConnectionId
|
||||
| otherwise -> Right cmd
|
||||
-- other client commands must have both signature and queue ID
|
||||
Cmd SRecipient _
|
||||
| B.null signature || B.null queueId -> Left $ SYNTAX errNoCredentials
|
||||
| otherwise -> Right cmd
|
||||
|
||||
cmdWithMsgBody :: Cmd -> m (Either ErrorType Cmd)
|
||||
cmdWithMsgBody = \case
|
||||
Cmd SSender (SEND body) ->
|
||||
Cmd SSender . SEND <$$> getMsgBody body
|
||||
Cmd SBroker (MSG msgId ts body) ->
|
||||
Cmd SBroker . MSG msgId ts <$$> getMsgBody body
|
||||
cmd -> return $ Right cmd
|
||||
|
||||
getMsgBody :: MsgBody -> m (Either ErrorType MsgBody)
|
||||
getMsgBody msgBody =
|
||||
case B.unpack msgBody of
|
||||
':' : body -> return . Right $ B.pack body
|
||||
str -> case readMaybe str :: Maybe Int of
|
||||
Just size -> do
|
||||
body <- getBytes h size
|
||||
s <- getLn h
|
||||
return if B.null s then Right body else Left SIZE
|
||||
Nothing -> return . Left $ SYNTAX errMessageBody
|
||||
|
||||
infixl 4 <$$>
|
||||
|
||||
(<$$>) :: (Functor f, Functor g) => (a -> b) -> f (g a) -> f (g b)
|
||||
(<$$>) = fmap . fmap
|
||||
Reference in new issue
Block a user