mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-10-23 17:22:34 +00:00
remove entities, remove broadcasts (#158)
This commit is contained in:
1 parent
ab89963f45
commit
bf5561c89c
7 files changed
+295
-811
No files matched your search
@@ -1,9 +0,0 @@
|
||||
CREATE TABLE broadcasts (
|
||||
broadcast_id BLOB NOT NULL PRIMARY KEY
|
||||
) WITHOUT ROWID;
|
||||
|
||||
CREATE TABLE broadcast_connections (
|
||||
broadcast_id BLOB NOT NULL REFERENCES broadcasts (broadcast_id) ON DELETE CASCADE,
|
||||
conn_alias BLOB NOT NULL REFERENCES connections (conn_alias),
|
||||
PRIMARY KEY (broadcast_id, conn_alias)
|
||||
) WITHOUT ROWID;
|
||||
+113
-159
@@ -109,10 +109,10 @@ receive h c@AgentClient {rcvQ, sndQ} = forever loop
|
||||
where
|
||||
loop :: m ()
|
||||
loop = do
|
||||
ATransmissionOrError corrId entity cmdOrErr <- tGet SClient h
|
||||
(corrId, connId, cmdOrErr) <- tGet SClient h
|
||||
case cmdOrErr of
|
||||
Right cmd -> write rcvQ $ ATransmission corrId entity cmd
|
||||
Left e -> write sndQ $ ATransmission corrId entity $ ERR e
|
||||
Right cmd -> write rcvQ (corrId, connId, cmd)
|
||||
Left e -> write sndQ (corrId, connId, ERR e)
|
||||
write :: TBQueue (ATransmission p) -> ATransmission p -> m ()
|
||||
write q t = do
|
||||
logClient c "-->" t
|
||||
@@ -125,17 +125,17 @@ send h c@AgentClient {sndQ} = forever $ do
|
||||
logClient c "<--" t
|
||||
|
||||
logClient :: MonadUnliftIO m => AgentClient -> ByteString -> ATransmission a -> m ()
|
||||
logClient AgentClient {clientId} dir (ATransmission corrId entity cmd) = do
|
||||
logInfo . decodeUtf8 $ B.unwords [bshow clientId, dir, "A :", corrId, serializeEntity entity, B.takeWhile (/= ' ') $ serializeCommand cmd]
|
||||
logClient AgentClient {clientId} dir (corrId, connId, cmd) = do
|
||||
logInfo . decodeUtf8 $ B.unwords [bshow clientId, dir, "A :", corrId, connId, B.takeWhile (/= ' ') $ serializeCommand cmd]
|
||||
|
||||
client :: forall m. (MonadFail m, MonadUnliftIO m, MonadReader Env m) => AgentClient -> SQLiteStore -> m ()
|
||||
client c@AgentClient {rcvQ, sndQ} st = forever loop
|
||||
where
|
||||
loop :: m ()
|
||||
loop = do
|
||||
t@(ATransmission corrId entity _) <- atomically $ readTBQueue rcvQ
|
||||
t@(corrId, connId, _) <- atomically $ readTBQueue rcvQ
|
||||
runExceptT (processCommand c st t) >>= \case
|
||||
Left e -> atomically . writeTBQueue sndQ $ ATransmission corrId entity (ERR e)
|
||||
Left e -> atomically $ writeTBQueue sndQ (corrId, connId, ERR e)
|
||||
Right _ -> pure ()
|
||||
|
||||
withStore ::
|
||||
@@ -157,36 +157,21 @@ withStore action = do
|
||||
SEConnDuplicate -> CONN DUPLICATE
|
||||
SEBadConnType CRcv -> CONN SIMPLEX
|
||||
SEBadConnType CSnd -> CONN SIMPLEX
|
||||
SEBcastNotFound -> BCAST B_NOT_FOUND
|
||||
SEBcastDuplicate -> BCAST B_DUPLICATE
|
||||
e -> INTERNAL $ show e
|
||||
|
||||
processCommand :: forall m. AgentMonad m => AgentClient -> SQLiteStore -> ATransmission 'Client -> m ()
|
||||
processCommand c st (ATransmission corrId entity cmd) = process c st corrId entity cmd
|
||||
where
|
||||
process = case entity of
|
||||
Conn _ -> processConnCommand
|
||||
Broadcast _ -> processBroadcastCommand
|
||||
_ -> unsupportedEntity
|
||||
|
||||
unsupportedEntity :: AgentMonad m => AgentClient -> SQLiteStore -> ACorrId -> Entity t -> ACommand 'Client c -> m ()
|
||||
unsupportedEntity c _ corrId entity _ =
|
||||
atomically . writeTBQueue (sndQ c) . ATransmission corrId entity . ERR $ CMD UNSUPPORTED
|
||||
|
||||
processConnCommand ::
|
||||
forall c m. (AgentMonad m, EntityCommand 'Conn_ c) => AgentClient -> SQLiteStore -> ACorrId -> Entity 'Conn_ -> ACommand 'Client c -> m ()
|
||||
processConnCommand c@AgentClient {sndQ} st corrId conn@(Conn connId) = \case
|
||||
processCommand c@AgentClient {sndQ} st (corrId, connId, cmd) = case cmd of
|
||||
NEW -> createNewConnection Nothing 0 >>= uncurry respond
|
||||
JOIN smpQueueInfo replyMode -> joinConnection smpQueueInfo replyMode Nothing 0 >> pure () -- >>= (`respond` OK)
|
||||
INTRO reEntity reInfo -> makeIntroduction reEntity reInfo
|
||||
ACPT inv eInfo -> acceptInvitation inv eInfo
|
||||
SUB -> subscribeConnection conn
|
||||
INTRO reConnId reInfo -> makeIntroduction reConnId reInfo
|
||||
ACPT invId connInfo -> acceptInvitation invId connInfo
|
||||
SUB -> subscribeConnection connId
|
||||
SUBALL -> subscribeAll
|
||||
SEND msgBody -> sendClientMessage c st corrId conn msgBody
|
||||
OFF -> suspendConnection conn
|
||||
DEL -> deleteConnection conn
|
||||
SEND msgBody -> sendMessage msgBody
|
||||
OFF -> suspendConnection
|
||||
DEL -> deleteConnection
|
||||
where
|
||||
createNewConnection :: Maybe InvitationId -> Int -> m (Entity 'Conn_, ACommand 'Agent 'INV_)
|
||||
createNewConnection :: Maybe InvitationId -> Int -> m (ConnId, ACommand 'Agent)
|
||||
createNewConnection viaInv connLevel = do
|
||||
-- TODO create connection alias if not passed
|
||||
-- make connId Maybe?
|
||||
@@ -196,7 +181,7 @@ processConnCommand c@AgentClient {sndQ} st corrId conn@(Conn connId) = \case
|
||||
let cData = ConnData {connId, viaInv, connLevel}
|
||||
connId' <- withStore $ createRcvConn st g cData rq
|
||||
addSubscription c rq connId'
|
||||
pure (Conn connId', INV qInfo)
|
||||
pure (connId', INV qInfo)
|
||||
|
||||
getSMPServer :: m SMPServer
|
||||
getSMPServer =
|
||||
@@ -207,7 +192,7 @@ processConnCommand c@AgentClient {sndQ} st corrId conn@(Conn connId) = \case
|
||||
i <- atomically . stateTVar gen $ randomR (0, L.length servers - 1)
|
||||
pure $ servers L.!! i
|
||||
|
||||
joinConnection :: SMPQueueInfo -> ReplyMode -> Maybe InvitationId -> Int -> m (Entity 'Conn_)
|
||||
joinConnection :: SMPQueueInfo -> ReplyMode -> Maybe InvitationId -> Int -> m ConnId
|
||||
joinConnection qInfo (ReplyMode replyMode) viaInv connLevel = do
|
||||
(sq, senderKey, verifyKey) <- newSendQueue qInfo
|
||||
g <- asks idsDrg
|
||||
@@ -215,73 +200,95 @@ processConnCommand c@AgentClient {sndQ} st corrId conn@(Conn connId) = \case
|
||||
connId' <- withStore $ createSndConn st g cData sq
|
||||
connectToSendQueue c st sq senderKey verifyKey
|
||||
when (replyMode == On) $ createReplyQueue connId' sq
|
||||
pure $ Conn connId'
|
||||
pure connId'
|
||||
|
||||
makeIntroduction :: IntroEntity -> EntityInfo -> m ()
|
||||
makeIntroduction (IE reEntity) reInfo = case reEntity of
|
||||
Conn reConn ->
|
||||
withStore ((,) <$> getConn st connId <*> getConn st reConn) >>= \case
|
||||
(SomeConn _ (DuplexConnection _ _ sq), SomeConn _ DuplexConnection {}) -> do
|
||||
g <- asks idsDrg
|
||||
introId <- withStore $ createIntro st g NewIntroduction {toConn = connId, reConn, reInfo}
|
||||
sendControlMessage c sq $ A_INTRO (IE (Conn introId)) reInfo
|
||||
respond conn OK
|
||||
_ -> throwError $ CONN SIMPLEX
|
||||
_ -> throwError $ CMD UNSUPPORTED
|
||||
makeIntroduction :: IntroId -> ConnInfo -> m ()
|
||||
makeIntroduction reConn reInfo =
|
||||
withStore ((,) <$> getConn st connId <*> getConn st reConn) >>= \case
|
||||
(SomeConn _ (DuplexConnection _ _ sq), SomeConn _ DuplexConnection {}) -> do
|
||||
g <- asks idsDrg
|
||||
introId <- withStore $ createIntro st g NewIntroduction {toConn = connId, reConn, reInfo}
|
||||
sendControlMessage c sq $ A_INTRO introId reInfo
|
||||
respond connId OK
|
||||
_ -> throwError $ CONN SIMPLEX
|
||||
|
||||
acceptInvitation :: IntroEntity -> EntityInfo -> m ()
|
||||
acceptInvitation (IE invEntity) eInfo = case invEntity of
|
||||
Conn invId -> do
|
||||
withStore (getInvitation st invId) >>= \case
|
||||
Invitation {viaConn, qInfo, externalIntroId, status = InvNew} ->
|
||||
withStore (getConn st viaConn) >>= \case
|
||||
SomeConn _ (DuplexConnection ConnData {connLevel} _ sq) -> case qInfo of
|
||||
Nothing -> do
|
||||
(conn', INV qInfo') <- createNewConnection (Just invId) (connLevel + 1)
|
||||
withStore $ addInvitationConn st invId $ fromConn conn'
|
||||
sendControlMessage c sq $ A_INV (Conn externalIntroId) qInfo' eInfo
|
||||
respond conn' OK
|
||||
Just qInfo' -> do
|
||||
conn' <- joinConnection qInfo' (ReplyMode On) (Just invId) (connLevel + 1)
|
||||
withStore $ addInvitationConn st invId $ fromConn conn'
|
||||
respond conn' OK
|
||||
_ -> throwError $ CONN SIMPLEX
|
||||
_ -> throwError $ CMD PROHIBITED
|
||||
_ -> throwError $ CMD UNSUPPORTED
|
||||
acceptInvitation :: InvitationId -> ConnInfo -> m ()
|
||||
acceptInvitation invId connInfo =
|
||||
withStore (getInvitation st invId) >>= \case
|
||||
Invitation {viaConn, qInfo, externalIntroId, status = InvNew} ->
|
||||
withStore (getConn st viaConn) >>= \case
|
||||
SomeConn _ (DuplexConnection ConnData {connLevel} _ sq) -> case qInfo of
|
||||
Nothing -> do
|
||||
(connId', INV qInfo') <- createNewConnection (Just invId) (connLevel + 1)
|
||||
withStore $ addInvitationConn st invId connId'
|
||||
sendControlMessage c sq $ A_INV externalIntroId qInfo' connInfo
|
||||
respond connId' OK
|
||||
Just qInfo' -> do
|
||||
connId' <- joinConnection qInfo' (ReplyMode On) (Just invId) (connLevel + 1)
|
||||
withStore $ addInvitationConn st invId connId'
|
||||
respond connId' OK
|
||||
_ -> throwError $ CONN SIMPLEX
|
||||
_ -> throwError $ CMD PROHIBITED
|
||||
|
||||
subscribeConnection :: Entity 'Conn_ -> m ()
|
||||
subscribeConnection conn'@(Conn cId) =
|
||||
subscribeConnection :: ConnId -> m ()
|
||||
subscribeConnection cId =
|
||||
withStore (getConn st cId) >>= \case
|
||||
SomeConn _ (DuplexConnection _ rq _) -> subscribe rq
|
||||
SomeConn _ (RcvConnection _ rq) -> subscribe rq
|
||||
_ -> throwError $ CONN SIMPLEX
|
||||
where
|
||||
subscribe rq = subscribeQueue c rq cId >> respond conn' OK
|
||||
subscribe rq = subscribeQueue c rq cId >> respond cId OK
|
||||
|
||||
-- TODO remove - hack for subscribing to all; respond' and parameterization of subscribeConnection are byproduct
|
||||
subscribeAll :: m ()
|
||||
subscribeAll = withStore (getAllConnIds st) >>= mapM_ (subscribeConnection . Conn)
|
||||
subscribeAll = withStore (getAllConnIds st) >>= mapM_ subscribeConnection
|
||||
|
||||
suspendConnection :: Entity 'Conn_ -> m ()
|
||||
suspendConnection (Conn cId) =
|
||||
withStore (getConn st cId) >>= \case
|
||||
sendMessage :: MsgBody -> m ()
|
||||
sendMessage msgBody =
|
||||
withStore (getConn st connId) >>= \case
|
||||
SomeConn _ (DuplexConnection _ _ sq) -> sendMsg sq
|
||||
SomeConn _ (SndConnection _ sq) -> sendMsg sq
|
||||
_ -> throwError $ CONN SIMPLEX
|
||||
where
|
||||
sendMsg :: SndQueue -> m ()
|
||||
sendMsg sq = do
|
||||
internalTs <- liftIO getCurrentTime
|
||||
(internalId, internalSndId, previousMsgHash) <- withStore $ updateSndIds st connId
|
||||
let msgStr =
|
||||
serializeSMPMessage
|
||||
SMPMessage
|
||||
{ senderMsgId = unSndId internalSndId,
|
||||
senderTimestamp = internalTs,
|
||||
previousMsgHash,
|
||||
agentMessage = A_MSG msgBody
|
||||
}
|
||||
msgHash = C.sha256Hash msgStr
|
||||
withStore $
|
||||
createSndMsg st connId $
|
||||
SndMsgData {internalId, internalSndId, internalTs, msgBody, internalHash = msgHash}
|
||||
sendAgentMessage c sq msgStr
|
||||
atomically $ writeTBQueue sndQ (corrId, connId, SENT $ unId internalId)
|
||||
|
||||
suspendConnection :: m ()
|
||||
suspendConnection =
|
||||
withStore (getConn st connId) >>= \case
|
||||
SomeConn _ (DuplexConnection _ rq _) -> suspend rq
|
||||
SomeConn _ (RcvConnection _ rq) -> suspend rq
|
||||
_ -> throwError $ CONN SIMPLEX
|
||||
where
|
||||
suspend rq = suspendQueue c rq >> respond conn OK
|
||||
suspend rq = suspendQueue c rq >> respond connId OK
|
||||
|
||||
deleteConnection :: Entity 'Conn_ -> m ()
|
||||
deleteConnection (Conn cId) =
|
||||
withStore (getConn st cId) >>= \case
|
||||
deleteConnection :: m ()
|
||||
deleteConnection =
|
||||
withStore (getConn st connId) >>= \case
|
||||
SomeConn _ (DuplexConnection _ rq _) -> delete rq
|
||||
SomeConn _ (RcvConnection _ rq) -> delete rq
|
||||
_ -> delConn
|
||||
where
|
||||
delConn = withStore (deleteConn st cId) >> respond conn OK
|
||||
delConn = withStore (deleteConn st connId) >> respond connId OK
|
||||
delete rq = do
|
||||
deleteQueue c rq
|
||||
removeSubscription c cId
|
||||
removeSubscription c connId
|
||||
delConn
|
||||
|
||||
createReplyQueue :: ConnId -> SndQueue -> m ()
|
||||
@@ -292,8 +299,8 @@ processConnCommand c@AgentClient {sndQ} st corrId conn@(Conn connId) = \case
|
||||
withStore $ upgradeSndConnToDuplex st cId rq
|
||||
sendControlMessage c sq $ REPLY qInfo
|
||||
|
||||
respond :: EntityCommand t c' => Entity t -> ACommand 'Agent c' -> m ()
|
||||
respond ent resp = atomically . writeTBQueue sndQ $ ATransmission corrId ent resp
|
||||
respond :: ConnId -> ACommand 'Agent -> m ()
|
||||
respond cId resp = atomically . writeTBQueue sndQ $ (corrId, cId, resp)
|
||||
|
||||
sendControlMessage :: AgentMonad m => AgentClient -> SndQueue -> AMessage -> m ()
|
||||
sendControlMessage c sq agentMessage = do
|
||||
@@ -306,57 +313,6 @@ sendControlMessage c sq agentMessage = do
|
||||
agentMessage
|
||||
}
|
||||
|
||||
sendClientMessage :: forall m. AgentMonad m => AgentClient -> SQLiteStore -> ACorrId -> Entity 'Conn_ -> MsgBody -> m ()
|
||||
sendClientMessage c st corrId (Conn cId) msgBody =
|
||||
withStore (getConn st cId) >>= \case
|
||||
SomeConn _ (DuplexConnection _ _ sq) -> sendMsg sq
|
||||
SomeConn _ (SndConnection _ sq) -> sendMsg sq
|
||||
_ -> throwError $ CONN SIMPLEX
|
||||
where
|
||||
sendMsg :: SndQueue -> m ()
|
||||
sendMsg sq = do
|
||||
internalTs <- liftIO getCurrentTime
|
||||
(internalId, internalSndId, previousMsgHash) <- withStore $ updateSndIds st cId
|
||||
let msgStr =
|
||||
serializeSMPMessage
|
||||
SMPMessage
|
||||
{ senderMsgId = unSndId internalSndId,
|
||||
senderTimestamp = internalTs,
|
||||
previousMsgHash,
|
||||
agentMessage = A_MSG msgBody
|
||||
}
|
||||
msgHash = C.sha256Hash msgStr
|
||||
withStore $
|
||||
createSndMsg st cId $
|
||||
SndMsgData {internalId, internalSndId, internalTs, msgBody, internalHash = msgHash}
|
||||
sendAgentMessage c sq msgStr
|
||||
atomically . writeTBQueue (sndQ c) $ ATransmission corrId (Conn cId) $ SENT (unId internalId)
|
||||
|
||||
processBroadcastCommand ::
|
||||
forall c m. (AgentMonad m, EntityCommand 'Broadcast_ c) => AgentClient -> SQLiteStore -> ACorrId -> Entity 'Broadcast_ -> ACommand 'Client c -> m ()
|
||||
processBroadcastCommand c st corrId bcast@(Broadcast bId) = \case
|
||||
NEW -> createNewBroadcast
|
||||
ADD (Conn cId) -> withStore (addBcastConn st bId cId) >> ok
|
||||
REM (Conn cId) -> withStore (removeBcastConn st bId cId) >> ok
|
||||
LS -> withStore (getBcast st bId) >>= respond bcast . MS . map Conn
|
||||
SEND msgBody -> withStore (getBcast st bId) >>= mapM_ (sendMsg msgBody) >> respond bcast (SENT 0)
|
||||
DEL -> withStore (deleteBcast st bId) >> ok
|
||||
where
|
||||
createNewBroadcast :: m ()
|
||||
createNewBroadcast = do
|
||||
g <- asks idsDrg
|
||||
bId' <- withStore $ createBcast st g bId
|
||||
respond (Broadcast bId') OK
|
||||
|
||||
sendMsg :: MsgBody -> ConnId -> m ()
|
||||
sendMsg msgBody cId = sendClientMessage c st corrId (Conn cId) msgBody
|
||||
|
||||
ok :: m ()
|
||||
ok = respond bcast OK
|
||||
|
||||
respond :: EntityCommand t c' => Entity t -> ACommand 'Agent c' -> m ()
|
||||
respond ent resp = atomically . writeTBQueue (sndQ c) $ ATransmission corrId ent resp
|
||||
|
||||
subscriber :: (MonadFail m, MonadUnliftIO m, MonadReader Env m) => AgentClient -> SQLiteStore -> m ()
|
||||
subscriber c@AgentClient {msgQ} st = forever $ do
|
||||
-- TODO this will only process messages and notifications
|
||||
@@ -370,7 +326,7 @@ processSMPTransmission c@AgentClient {sndQ} st (srv, rId, cmd) = do
|
||||
withStore (getRcvConn st srv rId) >>= \case
|
||||
SomeConn SCDuplex (DuplexConnection cData rq _) -> processSMP SCDuplex cData rq
|
||||
SomeConn SCRcv (RcvConnection cData rq) -> processSMP SCRcv cData rq
|
||||
_ -> atomically . writeTBQueue sndQ $ ATransmission "" (Conn "") (ERR $ CONN SIMPLEX)
|
||||
_ -> atomically $ writeTBQueue sndQ ("", "", ERR $ CONN NOT_FOUND)
|
||||
where
|
||||
processSMP :: SConnType c -> ConnData -> RcvQueue -> m ()
|
||||
processSMP cType ConnData {connId} rq@RcvQueue {status} =
|
||||
@@ -387,10 +343,10 @@ processSMPTransmission c@AgentClient {sndQ} st (srv, rId, cmd) = do
|
||||
HELLO verifyKey _ -> helloMsg verifyKey msgBody
|
||||
REPLY qInfo -> replyMsg qInfo
|
||||
A_MSG body -> agentClientMsg previousMsgHash (senderMsgId, senderTimestamp) (srvMsgId, srvTs) body msgHash
|
||||
A_INTRO entity eInfo -> introMsg entity eInfo
|
||||
A_INV conn qInfo cInfo -> invMsg conn qInfo cInfo
|
||||
A_REQ conn qInfo cInfo -> reqMsg conn qInfo cInfo
|
||||
A_CON conn -> conMsg conn
|
||||
A_INTRO introId cInfo -> introMsg introId cInfo
|
||||
A_INV introId qInfo cInfo -> invMsg introId qInfo cInfo
|
||||
A_REQ introId qInfo cInfo -> reqMsg introId qInfo cInfo
|
||||
A_CON introId -> conMsg introId
|
||||
sendAck c rq
|
||||
return ()
|
||||
SMP.END -> do
|
||||
@@ -401,8 +357,8 @@ processSMPTransmission c@AgentClient {sndQ} st (srv, rId, cmd) = do
|
||||
logServer "<--" c srv rId $ "unexpected: " <> bshow cmd
|
||||
notify . ERR $ BROKER UNEXPECTED
|
||||
where
|
||||
notify :: EntityCommand 'Conn_ c => ACommand 'Agent c -> m ()
|
||||
notify msg = atomically . writeTBQueue sndQ $ ATransmission "" (Conn connId) msg
|
||||
notify :: ACommand 'Agent -> m ()
|
||||
notify msg = atomically $ writeTBQueue sndQ ("", connId, msg)
|
||||
|
||||
prohibited :: m ()
|
||||
prohibited = notify . ERR $ AGENT A_PROHIBITED
|
||||
@@ -448,19 +404,19 @@ processSMPTransmission c@AgentClient {sndQ} st (srv, rId, cmd) = do
|
||||
withStore (getConnInvitation st connId) >>= \case
|
||||
Just (Invitation {invId, externalIntroId}, DuplexConnection _ _ sq) -> do
|
||||
withStore $ setInvitationStatus st invId InvCon
|
||||
sendControlMessage c sq $ A_CON (Conn externalIntroId)
|
||||
sendControlMessage c sq $ A_CON externalIntroId
|
||||
_ -> pure ()
|
||||
notify CON
|
||||
|
||||
introMsg :: IntroEntity -> EntityInfo -> m ()
|
||||
introMsg (IE entity) entityInfo = do
|
||||
introMsg :: IntroId -> ConnInfo -> m ()
|
||||
introMsg introId reInfo = do
|
||||
logServer "<--" c srv rId "MSG <INTRO>"
|
||||
case (cType, entity) of
|
||||
(SCDuplex, intro@Conn {}) -> createInv intro Nothing entityInfo
|
||||
case cType of
|
||||
SCDuplex -> createInv introId Nothing reInfo
|
||||
_ -> prohibited
|
||||
|
||||
invMsg :: Entity 'Conn_ -> SMPQueueInfo -> EntityInfo -> m ()
|
||||
invMsg (Conn introId) qInfo toInfo = do
|
||||
invMsg :: IntroId -> SMPQueueInfo -> ConnInfo -> m ()
|
||||
invMsg introId qInfo toInfo = do
|
||||
logServer "<--" c srv rId "MSG <INV>"
|
||||
case cType of
|
||||
SCDuplex ->
|
||||
@@ -470,43 +426,41 @@ processSMPTransmission c@AgentClient {sndQ} st (srv, rId, cmd) = do
|
||||
| otherwise ->
|
||||
withStore (addIntroInvitation st introId toInfo qInfo >> getConn st reConn) >>= \case
|
||||
SomeConn _ (DuplexConnection _ _ sq) -> do
|
||||
sendControlMessage c sq $ A_REQ (Conn introId) qInfo toInfo
|
||||
sendControlMessage c sq $ A_REQ introId qInfo toInfo
|
||||
withStore $ setIntroReStatus st introId IntroInv
|
||||
_ -> prohibited
|
||||
_ -> prohibited
|
||||
_ -> prohibited
|
||||
|
||||
reqMsg :: Entity 'Conn_ -> SMPQueueInfo -> EntityInfo -> m ()
|
||||
reqMsg intro qInfo connInfo = do
|
||||
reqMsg :: IntroId -> SMPQueueInfo -> ConnInfo -> m ()
|
||||
reqMsg introId qInfo connInfo = do
|
||||
logServer "<--" c srv rId "MSG <REQ>"
|
||||
case cType of
|
||||
SCDuplex -> createInv intro (Just qInfo) connInfo
|
||||
SCDuplex -> createInv introId (Just qInfo) connInfo
|
||||
_ -> prohibited
|
||||
|
||||
createInv :: Entity 'Conn_ -> Maybe SMPQueueInfo -> EntityInfo -> m ()
|
||||
createInv (Conn externalIntroId) qInfo entityInfo = do
|
||||
createInv :: IntroId -> Maybe SMPQueueInfo -> ConnInfo -> m ()
|
||||
createInv externalIntroId qInfo connInfo = do
|
||||
g <- asks idsDrg
|
||||
let newInv = NewInvitation {viaConn = connId, externalIntroId, entityInfo, qInfo}
|
||||
let newInv = NewInvitation {viaConn = connId, externalIntroId, connInfo, qInfo}
|
||||
invId <- withStore $ createInvitation st g newInv
|
||||
notify $ REQ (IE (Conn invId)) entityInfo
|
||||
notify $ REQ invId connInfo
|
||||
|
||||
conMsg :: Entity 'Conn_ -> m ()
|
||||
conMsg (Conn introId) = do
|
||||
conMsg :: IntroId -> m ()
|
||||
conMsg introId = do
|
||||
logServer "<--" c srv rId "MSG <CON>"
|
||||
withStore (getIntro st introId) >>= \case
|
||||
Introduction {toConn, toStatus, reConn, reStatus}
|
||||
| toConn == connId && toStatus == IntroInv -> do
|
||||
withStore $ setIntroToStatus st introId IntroCon
|
||||
sendConMsg toConn reConn reStatus
|
||||
when (reStatus == IntroCon) $ sendConMsg toConn reConn
|
||||
| reConn == connId && reStatus == IntroInv -> do
|
||||
withStore $ setIntroReStatus st introId IntroCon
|
||||
sendConMsg toConn reConn toStatus
|
||||
when (toStatus == IntroCon) $ sendConMsg toConn reConn
|
||||
| otherwise -> prohibited
|
||||
where
|
||||
sendConMsg :: ConnId -> ConnId -> IntroStatus -> m ()
|
||||
sendConMsg toConn reConn IntroCon =
|
||||
atomically . writeTBQueue sndQ $ ATransmission "" (Conn toConn) $ ICON $ IE (Conn reConn)
|
||||
sendConMsg _ _ _ = pure ()
|
||||
sendConMsg :: ConnId -> ConnId -> m ()
|
||||
sendConMsg toConn reConn = atomically $ writeTBQueue sndQ ("", toConn, ICON reConn)
|
||||
|
||||
agentClientMsg :: PrevRcvMsgHash -> (ExternalSndId, ExternalSndTs) -> (BrokerId, BrokerTs) -> MsgBody -> MsgHash -> m ()
|
||||
agentClientMsg receivedPrevMsgHash senderMeta brokerMeta msgBody msgHash = do
|
||||
|
||||
@@ -119,7 +119,7 @@ getSMPServerClient c@AgentClient {smpClients, msgQ} srv =
|
||||
deleteKeys ks m = S.foldr' M.delete m ks
|
||||
|
||||
notifySub :: ConnId -> IO ()
|
||||
notifySub connAlias = atomically . writeTBQueue (sndQ c) $ ATransmission "" (Conn connAlias) END
|
||||
notifySub connId = atomically $ writeTBQueue (sndQ c) ("", connId, END)
|
||||
|
||||
closeSMPServerClients :: MonadUnliftIO m => AgentClient -> m ()
|
||||
closeSMPServerClients c = liftIO $ readTVarIO (smpClients c) >>= mapM_ closeSMPClient
|
||||
@@ -181,31 +181,31 @@ newReceiveQueue c srv = do
|
||||
return (rq, SMPQueueInfo srv sId encryptKey)
|
||||
|
||||
subscribeQueue :: AgentMonad m => AgentClient -> RcvQueue -> ConnId -> m ()
|
||||
subscribeQueue c rq@RcvQueue {server, rcvPrivateKey, rcvId} connAlias = do
|
||||
subscribeQueue c rq@RcvQueue {server, rcvPrivateKey, rcvId} connId = do
|
||||
withLogSMP c server rcvId "SUB" $ \smp ->
|
||||
subscribeSMPQueue smp rcvPrivateKey rcvId
|
||||
addSubscription c rq connAlias
|
||||
addSubscription c rq connId
|
||||
|
||||
addSubscription :: MonadUnliftIO m => AgentClient -> RcvQueue -> ConnId -> m ()
|
||||
addSubscription c RcvQueue {server} connAlias = atomically $ do
|
||||
modifyTVar (subscrConns c) $ M.insert connAlias server
|
||||
addSubscription c RcvQueue {server} connId = atomically $ do
|
||||
modifyTVar (subscrConns c) $ M.insert connId server
|
||||
modifyTVar (subscrSrvrs c) $ M.alter (Just . addSub) server
|
||||
where
|
||||
addSub :: Maybe (Set ConnId) -> Set ConnId
|
||||
addSub (Just cs) = S.insert connAlias cs
|
||||
addSub _ = S.singleton connAlias
|
||||
addSub (Just cs) = S.insert connId cs
|
||||
addSub _ = S.singleton connId
|
||||
|
||||
removeSubscription :: AgentMonad m => AgentClient -> ConnId -> m ()
|
||||
removeSubscription AgentClient {subscrConns, subscrSrvrs} connAlias = atomically $ do
|
||||
removeSubscription AgentClient {subscrConns, subscrSrvrs} connId = atomically $ do
|
||||
cs <- readTVar subscrConns
|
||||
writeTVar subscrConns $ M.delete connAlias cs
|
||||
writeTVar subscrConns $ M.delete connId cs
|
||||
mapM_
|
||||
(modifyTVar subscrSrvrs . M.alter (>>= delSub))
|
||||
(M.lookup connAlias cs)
|
||||
(M.lookup connId cs)
|
||||
where
|
||||
delSub :: Set ConnId -> Maybe (Set ConnId)
|
||||
delSub cs =
|
||||
let cs' = S.delete connAlias cs
|
||||
let cs' = S.delete connId cs
|
||||
in if S.null cs' then Nothing else Just cs'
|
||||
|
||||
logServer :: AgentMonad m => ByteString -> AgentClient -> SMPServer -> QueueId -> ByteString -> m ()
|
||||
|
||||
@@ -10,9 +10,7 @@
|
||||
{-# LANGUAGE RankNTypes #-}
|
||||
{-# LANGUAGE ScopedTypeVariables #-}
|
||||
{-# LANGUAGE StandaloneDeriving #-}
|
||||
{-# LANGUAGE TypeApplications #-}
|
||||
{-# LANGUAGE TypeFamilies #-}
|
||||
{-# LANGUAGE TypeOperators #-}
|
||||
{-# LANGUAGE UndecidableInstances #-}
|
||||
{-# OPTIONS_GHC -fno-warn-unticked-promoted-constructors #-}
|
||||
|
||||
@@ -30,17 +28,9 @@
|
||||
-- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/agent-protocol.md
|
||||
module Simplex.Messaging.Agent.Protocol
|
||||
( -- * SMP agent protocol types
|
||||
Entity (..),
|
||||
EntityTag (..),
|
||||
AnEntity (..),
|
||||
IntroEntity (..),
|
||||
EntityInfo,
|
||||
EntityCommand,
|
||||
entityCommand,
|
||||
ConnInfo,
|
||||
ACommand (..),
|
||||
ACmdTag (..),
|
||||
AParty (..),
|
||||
APartyCmd (..),
|
||||
SAParty (..),
|
||||
SMPMessage (..),
|
||||
AMessage (..),
|
||||
@@ -49,13 +39,14 @@ module Simplex.Messaging.Agent.Protocol
|
||||
AgentErrorType (..),
|
||||
CommandErrorType (..),
|
||||
ConnectionErrorType (..),
|
||||
BroadcastErrorType (..),
|
||||
BrokerErrorType (..),
|
||||
SMPAgentError (..),
|
||||
ATransmission (..),
|
||||
ATransmissionOrError (..),
|
||||
ATransmission,
|
||||
ATransmissionOrError,
|
||||
ARawTransmission,
|
||||
ConnId,
|
||||
IntroId,
|
||||
InvitationId,
|
||||
ReplyMode (..),
|
||||
AckMode (..),
|
||||
OnOff (..),
|
||||
@@ -71,14 +62,12 @@ module Simplex.Messaging.Agent.Protocol
|
||||
|
||||
-- * Parse and serialize
|
||||
serializeCommand,
|
||||
serializeEntity,
|
||||
serializeSMPMessage,
|
||||
serializeMsgIntegrity,
|
||||
serializeServer,
|
||||
serializeSmpQueueInfo,
|
||||
serializeAgentError,
|
||||
commandP,
|
||||
anEntityP,
|
||||
parseSMPMessage,
|
||||
smpServerP,
|
||||
smpQueueInfoP,
|
||||
@@ -100,18 +89,15 @@ import qualified Data.Attoparsec.ByteString.Char8 as A
|
||||
import Data.ByteString.Base64
|
||||
import Data.ByteString.Char8 (ByteString)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
import Data.Constraint (Dict (..))
|
||||
import Data.Functor (($>))
|
||||
import Data.Int (Int64)
|
||||
import Data.Kind (Constraint, Type)
|
||||
import Data.Maybe (isJust)
|
||||
import Data.Kind (Type)
|
||||
import Data.String (IsString (..))
|
||||
import Data.Time.Clock (UTCTime)
|
||||
import Data.Time.ISO8601
|
||||
import Data.Type.Equality
|
||||
import Data.Typeable ()
|
||||
import GHC.Generics (Generic)
|
||||
import GHC.TypeLits (ErrorMessage (..), TypeError)
|
||||
import Generic.Random (genericArbitraryU)
|
||||
import Network.Socket (HostName, ServiceName)
|
||||
import qualified Simplex.Messaging.Crypto as C
|
||||
@@ -133,12 +119,10 @@ import UnliftIO.Exception
|
||||
type ARawTransmission = (ByteString, ByteString, ByteString)
|
||||
|
||||
-- | Parsed SMP agent protocol transmission.
|
||||
data ATransmission p = forall t c. EntityCommand t c => ATransmission ACorrId (Entity t) (ACommand p c)
|
||||
type ATransmission p = (ACorrId, ConnId, ACommand p)
|
||||
|
||||
-- | SMP agent protocol transmission or transmission error.
|
||||
data ATransmissionOrError p = forall t c. EntityCommand t c => ATransmissionOrError ACorrId (Entity t) (Either AgentErrorType (ACommand p c))
|
||||
|
||||
deriving instance Show (ATransmissionOrError p)
|
||||
type ATransmissionOrError p = (ACorrId, ConnId, Either AgentErrorType (ACommand p))
|
||||
|
||||
type ACorrId = ByteString
|
||||
|
||||
@@ -160,173 +144,29 @@ instance TestEquality SAParty where
|
||||
testEquality SClient SClient = Just Refl
|
||||
testEquality _ _ = Nothing
|
||||
|
||||
-- | SMP agent protocol entity types
|
||||
data EntityTag = Conn_ | OpenConn_ | Broadcast_ | AGroup_
|
||||
|
||||
data Entity :: EntityTag -> Type where
|
||||
Conn :: {fromConn :: ByteString} -> Entity Conn_
|
||||
OpenConn :: {fromOpenConn :: ByteString} -> Entity OpenConn_
|
||||
Broadcast :: {fromBroadcast :: ByteString} -> Entity Broadcast_
|
||||
AGroup :: {fromAGroup :: ByteString} -> Entity AGroup_
|
||||
|
||||
deriving instance Eq (Entity t)
|
||||
|
||||
deriving instance Show (Entity t)
|
||||
|
||||
instance TestEquality Entity where
|
||||
testEquality (Conn c) (Conn c') = refl c c'
|
||||
testEquality (OpenConn c) (OpenConn c') = refl c c'
|
||||
testEquality (Broadcast c) (Broadcast c') = refl c c'
|
||||
testEquality (AGroup c) (AGroup c') = refl c c'
|
||||
testEquality _ _ = Nothing
|
||||
|
||||
entityId :: Entity t -> ByteString
|
||||
entityId = \case
|
||||
Conn bs -> bs
|
||||
OpenConn bs -> bs
|
||||
Broadcast bs -> bs
|
||||
AGroup bs -> bs
|
||||
|
||||
data AnEntity = forall t. AE (Entity t)
|
||||
|
||||
data ACmd = forall (p :: AParty) (c :: ACmdTag). ACmd (SAParty p) (ACommand p c)
|
||||
data ACmd = forall p. ACmd (SAParty p) (ACommand p)
|
||||
|
||||
deriving instance Show ACmd
|
||||
|
||||
data APartyCmd (p :: AParty) = forall c. APartyCmd (ACommand p c)
|
||||
|
||||
instance Eq (APartyCmd p) where
|
||||
APartyCmd c1 == APartyCmd c2 = isJust $ testEquality c1 c2
|
||||
|
||||
deriving instance Show (APartyCmd p)
|
||||
|
||||
type family EntityCommand (t :: EntityTag) (c :: ACmdTag) :: Constraint where
|
||||
EntityCommand Conn_ NEW_ = ()
|
||||
EntityCommand Conn_ INV_ = ()
|
||||
EntityCommand Conn_ JOIN_ = ()
|
||||
EntityCommand Conn_ INTRO_ = ()
|
||||
EntityCommand Conn_ REQ_ = ()
|
||||
EntityCommand Conn_ ACPT_ = ()
|
||||
EntityCommand Conn_ CON_ = ()
|
||||
EntityCommand Conn_ ICON_ = ()
|
||||
EntityCommand Conn_ SUB_ = ()
|
||||
EntityCommand Conn_ SUBALL_ = ()
|
||||
EntityCommand Conn_ END_ = ()
|
||||
EntityCommand Conn_ SEND_ = ()
|
||||
EntityCommand Conn_ SENT_ = ()
|
||||
EntityCommand Conn_ MSG_ = ()
|
||||
EntityCommand Conn_ OFF_ = ()
|
||||
EntityCommand Conn_ DEL_ = ()
|
||||
EntityCommand Conn_ OK_ = ()
|
||||
EntityCommand Conn_ ERR_ = ()
|
||||
EntityCommand Broadcast_ NEW_ = ()
|
||||
EntityCommand Broadcast_ ADD_ = ()
|
||||
EntityCommand Broadcast_ REM_ = ()
|
||||
EntityCommand Broadcast_ LS_ = ()
|
||||
EntityCommand Broadcast_ MS_ = ()
|
||||
EntityCommand Broadcast_ SEND_ = ()
|
||||
EntityCommand Broadcast_ SENT_ = ()
|
||||
EntityCommand Broadcast_ DEL_ = ()
|
||||
EntityCommand Broadcast_ OK_ = ()
|
||||
EntityCommand Broadcast_ ERR_ = ()
|
||||
EntityCommand _ ERR_ = ()
|
||||
EntityCommand t c =
|
||||
(Int ~ Bool, TypeError (Text "Entity " :<>: ShowType t :<>: Text " does not support command " :<>: ShowType c))
|
||||
|
||||
entityCommand :: Entity t -> ACommand p c -> Maybe (Dict (EntityCommand t c))
|
||||
entityCommand = \case
|
||||
Conn _ -> \case
|
||||
NEW -> Just Dict
|
||||
INV _ -> Just Dict
|
||||
JOIN {} -> Just Dict
|
||||
INTRO {} -> Just Dict
|
||||
REQ {} -> Just Dict
|
||||
ACPT {} -> Just Dict
|
||||
CON -> Just Dict
|
||||
ICON {} -> Just Dict
|
||||
SUB -> Just Dict
|
||||
SUBALL -> Just Dict
|
||||
END -> Just Dict
|
||||
SEND _ -> Just Dict
|
||||
SENT _ -> Just Dict
|
||||
MSG {} -> Just Dict
|
||||
OFF -> Just Dict
|
||||
DEL -> Just Dict
|
||||
OK -> Just Dict
|
||||
ERR _ -> Just Dict
|
||||
_ -> Nothing
|
||||
Broadcast _ -> \case
|
||||
NEW -> Just Dict
|
||||
ADD _ -> Just Dict
|
||||
REM _ -> Just Dict
|
||||
LS -> Just Dict
|
||||
MS _ -> Just Dict
|
||||
SEND _ -> Just Dict
|
||||
SENT _ -> Just Dict
|
||||
DEL -> Just Dict
|
||||
OK -> Just Dict
|
||||
ERR _ -> Just Dict
|
||||
_ -> Nothing
|
||||
_ -> \case
|
||||
ERR _ -> Just Dict
|
||||
_ -> Nothing
|
||||
|
||||
data ACmdTag
|
||||
= NEW_
|
||||
| INV_
|
||||
| JOIN_
|
||||
| INTRO_
|
||||
| REQ_
|
||||
| ACPT_
|
||||
| CON_
|
||||
| ICON_
|
||||
| SUB_
|
||||
| SUBALL_
|
||||
| END_
|
||||
| SEND_
|
||||
| SENT_
|
||||
| MSG_
|
||||
| OFF_
|
||||
| DEL_
|
||||
| ADD_
|
||||
| REM_
|
||||
| LS_
|
||||
| MS_
|
||||
| OK_
|
||||
| ERR_
|
||||
|
||||
type family Introduction (t :: EntityTag) :: Constraint where
|
||||
Introduction Conn_ = ()
|
||||
Introduction OpenConn_ = ()
|
||||
Introduction AGroup_ = ()
|
||||
Introduction t = (Int ~ Bool, TypeError (Text "Entity " :<>: ShowType t :<>: Text " cannot be INTRO'd to"))
|
||||
|
||||
data IntroEntity = forall t. Introduction t => IE (Entity t)
|
||||
|
||||
instance Eq IntroEntity where
|
||||
IE e1 == IE e2 = isJust $ testEquality e1 e2
|
||||
|
||||
deriving instance Show IntroEntity
|
||||
|
||||
type EntityInfo = ByteString
|
||||
type ConnInfo = ByteString
|
||||
|
||||
-- | Parameterized type for SMP agent protocol commands and responses from all participants.
|
||||
data ACommand (p :: AParty) (c :: ACmdTag) where
|
||||
NEW :: ACommand Client NEW_ -- response INV
|
||||
INV :: SMPQueueInfo -> ACommand Agent INV_
|
||||
JOIN :: SMPQueueInfo -> ReplyMode -> ACommand Client JOIN_ -- response OK
|
||||
INTRO :: IntroEntity -> EntityInfo -> ACommand Client INTRO_
|
||||
REQ :: IntroEntity -> EntityInfo -> ACommand Agent INTRO_
|
||||
ACPT :: IntroEntity -> EntityInfo -> ACommand Client ACPT_
|
||||
CON :: ACommand Agent CON_ -- notification that connection is established
|
||||
ICON :: IntroEntity -> ACommand Agent ICON_
|
||||
SUB :: ACommand Client SUB_
|
||||
SUBALL :: ACommand Client SUBALL_ -- TODO should be moved to chat protocol - hack for subscribing to all
|
||||
END :: ACommand Agent END_
|
||||
data ACommand (p :: AParty) where
|
||||
NEW :: ACommand Client -- response INV
|
||||
INV :: SMPQueueInfo -> ACommand Agent
|
||||
JOIN :: SMPQueueInfo -> ReplyMode -> ACommand Client -- response OK
|
||||
INTRO :: ConnId -> ConnInfo -> ACommand Client
|
||||
REQ :: InvitationId -> ConnInfo -> ACommand Agent
|
||||
ACPT :: InvitationId -> ConnInfo -> ACommand Client
|
||||
CON :: ACommand Agent -- notification that connection is established
|
||||
ICON :: ConnId -> ACommand Agent
|
||||
SUB :: ACommand Client
|
||||
SUBALL :: ACommand Client -- TODO should be moved to chat protocol - hack for subscribing to all
|
||||
END :: ACommand Agent
|
||||
-- QST :: QueueDirection -> ACommand Client
|
||||
-- STAT :: QueueDirection -> Maybe QueueStatus -> Maybe SubMode -> ACommand Agent
|
||||
SEND :: MsgBody -> ACommand Client SEND_
|
||||
SENT :: AgentMsgId -> ACommand Agent SENT_
|
||||
SEND :: MsgBody -> ACommand Client
|
||||
SENT :: AgentMsgId -> ACommand Agent
|
||||
MSG ::
|
||||
{ recipientMeta :: (AgentMsgId, UTCTime),
|
||||
brokerMeta :: (MsgId, UTCTime),
|
||||
@@ -334,46 +174,17 @@ data ACommand (p :: AParty) (c :: ACmdTag) where
|
||||
msgIntegrity :: MsgIntegrity,
|
||||
msgBody :: MsgBody
|
||||
} ->
|
||||
ACommand Agent MSG_
|
||||
ACommand Agent
|
||||
-- ACK :: AgentMsgId -> ACommand Client
|
||||
-- RCVD :: AgentMsgId -> ACommand Agent
|
||||
OFF :: ACommand Client MSG_
|
||||
DEL :: ACommand Client DEL_
|
||||
ADD :: Entity Conn_ -> ACommand Client ADD_
|
||||
REM :: Entity Conn_ -> ACommand Client REM_
|
||||
LS :: ACommand Client LS_
|
||||
MS :: [Entity Conn_] -> ACommand Agent MS_
|
||||
OK :: ACommand Agent OK_
|
||||
ERR :: AgentErrorType -> ACommand Agent ERR_
|
||||
OFF :: ACommand Client
|
||||
DEL :: ACommand Client
|
||||
OK :: ACommand Agent
|
||||
ERR :: AgentErrorType -> ACommand Agent
|
||||
|
||||
deriving instance Eq (ACommand p c)
|
||||
deriving instance Eq (ACommand p)
|
||||
|
||||
deriving instance Show (ACommand p c)
|
||||
|
||||
instance TestEquality (ACommand p) where
|
||||
testEquality NEW NEW = Just Refl
|
||||
testEquality c@INV {} c'@INV {} = refl c c'
|
||||
testEquality c@JOIN {} c'@JOIN {} = refl c c'
|
||||
testEquality CON CON = Just Refl
|
||||
testEquality c@ICON {} c'@ICON {} = refl c c'
|
||||
testEquality SUB SUB = Just Refl
|
||||
testEquality SUBALL SUBALL = Just Refl
|
||||
testEquality END END = Just Refl
|
||||
testEquality c@SEND {} c'@SEND {} = refl c c'
|
||||
testEquality c@SENT {} c'@SENT {} = refl c c'
|
||||
testEquality c@MSG {} c'@MSG {} = refl c c'
|
||||
testEquality OFF OFF = Just Refl
|
||||
testEquality DEL DEL = Just Refl
|
||||
testEquality c@ADD {} c'@ADD {} = refl c c'
|
||||
testEquality c@REM {} c'@REM {} = refl c c'
|
||||
testEquality c@LS {} c'@LS {} = refl c c'
|
||||
testEquality c@MS {} c'@MS {} = refl c c'
|
||||
testEquality OK OK = Just Refl
|
||||
testEquality c@ERR {} c'@ERR {} = refl c c'
|
||||
testEquality _ _ = Nothing
|
||||
|
||||
refl :: Eq a => a -> a -> Maybe (t :~: t)
|
||||
refl x x' = if x == x' then Just Refl else Nothing
|
||||
deriving instance Show (ACommand p)
|
||||
|
||||
-- | SMP message formats.
|
||||
data SMPMessage
|
||||
@@ -405,13 +216,13 @@ data AMessage where
|
||||
-- | agent envelope for the client message
|
||||
A_MSG :: MsgBody -> AMessage
|
||||
-- | agent message for introduction
|
||||
A_INTRO :: IntroEntity -> EntityInfo -> AMessage
|
||||
A_INTRO :: IntroId -> ConnInfo -> AMessage
|
||||
-- | agent envelope for the sent invitation
|
||||
A_INV :: Entity Conn_ -> SMPQueueInfo -> EntityInfo -> AMessage
|
||||
A_INV :: IntroId -> SMPQueueInfo -> ConnInfo -> AMessage
|
||||
-- | agent envelope for the forwarded invitation
|
||||
A_REQ :: Entity Conn_ -> SMPQueueInfo -> EntityInfo -> AMessage
|
||||
A_REQ :: IntroId -> SMPQueueInfo -> ConnInfo -> AMessage
|
||||
-- | agent message for intro/group request
|
||||
A_CON :: Entity Conn_ -> AMessage
|
||||
A_CON :: IntroId -> AMessage
|
||||
deriving (Show)
|
||||
|
||||
-- | Parse SMP message.
|
||||
@@ -462,11 +273,11 @@ agentMessageP =
|
||||
hello = HELLO <$> C.pubKeyP <*> ackMode
|
||||
reply = REPLY <$> smpQueueInfoP
|
||||
a_msg = A_MSG <$> binaryBody
|
||||
a_intro = A_INTRO <$> introEntityP <* A.space <*> binaryBody
|
||||
a_intro = A_INTRO <$> A.takeTill (== ' ') <* A.space <*> binaryBody
|
||||
a_inv = invP A_INV
|
||||
a_req = invP A_REQ
|
||||
a_con = A_CON <$> connEntityP
|
||||
invP f = f <$> connEntityP <* A.space <*> smpQueueInfoP <* A.space <*> binaryBody
|
||||
a_con = A_CON <$> A.takeTill wordEnd
|
||||
invP f = f <$> A.takeTill (== ' ') <* A.space <*> smpQueueInfoP <* A.space <*> binaryBody
|
||||
binaryBody = do
|
||||
size :: Int <- A.decimal <* A.endOfLine
|
||||
A.take size <* A.endOfLine
|
||||
@@ -490,13 +301,13 @@ serializeAgentMessage = \case
|
||||
HELLO verifyKey ackMode -> "HELLO " <> C.serializePubKey verifyKey <> if ackMode == AckMode Off then " NO_ACK" else ""
|
||||
REPLY qInfo -> "REPLY " <> serializeSmpQueueInfo qInfo
|
||||
A_MSG body -> "MSG " <> serializeMsg body <> "\n"
|
||||
A_INTRO (IE entity) eInfo -> "INTRO " <> serializeIntro entity eInfo <> "\n"
|
||||
A_INV conn qInfo eInfo -> "INV " <> serializeInv conn qInfo eInfo
|
||||
A_REQ conn qInfo eInfo -> "REQ " <> serializeInv conn qInfo eInfo
|
||||
A_CON conn -> "CON " <> serializeEntity conn
|
||||
A_INTRO introId cInfo -> "INTRO " <> introId <> " " <> serializeMsg cInfo <> "\n"
|
||||
A_INV introId qInfo cInfo -> "INV " <> serializeInv introId qInfo cInfo
|
||||
A_REQ introId qInfo cInfo -> "REQ " <> serializeInv introId qInfo cInfo
|
||||
A_CON introId -> "CON " <> introId
|
||||
where
|
||||
serializeInv conn qInfo eInfo =
|
||||
B.intercalate " " [serializeEntity conn, serializeSmpQueueInfo qInfo, serializeMsg eInfo] <> "\n"
|
||||
serializeInv introId qInfo cInfo =
|
||||
B.intercalate " " [introId, serializeSmpQueueInfo qInfo, serializeMsg cInfo] <> "\n"
|
||||
|
||||
-- | Serialize SMP queue information that is sent out-of-band.
|
||||
serializeSmpQueueInfo :: SMPQueueInfo -> ByteString
|
||||
@@ -522,6 +333,10 @@ instance IsString SMPServer where
|
||||
-- | SMP agent connection alias.
|
||||
type ConnId = ByteString
|
||||
|
||||
type IntroId = ByteString
|
||||
|
||||
type InvitationId = ByteString
|
||||
|
||||
-- | Connection modes.
|
||||
data OnOff = On | Off deriving (Eq, Show, Read)
|
||||
|
||||
@@ -583,8 +398,6 @@ data AgentErrorType
|
||||
CMD CommandErrorType
|
||||
| -- | connection errors
|
||||
CONN ConnectionErrorType
|
||||
| -- | broadcast errors
|
||||
BCAST BroadcastErrorType
|
||||
| -- | SMP protocol errors forwarded to agent clients
|
||||
SMP ErrorType
|
||||
| -- | SMP server errors
|
||||
@@ -599,14 +412,10 @@ data AgentErrorType
|
||||
data CommandErrorType
|
||||
= -- | command is prohibited in this context
|
||||
PROHIBITED
|
||||
| -- | command is not supported by this entity
|
||||
UNSUPPORTED
|
||||
| -- | command syntax is invalid
|
||||
SYNTAX
|
||||
| -- | cannot parse entity
|
||||
BAD_ENTITY
|
||||
| -- | entity ID is required with this command
|
||||
NO_ENTITY
|
||||
NO_CONN
|
||||
| -- | message size is not correct (no terminating space)
|
||||
SIZE
|
||||
| -- | message does not fit in SMP block
|
||||
@@ -623,14 +432,6 @@ data ConnectionErrorType
|
||||
SIMPLEX
|
||||
deriving (Eq, Generic, Read, Show, Exception)
|
||||
|
||||
-- | Broadcast error
|
||||
data BroadcastErrorType
|
||||
= -- | broadcast ID is not in the database
|
||||
B_NOT_FOUND
|
||||
| -- | broadcast ID already exists
|
||||
B_DUPLICATE
|
||||
deriving (Eq, Generic, Read, Show, Exception)
|
||||
|
||||
-- | SMP server errors.
|
||||
data BrokerErrorType
|
||||
= -- | invalid server response (failed to parse)
|
||||
@@ -663,41 +464,10 @@ instance Arbitrary CommandErrorType where arbitrary = genericArbitraryU
|
||||
|
||||
instance Arbitrary ConnectionErrorType where arbitrary = genericArbitraryU
|
||||
|
||||
instance Arbitrary BroadcastErrorType where arbitrary = genericArbitraryU
|
||||
|
||||
instance Arbitrary BrokerErrorType where arbitrary = genericArbitraryU
|
||||
|
||||
instance Arbitrary SMPAgentError where arbitrary = genericArbitraryU
|
||||
|
||||
anEntityP :: Parser AnEntity
|
||||
anEntityP =
|
||||
($)
|
||||
<$> ( "C:" $> AE . Conn
|
||||
<|> "O:" $> AE . OpenConn
|
||||
<|> "B:" $> AE . Broadcast
|
||||
<|> "G:" $> AE . AGroup
|
||||
)
|
||||
<*> A.takeTill wordEnd
|
||||
|
||||
connEntityP :: Parser (Entity Conn_)
|
||||
connEntityP = "C:" *> (Conn <$> A.takeTill wordEnd)
|
||||
|
||||
introEntityP :: Parser IntroEntity
|
||||
introEntityP =
|
||||
($)
|
||||
<$> ( "C:" $> IE . Conn
|
||||
<|> "O:" $> IE . OpenConn
|
||||
<|> "G:" $> IE . AGroup
|
||||
)
|
||||
<*> A.takeTill wordEnd
|
||||
|
||||
serializeEntity :: Entity t -> ByteString
|
||||
serializeEntity = \case
|
||||
Conn s -> "C:" <> s
|
||||
OpenConn s -> "O:" <> s
|
||||
Broadcast s -> "B:" <> s
|
||||
AGroup s -> "G:" <> s
|
||||
|
||||
-- | SMP agent command and response parser
|
||||
commandP :: Parser ACmd
|
||||
commandP =
|
||||
@@ -715,10 +485,6 @@ commandP =
|
||||
<|> "MSG " *> message
|
||||
<|> "OFF" $> ACmd SClient OFF
|
||||
<|> "DEL" $> ACmd SClient DEL
|
||||
<|> "ADD " *> addCmd
|
||||
<|> "REM " *> removeCmd
|
||||
<|> "LS" $> ACmd SClient LS
|
||||
<|> "MS " *> membersResp
|
||||
<|> "ERR " *> agentError
|
||||
<|> "ICON " *> iconMsg
|
||||
<|> "CON" $> ACmd SAgent CON
|
||||
@@ -731,10 +497,7 @@ commandP =
|
||||
acptCmd = ACmd SClient <$> introP ACPT
|
||||
sendCmd = ACmd SClient . SEND <$> A.takeByteString
|
||||
sentResp = ACmd SAgent . SENT <$> A.decimal
|
||||
addCmd = ACmd SClient . ADD <$> connEntityP
|
||||
removeCmd = ACmd SClient . REM <$> connEntityP
|
||||
membersResp = ACmd SAgent . MS <$> (connEntityP `A.sepBy'` A.char ' ')
|
||||
iconMsg = ACmd SAgent . ICON <$> introEntityP
|
||||
iconMsg = ACmd SAgent . ICON <$> A.takeTill wordEnd
|
||||
message = do
|
||||
msgIntegrity <- msgIntegrityP <* A.space
|
||||
recipientMeta <- "R=" *> partyMeta A.decimal
|
||||
@@ -742,7 +505,7 @@ commandP =
|
||||
senderMeta <- "S=" *> partyMeta A.decimal
|
||||
msgBody <- A.takeByteString
|
||||
return $ ACmd SAgent MSG {recipientMeta, brokerMeta, senderMeta, msgIntegrity, msgBody}
|
||||
introP f = f <$> introEntityP <* A.space <*> A.takeByteString
|
||||
introP f = f <$> A.takeTill (== ' ') <* A.space <*> A.takeByteString
|
||||
replyMode = ReplyMode <$> (" NO_REPLY" $> Off <|> pure On)
|
||||
partyMeta idParser = (,) <$> idParser <* "," <*> tsISO8601P <* A.space
|
||||
agentError = ACmd SAgent . ERR <$> agentErrorTypeP
|
||||
@@ -761,14 +524,14 @@ parseCommand :: ByteString -> Either AgentErrorType ACmd
|
||||
parseCommand = parse commandP $ CMD SYNTAX
|
||||
|
||||
-- | Serialize SMP agent command.
|
||||
serializeCommand :: ACommand p c -> ByteString
|
||||
serializeCommand :: ACommand p -> ByteString
|
||||
serializeCommand = \case
|
||||
NEW -> "NEW"
|
||||
INV qInfo -> "INV " <> serializeSmpQueueInfo qInfo
|
||||
JOIN qInfo rMode -> "JOIN " <> serializeSmpQueueInfo qInfo <> replyMode rMode
|
||||
INTRO (IE entity) eInfo -> "INTRO " <> serializeIntro entity eInfo
|
||||
REQ (IE entity) eInfo -> "REQ " <> serializeIntro entity eInfo
|
||||
ACPT (IE entity) eInfo -> "ACPT " <> serializeIntro entity eInfo
|
||||
INTRO connId cInfo -> "INTRO " <> connId <> " " <> serializeMsg cInfo
|
||||
REQ invId cInfo -> "REQ " <> invId <> " " <> serializeMsg cInfo
|
||||
ACPT invId cInfo -> "ACPT " <> invId <> " " <> serializeMsg cInfo
|
||||
SUB -> "SUB"
|
||||
SUBALL -> "SUBALL" -- TODO remove - hack for subscribing to all
|
||||
END -> "END"
|
||||
@@ -785,12 +548,8 @@ serializeCommand = \case
|
||||
]
|
||||
OFF -> "OFF"
|
||||
DEL -> "DEL"
|
||||
ADD c -> "ADD " <> serializeEntity c
|
||||
REM c -> "REM " <> serializeEntity c
|
||||
LS -> "LS"
|
||||
MS cs -> "MS " <> B.intercalate " " (map serializeEntity cs)
|
||||
CON -> "CON"
|
||||
ICON (IE entity) -> "ICON " <> serializeEntity entity
|
||||
ICON introId -> "ICON " <> introId
|
||||
ERR e -> "ERR " <> serializeAgentError e
|
||||
OK -> "OK"
|
||||
where
|
||||
@@ -801,9 +560,6 @@ serializeCommand = \case
|
||||
showTs :: UTCTime -> ByteString
|
||||
showTs = B.pack . formatISO8601Millis
|
||||
|
||||
serializeIntro :: Entity t -> ByteString -> ByteString
|
||||
serializeIntro entity eInfo = serializeEntity entity <> " " <> serializeMsg eInfo
|
||||
|
||||
-- | Serialize message integrity validation result.
|
||||
serializeMsgIntegrity :: MsgIntegrity -> ByteString
|
||||
serializeMsgIntegrity = \case
|
||||
@@ -820,7 +576,6 @@ serializeMsgIntegrity = \case
|
||||
agentErrorTypeP :: Parser AgentErrorType
|
||||
agentErrorTypeP =
|
||||
"SMP " *> (SMP <$> SMP.errorTypeP)
|
||||
<|> "BCAST " *> (BCAST <$> bcastErrorP)
|
||||
<|> "BROKER RESPONSE " *> (BROKER . RESPONSE <$> SMP.errorTypeP)
|
||||
<|> "BROKER TRANSPORT " *> (BROKER . TRANSPORT <$> transportErrorP)
|
||||
<|> "INTERNAL " *> (INTERNAL <$> parseRead A.takeByteString)
|
||||
@@ -830,19 +585,10 @@ agentErrorTypeP =
|
||||
serializeAgentError :: AgentErrorType -> ByteString
|
||||
serializeAgentError = \case
|
||||
SMP e -> "SMP " <> SMP.serializeErrorType e
|
||||
BCAST e -> "BCAST " <> serializeBcastError e
|
||||
BROKER (RESPONSE e) -> "BROKER RESPONSE " <> SMP.serializeErrorType e
|
||||
BROKER (TRANSPORT e) -> "BROKER TRANSPORT " <> serializeTransportError e
|
||||
e -> bshow e
|
||||
|
||||
bcastErrorP :: Parser BroadcastErrorType
|
||||
bcastErrorP = "NOT_FOUND" $> B_NOT_FOUND <|> "DUPLICATE" $> B_DUPLICATE
|
||||
|
||||
serializeBcastError :: BroadcastErrorType -> ByteString
|
||||
serializeBcastError = \case
|
||||
B_NOT_FOUND -> "NOT_FOUND"
|
||||
B_DUPLICATE -> "DUPLICATE"
|
||||
|
||||
serializeMsg :: ByteString -> ByteString
|
||||
serializeMsg body = bshow (B.length body) <> "\n" <> body
|
||||
|
||||
@@ -859,58 +605,45 @@ tGetRaw h = (,,) <$> getLn h <*> getLn h <*> getLn h
|
||||
|
||||
-- | Send SMP agent protocol command (or response) to TCP connection.
|
||||
tPut :: (Transport c, MonadIO m) => c -> ATransmission p -> m ()
|
||||
tPut h (ATransmission corrId ent cmd) =
|
||||
liftIO $ tPutRaw h (corrId, serializeEntity ent, serializeCommand cmd)
|
||||
tPut h (corrId, connAlias, command) =
|
||||
liftIO $ tPutRaw h (corrId, connAlias, serializeCommand command)
|
||||
|
||||
-- | Receive client and agent transmissions from TCP connection.
|
||||
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)
|
||||
tParseLoadBody (corrId, entityStr, command) =
|
||||
case parseAll anEntityP entityStr of
|
||||
Left _ -> pure $ ATransmissionOrError @_ @_ @ERR_ corrId (Conn "") $ Left $ CMD BAD_ENTITY
|
||||
Right entity -> do
|
||||
let cmd = parseCommand command >>= fromParty >>= hasEntityId entity
|
||||
makeTransmission corrId entity <$> either (pure . Left) cmdWithMsgBody cmd
|
||||
tParseLoadBody t@(corrId, connId, command) = do
|
||||
let cmd = parseCommand command >>= fromParty >>= tConnId t
|
||||
fullCmd <- either (return . Left) cmdWithMsgBody cmd
|
||||
return (corrId, connId, fullCmd)
|
||||
|
||||
fromParty :: ACmd -> Either AgentErrorType (APartyCmd p)
|
||||
fromParty :: ACmd -> Either AgentErrorType (ACommand p)
|
||||
fromParty (ACmd (p :: p1) cmd) = case testEquality party p of
|
||||
Just Refl -> Right $ APartyCmd cmd
|
||||
Just Refl -> Right cmd
|
||||
_ -> Left $ CMD PROHIBITED
|
||||
|
||||
hasEntityId :: AnEntity -> APartyCmd p -> Either AgentErrorType (APartyCmd p)
|
||||
hasEntityId (AE entity) (APartyCmd cmd) =
|
||||
APartyCmd <$> case cmd of
|
||||
-- NEW, JOIN and ACPT have optional entity
|
||||
NEW -> Right cmd
|
||||
JOIN {} -> Right cmd
|
||||
ACPT {} -> Right cmd
|
||||
-- ERROR response does not always have entity
|
||||
ERR _ -> Right cmd
|
||||
-- other responses must have entity
|
||||
_
|
||||
| B.null (entityId entity) -> Left $ CMD NO_ENTITY
|
||||
| otherwise -> Right cmd
|
||||
tConnId :: ARawTransmission -> ACommand p -> Either AgentErrorType (ACommand p)
|
||||
tConnId (_, connId, _) cmd = case cmd of
|
||||
-- NEW, JOIN and ACPT have optional connId
|
||||
NEW -> Right cmd
|
||||
JOIN {} -> Right cmd
|
||||
ACPT {} -> Right cmd
|
||||
-- ERROR response does not always have connId
|
||||
ERR _ -> Right cmd
|
||||
-- other responses must have connId
|
||||
_
|
||||
| B.null connId -> Left $ CMD NO_CONN
|
||||
| otherwise -> Right cmd
|
||||
|
||||
makeTransmission :: ACorrId -> AnEntity -> Either AgentErrorType (APartyCmd p) -> ATransmissionOrError p
|
||||
makeTransmission corrId (AE entity) = \case
|
||||
Left e -> err e
|
||||
Right (APartyCmd cmd) -> case entityCommand entity cmd of
|
||||
Just Dict -> ATransmissionOrError corrId entity $ Right cmd
|
||||
_ -> err $ CMD UNSUPPORTED
|
||||
where
|
||||
err e = ATransmissionOrError @_ @_ @ERR_ corrId entity $ Left e
|
||||
|
||||
cmdWithMsgBody :: APartyCmd p -> m (Either AgentErrorType (APartyCmd p))
|
||||
cmdWithMsgBody (APartyCmd cmd) =
|
||||
APartyCmd <$$> case cmd of
|
||||
SEND body -> SEND <$$> getMsgBody body
|
||||
MSG agentMsgId srvTS agentTS integrity body -> MSG agentMsgId srvTS agentTS integrity <$$> getMsgBody body
|
||||
INTRO entity eInfo -> INTRO entity <$$> getMsgBody eInfo
|
||||
REQ entity eInfo -> REQ entity <$$> getMsgBody eInfo
|
||||
ACPT entity eInfo -> ACPT entity <$$> getMsgBody eInfo
|
||||
_ -> pure $ Right cmd
|
||||
cmdWithMsgBody :: ACommand p -> m (Either AgentErrorType (ACommand p))
|
||||
cmdWithMsgBody = \case
|
||||
SEND body -> SEND <$$> getMsgBody body
|
||||
MSG agentMsgId srvTS agentTS integrity body -> MSG agentMsgId srvTS agentTS integrity <$$> getMsgBody body
|
||||
INTRO introId cInfo -> INTRO introId <$$> getMsgBody cInfo
|
||||
REQ introId cInfo -> REQ introId <$$> getMsgBody cInfo
|
||||
ACPT introId cInfo -> ACPT introId <$$> getMsgBody cInfo
|
||||
cmd -> pure $ Right cmd
|
||||
|
||||
-- TODO refactor with server
|
||||
getMsgBody :: MsgBody -> m (Either AgentErrorType MsgBody)
|
||||
|
||||
@@ -56,17 +56,10 @@ class Monad m => MonadAgentStore s m where
|
||||
|
||||
getMsg :: s -> ConnId -> InternalId -> m Msg
|
||||
|
||||
-- Broadcasts
|
||||
createBcast :: s -> TVar ChaChaDRG -> BroadcastId -> m BroadcastId
|
||||
addBcastConn :: s -> BroadcastId -> ConnId -> m ()
|
||||
removeBcastConn :: s -> BroadcastId -> ConnId -> m ()
|
||||
deleteBcast :: s -> BroadcastId -> m ()
|
||||
getBcast :: s -> BroadcastId -> m [ConnId]
|
||||
|
||||
-- Introductions
|
||||
createIntro :: s -> TVar ChaChaDRG -> NewIntroduction -> m IntroId
|
||||
getIntro :: s -> IntroId -> m Introduction
|
||||
addIntroInvitation :: s -> IntroId -> EntityInfo -> SMPQueueInfo -> m ()
|
||||
addIntroInvitation :: s -> IntroId -> ConnInfo -> SMPQueueInfo -> m ()
|
||||
setIntroToStatus :: s -> IntroId -> IntroStatus -> m ()
|
||||
setIntroReStatus :: s -> IntroId -> IntroStatus -> m ()
|
||||
createInvitation :: s -> TVar ChaChaDRG -> NewInvitation -> m InvitationId
|
||||
@@ -334,12 +327,10 @@ introStatusT = \case
|
||||
"CON" -> Just IntroCon
|
||||
_ -> Nothing
|
||||
|
||||
type IntroId = ByteString
|
||||
|
||||
data NewInvitation = NewInvitation
|
||||
{ viaConn :: ConnId,
|
||||
externalIntroId :: IntroId,
|
||||
entityInfo :: EntityInfo,
|
||||
connInfo :: ConnInfo,
|
||||
qInfo :: Maybe SMPQueueInfo
|
||||
}
|
||||
|
||||
@@ -347,7 +338,7 @@ data Invitation = Invitation
|
||||
{ invId :: InvitationId,
|
||||
viaConn :: ConnId,
|
||||
externalIntroId :: IntroId,
|
||||
entityInfo :: EntityInfo,
|
||||
connInfo :: ConnInfo,
|
||||
qInfo :: Maybe SMPQueueInfo,
|
||||
connId :: Maybe ConnId,
|
||||
status :: InvitationStatus
|
||||
@@ -370,8 +361,6 @@ invStatusT = \case
|
||||
"CON" -> Just InvCon
|
||||
_ -> Nothing
|
||||
|
||||
type InvitationId = ByteString
|
||||
|
||||
-- * Store errors
|
||||
|
||||
-- | Agent store error.
|
||||
@@ -387,10 +376,6 @@ data StoreError
|
||||
| -- | Wrong connection type, e.g. "send" connection when "receive" or "duplex" is expected, or vice versa.
|
||||
-- 'upgradeRcvConnToDuplex' and 'upgradeSndConnToDuplex' do not allow duplex connections - they would also return this error.
|
||||
SEBadConnType ConnType
|
||||
| -- | Broadcast ID not found.
|
||||
SEBcastNotFound
|
||||
| -- | Broadcast ID already used.
|
||||
SEBcastDuplicate
|
||||
| -- | Introduction ID not found.
|
||||
SEIntroNotFound
|
||||
| -- | Invitation ID not found.
|
||||
|
||||
@@ -29,11 +29,9 @@ import Control.Monad (join, unless, when)
|
||||
import Control.Monad.Except (MonadError (throwError), MonadIO (liftIO))
|
||||
import Control.Monad.IO.Unlift (MonadUnliftIO)
|
||||
import Crypto.Random (ChaChaDRG, randomBytesGenerate)
|
||||
import Data.Bifunctor (first)
|
||||
import Data.ByteString (ByteString)
|
||||
import Data.ByteString.Base64 (encode)
|
||||
import Data.Char (toLower)
|
||||
import Data.Functor (($>))
|
||||
import Data.List (find)
|
||||
import Data.Maybe (fromMaybe)
|
||||
import Data.Text (Text)
|
||||
@@ -114,11 +112,8 @@ connectSQLiteStore dbFilePath = do
|
||||
|]
|
||||
pure SQLiteStore {dbFilePath, dbConn, dbNew}
|
||||
|
||||
checkConstraint :: StoreError -> IO a -> IO (Either StoreError a)
|
||||
checkConstraint err action = first (handleSQLError err) <$> E.try action
|
||||
|
||||
checkConstraint' :: StoreError -> IO (Either StoreError a) -> IO (Either StoreError a)
|
||||
checkConstraint' err action = action `E.catch` (pure . Left . handleSQLError err)
|
||||
checkConstraint :: StoreError -> IO (Either StoreError a) -> IO (Either StoreError a)
|
||||
checkConstraint err action = action `E.catch` (pure . Left . handleSQLError err)
|
||||
|
||||
handleSQLError :: StoreError -> SQLError -> StoreError
|
||||
handleSQLError err e
|
||||
@@ -142,7 +137,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
|
||||
createRcvConn SQLiteStore {dbConn} gVar cData q@RcvQueue {server} =
|
||||
-- TODO if schema has to be restarted, this function can be refactored
|
||||
-- to create connection first using createWithRandomId
|
||||
liftIOEither . checkConstraint' SEConnDuplicate . withTransaction dbConn $
|
||||
liftIOEither . checkConstraint SEConnDuplicate . withTransaction dbConn $
|
||||
getConnId_ dbConn gVar cData >>= traverse create
|
||||
where
|
||||
create :: ConnId -> IO ConnId
|
||||
@@ -156,7 +151,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
|
||||
createSndConn SQLiteStore {dbConn} gVar cData q@SndQueue {server} =
|
||||
-- TODO if schema has to be restarted, this function can be refactored
|
||||
-- to create connection first using createWithRandomId
|
||||
liftIOEither . checkConstraint' SEConnDuplicate . withTransaction dbConn $
|
||||
liftIOEither . checkConstraint SEConnDuplicate . withTransaction dbConn $
|
||||
getConnId_ dbConn gVar cData >>= traverse create
|
||||
where
|
||||
create :: ConnId -> IO ConnId
|
||||
@@ -303,55 +298,6 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
|
||||
getMsg :: SQLiteStore -> ConnId -> InternalId -> m Msg
|
||||
getMsg _st _connAlias _id = throwError SENotImplemented
|
||||
|
||||
createBcast :: SQLiteStore -> TVar ChaChaDRG -> BroadcastId -> m BroadcastId
|
||||
createBcast SQLiteStore {dbConn} gVar bcastId = liftIOEither $ case bcastId of
|
||||
"" -> createWithRandomId gVar create
|
||||
bId -> checkConstraint SEBcastDuplicate $ create bId $> bId
|
||||
where
|
||||
create bId = DB.execute dbConn "INSERT INTO broadcasts (broadcast_id) VALUES (?);" (Only bId)
|
||||
|
||||
addBcastConn :: SQLiteStore -> BroadcastId -> ConnId -> m ()
|
||||
addBcastConn SQLiteStore {dbConn} bId connId =
|
||||
liftIOEither . checkBroadcast dbConn bId $
|
||||
getConn_ dbConn connId >>= \case
|
||||
Left _ -> pure $ Left SEConnNotFound
|
||||
Right (SomeConn _ RcvConnection {}) -> pure . Left $ SEBadConnType CRcv
|
||||
Right _ ->
|
||||
checkConstraint SEConnDuplicate $
|
||||
DB.execute
|
||||
dbConn
|
||||
[sql|
|
||||
INSERT INTO broadcast_connections
|
||||
(broadcast_id, conn_alias) VALUES (?, ?);
|
||||
|]
|
||||
(bId, connId)
|
||||
|
||||
removeBcastConn :: SQLiteStore -> BroadcastId -> ConnId -> m ()
|
||||
removeBcastConn SQLiteStore {dbConn} bId connId =
|
||||
liftIOEither . checkBroadcast dbConn bId $
|
||||
bcastConnExists_ dbConn bId connId >>= \case
|
||||
False -> pure $ Left SEConnNotFound
|
||||
_ ->
|
||||
Right
|
||||
<$> DB.execute
|
||||
dbConn
|
||||
[sql|
|
||||
DELETE FROM broadcast_connections
|
||||
WHERE broadcast_id = ? AND conn_alias = ?;
|
||||
|]
|
||||
(bId, connId)
|
||||
|
||||
deleteBcast :: SQLiteStore -> BroadcastId -> m ()
|
||||
deleteBcast SQLiteStore {dbConn} bId =
|
||||
liftIOEither . checkBroadcast dbConn bId $
|
||||
Right <$> DB.execute dbConn "DELETE FROM broadcasts WHERE broadcast_id = ?;" (Only bId)
|
||||
|
||||
getBcast :: SQLiteStore -> BroadcastId -> m [ConnId]
|
||||
getBcast SQLiteStore {dbConn} bId =
|
||||
liftIOEither . checkBroadcast dbConn bId $
|
||||
Right . map fromOnly
|
||||
<$> DB.query dbConn "SELECT conn_alias FROM broadcast_connections WHERE broadcast_id = ?;" (Only bId)
|
||||
|
||||
createIntro :: SQLiteStore -> TVar ChaChaDRG -> NewIntroduction -> m IntroId
|
||||
createIntro SQLiteStore {dbConn} gVar NewIntroduction {toConn, reConn, reInfo} =
|
||||
liftIOEither . createWithRandomId gVar $ \introId ->
|
||||
@@ -380,7 +326,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
|
||||
Right $ Introduction {introId, toConn, toInfo, toStatus, reConn, reInfo, reStatus, qInfo}
|
||||
intro _ = Left SEIntroNotFound
|
||||
|
||||
addIntroInvitation :: SQLiteStore -> IntroId -> EntityInfo -> SMPQueueInfo -> m ()
|
||||
addIntroInvitation :: SQLiteStore -> IntroId -> ConnInfo -> SMPQueueInfo -> m ()
|
||||
addIntroInvitation SQLiteStore {dbConn} introId toInfo qInfo =
|
||||
liftIO $
|
||||
DB.executeNamed
|
||||
@@ -423,7 +369,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
|
||||
(reStatus, introId)
|
||||
|
||||
createInvitation :: SQLiteStore -> TVar ChaChaDRG -> NewInvitation -> m InvitationId
|
||||
createInvitation SQLiteStore {dbConn} gVar NewInvitation {viaConn, externalIntroId, entityInfo, qInfo} =
|
||||
createInvitation SQLiteStore {dbConn} gVar NewInvitation {viaConn, externalIntroId, connInfo, qInfo} =
|
||||
liftIOEither . createWithRandomId gVar $ \invId ->
|
||||
DB.execute
|
||||
dbConn
|
||||
@@ -431,7 +377,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
|
||||
INSERT INTO conn_invitations
|
||||
(inv_id, via_conn, external_intro_id, conn_info, queue_info) VALUES (?, ?, ?, ?, ?);
|
||||
|]
|
||||
(invId, viaConn, externalIntroId, entityInfo, qInfo)
|
||||
(invId, viaConn, externalIntroId, connInfo, qInfo)
|
||||
|
||||
getInvitation :: SQLiteStore -> InvitationId -> m Invitation
|
||||
getInvitation SQLiteStore {dbConn} invId =
|
||||
@@ -446,8 +392,8 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
|
||||
|]
|
||||
(Only invId)
|
||||
where
|
||||
invitation [(viaConn, externalIntroId, entityInfo, qInfo, connId, status)] =
|
||||
Right $ Invitation {invId, viaConn, externalIntroId, entityInfo, qInfo, connId, status}
|
||||
invitation [(viaConn, externalIntroId, connInfo, qInfo, connId, status)] =
|
||||
Right $ Invitation {invId, viaConn, externalIntroId, connInfo, qInfo, connId, status}
|
||||
invitation _ = Left SEInvitationNotFound
|
||||
|
||||
addInvitationConn :: SQLiteStore -> InvitationId -> ConnId -> m ()
|
||||
@@ -475,8 +421,8 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
|
||||
(Only cId)
|
||||
>>= fmap join . traverse getViaConn . invitation
|
||||
where
|
||||
invitation [(invId, viaConn, externalIntroId, entityInfo, qInfo, status)] =
|
||||
Just $ Invitation {invId, viaConn, externalIntroId, entityInfo, qInfo, connId = Just cId, status}
|
||||
invitation [(invId, viaConn, externalIntroId, connInfo, qInfo, status)] =
|
||||
Just $ Invitation {invId, viaConn, externalIntroId, connInfo, qInfo, connId = Just cId, status}
|
||||
invitation _ = Nothing
|
||||
getViaConn :: Invitation -> IO (Maybe (Invitation, Connection 'CDuplex))
|
||||
getViaConn inv@Invitation {viaConn} = fmap (inv,) . duplexConn <$> getConn_ dbConn viaConn
|
||||
@@ -926,34 +872,6 @@ updateHashSnd_ dbConn connId SndMsgData {..} =
|
||||
":last_internal_snd_msg_id" := internalSndId
|
||||
]
|
||||
|
||||
-- * Broadcast helpers
|
||||
|
||||
checkBroadcast :: DB.Connection -> BroadcastId -> IO (Either StoreError a) -> IO (Either StoreError a)
|
||||
checkBroadcast dbConn bId action =
|
||||
withTransaction dbConn $ do
|
||||
ok <- bcastExists_ dbConn bId
|
||||
if ok then action else pure $ Left SEBcastNotFound
|
||||
|
||||
bcastExists_ :: DB.Connection -> BroadcastId -> IO Bool
|
||||
bcastExists_ dbConn bId = not . null <$> queryBcast
|
||||
where
|
||||
queryBcast :: IO [Only BroadcastId]
|
||||
queryBcast = DB.query dbConn "SELECT broadcast_id FROM broadcasts WHERE broadcast_id = ?;" (Only bId)
|
||||
|
||||
bcastConnExists_ :: DB.Connection -> BroadcastId -> ConnId -> IO Bool
|
||||
bcastConnExists_ dbConn bId connId = not . null <$> queryBcastConn
|
||||
where
|
||||
queryBcastConn :: IO [(BroadcastId, ConnId)]
|
||||
queryBcastConn =
|
||||
DB.query
|
||||
dbConn
|
||||
[sql|
|
||||
SELECT broadcast_id, conn_alias
|
||||
FROM broadcast_connections
|
||||
WHERE broadcast_id = ? AND conn_alias = ?;
|
||||
|]
|
||||
(bId, connId)
|
||||
|
||||
-- create record with a random ID
|
||||
|
||||
getConnId_ :: DB.Connection -> TVar ChaChaDRG -> ConnData -> IO (Either StoreError ConnId)
|
||||
|
||||
+73
-170
@@ -8,7 +8,6 @@
|
||||
{-# LANGUAGE PostfixOperators #-}
|
||||
{-# LANGUAGE RankNTypes #-}
|
||||
{-# LANGUAGE ScopedTypeVariables #-}
|
||||
{-# LANGUAGE TupleSections #-}
|
||||
|
||||
module AgentTests where
|
||||
|
||||
@@ -18,7 +17,6 @@ import Data.ByteString.Char8 (ByteString)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
import SMPAgentClient
|
||||
import Simplex.Messaging.Agent.Protocol
|
||||
import Simplex.Messaging.Agent.Store (InvitationId)
|
||||
import Simplex.Messaging.Protocol (ErrorType (..), MsgBody)
|
||||
import Simplex.Messaging.Transport (ATransport (..), TProxy (..), Transport (..))
|
||||
import System.Timeout
|
||||
@@ -46,58 +44,41 @@ agentTests (ATransport t) = do
|
||||
smpAgentTest3_1_1 $ testSubscription t
|
||||
it "should send notifications to client when server disconnects" $
|
||||
smpAgentServerTest $ testSubscrNotification t
|
||||
describe "Broadcast" do
|
||||
it "should create broadcast and send messages" $
|
||||
smpAgentTest3 $ testBroadcast t
|
||||
it "should create broadcast and send messages (random IDs)" $
|
||||
smpAgentTest3 $ testBroadcastRandomIds t
|
||||
describe "Introduction" do
|
||||
it "should send and accept introduction" $
|
||||
smpAgentTest3 $ testIntroduction t
|
||||
it "should send and accept introduction (random IDs)" $
|
||||
smpAgentTest3 $ testIntroductionRandomIds t
|
||||
|
||||
type TestTransmission p = (ACorrId, ByteString, APartyCmd p)
|
||||
|
||||
type TestTransmission' p c = (ACorrId, ByteString, ACommand p c)
|
||||
|
||||
type TestTransmissionOrError p = (ACorrId, ByteString, Either AgentErrorType (APartyCmd p))
|
||||
|
||||
testTE :: ATransmissionOrError p -> TestTransmissionOrError p
|
||||
testTE (ATransmissionOrError corrId entity cmdOrErr) =
|
||||
(corrId,serializeEntity entity,) $ case cmdOrErr of
|
||||
Right cmd -> Right $ APartyCmd cmd
|
||||
Left e -> Left e
|
||||
|
||||
-- | receive message to handle `h`
|
||||
(<#:) :: Transport c => c -> IO (TestTransmissionOrError 'Agent)
|
||||
(<#:) h = testTE <$> tGet SAgent h
|
||||
(<#:) :: Transport c => c -> IO (ATransmissionOrError 'Agent)
|
||||
(<#:) = tGet SAgent
|
||||
|
||||
-- | send transmission `t` to handle `h` and get response
|
||||
(#:) :: Transport c => c -> (ByteString, ByteString, ByteString) -> IO (TestTransmissionOrError 'Agent)
|
||||
h #: t = tPutRaw h t >> (h <#:)
|
||||
(#:) :: Transport c => c -> (ByteString, ByteString, ByteString) -> IO (ATransmissionOrError 'Agent)
|
||||
h #: t = tPutRaw h t >> (<#:) h
|
||||
|
||||
-- | action and expected response
|
||||
-- `h #:t #> r` is the test that sends `t` to `h` and validates that the response is `r`
|
||||
(#>) :: IO (TestTransmissionOrError 'Agent) -> TestTransmission' 'Agent c -> Expectation
|
||||
action #> (corrId, cAlias, cmd) = action `shouldReturn` (corrId, cAlias, Right (APartyCmd cmd))
|
||||
(#>) :: IO (ATransmissionOrError 'Agent) -> ATransmission 'Agent -> Expectation
|
||||
action #> (corrId, cAlias, cmd) = action `shouldReturn` (corrId, cAlias, Right cmd)
|
||||
|
||||
-- | action and predicate for the response
|
||||
-- `h #:t =#> p` is the test that sends `t` to `h` and validates the response using `p`
|
||||
(=#>) :: IO (TestTransmissionOrError 'Agent) -> (TestTransmission 'Agent -> Bool) -> Expectation
|
||||
(=#>) :: IO (ATransmissionOrError 'Agent) -> (ATransmission 'Agent -> Bool) -> Expectation
|
||||
action =#> p = action >>= (`shouldSatisfy` p . correctTransmission)
|
||||
|
||||
correctTransmission :: TestTransmissionOrError p -> TestTransmission p
|
||||
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
|
||||
(<#) :: Transport c => c -> TestTransmission' 'Agent c' -> Expectation
|
||||
h <# (corrId, cAlias, cmd) = (h <#:) >>= (`shouldBe` (corrId, cAlias, Right (APartyCmd cmd)))
|
||||
(<#) :: Transport c => c -> ATransmission 'Agent -> Expectation
|
||||
h <# (corrId, cAlias, cmd) = (h <#:) >>= (`shouldBe` (corrId, cAlias, Right cmd))
|
||||
|
||||
-- | receive message to handle `h` and validate it using predicate `p`
|
||||
(<#=) :: Transport c => c -> (TestTransmission 'Agent -> Bool) -> Expectation
|
||||
(<#=) :: Transport c => c -> (ATransmission 'Agent -> Bool) -> Expectation
|
||||
h <#= p = (h <#:) >>= (`shouldSatisfy` p . correctTransmission)
|
||||
|
||||
-- | test that nothing is delivered to handle `h` during 10ms
|
||||
@@ -109,42 +90,39 @@ h #:# err = tryGet `shouldReturn` ()
|
||||
Just _ -> error err
|
||||
_ -> return ()
|
||||
|
||||
pattern Sent :: AgentMsgId -> APartyCmd 'Agent
|
||||
pattern Sent msgId <- APartyCmd (SENT msgId)
|
||||
pattern Msg :: MsgBody -> ACommand 'Agent
|
||||
pattern Msg msgBody <- MSG {msgBody, msgIntegrity = MsgOk}
|
||||
|
||||
pattern Msg :: MsgBody -> APartyCmd 'Agent
|
||||
pattern Msg msgBody <- APartyCmd MSG {msgBody, msgIntegrity = MsgOk}
|
||||
-- pattern Inv :: SMPQueueInfo -> Either AgentErrorType (ACommand 'Agent)
|
||||
-- pattern Inv invitation <- Right (INV invitation)
|
||||
|
||||
pattern Inv :: SMPQueueInfo -> Either AgentErrorType (APartyCmd 'Agent)
|
||||
pattern Inv invitation <- Right (APartyCmd (INV invitation))
|
||||
|
||||
pattern Req :: InvitationId -> EntityInfo -> Either AgentErrorType (APartyCmd 'Agent)
|
||||
pattern Req invId eInfo <- Right (APartyCmd (REQ (IE (Conn invId)) eInfo))
|
||||
-- pattern Req :: InvitationId -> ConnInfo -> Either AgentErrorType (ACommand 'Agent)
|
||||
-- pattern Req invId cInfo <- Right (REQ invId cInfo)
|
||||
|
||||
testDuplexConnection :: Transport c => TProxy c -> c -> c -> IO ()
|
||||
testDuplexConnection _ alice bob = do
|
||||
("1", "C:bob", Inv qInfo) <- alice #: ("1", "C:bob", "NEW")
|
||||
("1", "bob", Right (INV qInfo)) <- alice #: ("1", "bob", "NEW")
|
||||
let qInfo' = serializeSmpQueueInfo qInfo
|
||||
bob #: ("11", "C:alice", "JOIN " <> qInfo') #> ("", "C:alice", CON)
|
||||
alice <# ("", "C:bob", CON)
|
||||
alice #: ("2", "C:bob", "SEND :hello") #> ("2", "C:bob", SENT 1)
|
||||
alice #: ("3", "C:bob", "SEND :how are you?") #> ("3", "C:bob", SENT 2)
|
||||
bob <#= \case ("", "C:alice", Msg "hello") -> True; _ -> False
|
||||
bob <#= \case ("", "C:alice", Msg "how are you?") -> True; _ -> False
|
||||
bob #: ("14", "C:alice", "SEND 9\nhello too") #> ("14", "C:alice", SENT 3)
|
||||
alice <#= \case ("", "C:bob", Msg "hello too") -> True; _ -> False
|
||||
bob #: ("15", "C:alice", "SEND 9\nmessage 1") #> ("15", "C:alice", SENT 4)
|
||||
alice <#= \case ("", "C:bob", Msg "message 1") -> True; _ -> False
|
||||
alice #: ("5", "C:bob", "OFF") #> ("5", "C:bob", OK)
|
||||
bob #: ("17", "C:alice", "SEND 9\nmessage 3") #> ("17", "C:alice", ERR (SMP AUTH))
|
||||
alice #: ("6", "C:bob", "DEL") #> ("6", "C:bob", OK)
|
||||
bob #: ("11", "alice", "JOIN " <> qInfo') #> ("", "alice", CON)
|
||||
alice <# ("", "bob", CON)
|
||||
alice #: ("2", "bob", "SEND :hello") #> ("2", "bob", SENT 1)
|
||||
alice #: ("3", "bob", "SEND :how are you?") #> ("3", "bob", SENT 2)
|
||||
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", SENT 3)
|
||||
alice <#= \case ("", "bob", Msg "hello too") -> True; _ -> False
|
||||
bob #: ("15", "alice", "SEND 9\nmessage 1") #> ("15", "alice", SENT 4)
|
||||
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))
|
||||
alice #: ("6", "bob", "DEL") #> ("6", "bob", OK)
|
||||
alice #:# "nothing else should be delivered to alice"
|
||||
|
||||
testDuplexConnRandomIds :: Transport c => TProxy c -> c -> c -> IO ()
|
||||
testDuplexConnRandomIds _ alice bob = do
|
||||
("1", bobConn, Inv qInfo) <- alice #: ("1", "C:", "NEW")
|
||||
("1", bobConn, Right (INV qInfo)) <- alice #: ("1", "", "NEW")
|
||||
let qInfo' = serializeSmpQueueInfo qInfo
|
||||
("", aliceConn, Right (APartyCmd CON)) <- bob #: ("11", "C:", "JOIN " <> qInfo')
|
||||
("", aliceConn, Right CON) <- bob #: ("11", "", "JOIN " <> qInfo')
|
||||
alice <# ("", bobConn, CON)
|
||||
alice #: ("2", bobConn, "SEND :hello") #> ("2", bobConn, SENT 1)
|
||||
alice #: ("3", bobConn, "SEND :how are you?") #> ("3", bobConn, SENT 2)
|
||||
@@ -161,100 +139,26 @@ testDuplexConnRandomIds _ alice bob = do
|
||||
|
||||
testSubscription :: Transport c => TProxy c -> c -> c -> c -> IO ()
|
||||
testSubscription _ alice1 alice2 bob = do
|
||||
("1", "C:bob", Inv qInfo) <- alice1 #: ("1", "C:bob", "NEW")
|
||||
("1", "bob", Right (INV qInfo)) <- alice1 #: ("1", "bob", "NEW")
|
||||
let qInfo' = serializeSmpQueueInfo qInfo
|
||||
bob #: ("11", "C:alice", "JOIN " <> qInfo') #> ("", "C:alice", CON)
|
||||
bob #: ("12", "C:alice", "SEND 5\nhello") #> ("12", "C:alice", SENT 1)
|
||||
bob #: ("13", "C:alice", "SEND 11\nhello again") #> ("13", "C:alice", SENT 2)
|
||||
alice1 <# ("", "C:bob", CON)
|
||||
alice1 <#= \case ("", "C:bob", Msg "hello") -> True; _ -> False
|
||||
alice1 <#= \case ("", "C:bob", Msg "hello again") -> True; _ -> False
|
||||
alice2 #: ("21", "C:bob", "SUB") #> ("21", "C:bob", OK)
|
||||
alice1 <# ("", "C:bob", END)
|
||||
bob #: ("14", "C:alice", "SEND 2\nhi") #> ("14", "C:alice", SENT 3)
|
||||
alice2 <#= \case ("", "C:bob", Msg "hi") -> True; _ -> False
|
||||
bob #: ("11", "alice", "JOIN " <> qInfo') #> ("", "alice", CON)
|
||||
bob #: ("12", "alice", "SEND 5\nhello") #> ("12", "alice", SENT 1)
|
||||
bob #: ("13", "alice", "SEND 11\nhello again") #> ("13", "alice", SENT 2)
|
||||
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", SENT 3)
|
||||
alice2 <#= \case ("", "bob", Msg "hi") -> True; _ -> False
|
||||
alice1 #:# "nothing else should be delivered to alice1"
|
||||
|
||||
testSubscrNotification :: Transport c => TProxy c -> (ThreadId, ThreadId) -> c -> IO ()
|
||||
testSubscrNotification _ (server, _) client = do
|
||||
client #: ("1", "C:conn1", "NEW") =#> \case ("1", "C:conn1", APartyCmd INV {}) -> True; _ -> False
|
||||
client #: ("1", "conn1", "NEW") =#> \case ("1", "conn1", INV {}) -> True; _ -> False
|
||||
client #:# "nothing should be delivered to client before the server is killed"
|
||||
killThread server
|
||||
client <# ("", "C:conn1", END)
|
||||
|
||||
testBroadcast :: forall c. Transport c => TProxy c -> c -> c -> c -> IO ()
|
||||
testBroadcast _ alice bob tom = do
|
||||
-- establish connections
|
||||
(alice, "alice") `connect` (bob, "bob")
|
||||
(alice, "alice") `connect` (tom, "tom")
|
||||
-- create and set up broadcast
|
||||
alice #: ("1", "B:team", "NEW") #> ("1", "B:team", OK)
|
||||
alice #: ("2", "B:team", "ADD C:bob") #> ("2", "B:team", OK)
|
||||
alice #: ("3", "B:team", "ADD C:tom") #> ("3", "B:team", OK)
|
||||
-- commands with errors
|
||||
alice #: ("e1", "B:team", "NEW") #> ("e1", "B:team", ERR $ BCAST B_DUPLICATE)
|
||||
alice #: ("e2", "B:group", "ADD C:bob") #> ("e2", "B:group", ERR $ BCAST B_NOT_FOUND)
|
||||
alice #: ("e3", "B:team", "ADD C:unknown") #> ("e3", "B:team", ERR $ CONN NOT_FOUND)
|
||||
alice #: ("e4", "B:team", "ADD C:bob") #> ("e4", "B:team", ERR $ CONN DUPLICATE)
|
||||
-- send message
|
||||
alice #: ("4", "B:team", "SEND 5\nhello") =#> \case ("4", c, Sent 1) -> c == "C:bob" || c == "C:tom"; _ -> False
|
||||
alice <#= \case ("4", c, Sent 1) -> c == "C:bob" || c == "C:tom"; _ -> False
|
||||
alice <# ("4", "B:team", SENT 0)
|
||||
bob <#= \case ("", "C:alice", Msg "hello") -> True; _ -> False
|
||||
tom <#= \case ("", "C:alice", Msg "hello") -> True; _ -> False
|
||||
-- remove one connection
|
||||
alice #: ("5", "B:team", "REM C:tom") #> ("5", "B:team", OK)
|
||||
alice #: ("6", "B:team", "SEND 11\nhello again") #> ("6", "C:bob", SENT 2)
|
||||
alice <# ("6", "B:team", SENT 0)
|
||||
bob <#= \case ("", "C:alice", Msg "hello again") -> True; _ -> False
|
||||
tom #:# "nothing delivered to tom"
|
||||
-- commands with errors
|
||||
alice #: ("e5", "B:group", "REM C:bob") #> ("e5", "B:group", ERR $ BCAST B_NOT_FOUND)
|
||||
alice #: ("e6", "B:team", "REM C:unknown") #> ("e6", "B:team", ERR $ CONN NOT_FOUND)
|
||||
alice #: ("e7", "B:team", "REM C:tom") #> ("e7", "B:team", ERR $ CONN NOT_FOUND)
|
||||
-- delete broadcast
|
||||
alice #: ("7", "B:team", "DEL") #> ("7", "B:team", OK)
|
||||
alice #: ("8", "B:team", "SEND 11\ntry sending") #> ("8", "B:team", ERR $ BCAST B_NOT_FOUND)
|
||||
-- commands with errors
|
||||
alice #: ("e8", "B:team", "DEL") #> ("e8", "B:team", ERR $ BCAST B_NOT_FOUND)
|
||||
alice #: ("e9", "B:group", "DEL") #> ("e9", "B:group", ERR $ BCAST B_NOT_FOUND)
|
||||
|
||||
testBroadcastRandomIds :: forall c. Transport c => TProxy c -> c -> c -> c -> IO ()
|
||||
testBroadcastRandomIds _ alice bob tom = do
|
||||
-- establish connections
|
||||
(aliceB, bobA) <- alice `connect'` bob
|
||||
(aliceT, tomA) <- alice `connect'` tom
|
||||
-- create and set up broadcast
|
||||
("1", team, Right (APartyCmd OK)) <- alice #: ("1", "B:", "NEW")
|
||||
alice #: ("2", team, "ADD " <> bobA) #> ("2", team, OK)
|
||||
alice #: ("3", team, "ADD " <> tomA) #> ("3", team, OK)
|
||||
-- commands with errors
|
||||
alice #: ("e1", team, "NEW") #> ("e1", team, ERR $ BCAST B_DUPLICATE)
|
||||
alice #: ("e2", "B:group", "ADD " <> bobA) #> ("e2", "B:group", ERR $ BCAST B_NOT_FOUND)
|
||||
alice #: ("e3", team, "ADD C:unknown") #> ("e3", team, ERR $ CONN NOT_FOUND)
|
||||
alice #: ("e4", team, "ADD " <> bobA) #> ("e4", team, ERR $ CONN DUPLICATE)
|
||||
-- send message
|
||||
alice #: ("4", team, "SEND 5\nhello") =#> \case ("4", c, Sent 1) -> c == bobA || c == tomA; _ -> False
|
||||
alice <#= \case ("4", c, Sent 1) -> c == bobA || c == tomA; _ -> False
|
||||
alice <# ("4", team, SENT 0)
|
||||
bob <#= \case ("", c, Msg "hello") -> c == aliceB; _ -> False
|
||||
tom <#= \case ("", c, Msg "hello") -> c == aliceT; _ -> False
|
||||
-- remove one connection
|
||||
alice #: ("5", team, "REM " <> tomA) #> ("5", team, OK)
|
||||
alice #: ("6", team, "SEND 11\nhello again") #> ("6", bobA, SENT 2)
|
||||
alice <# ("6", team, SENT 0)
|
||||
bob <#= \case ("", c, Msg "hello again") -> c == aliceB; _ -> False
|
||||
tom #:# "nothing delivered to tom"
|
||||
-- commands with errors
|
||||
alice #: ("e5", "B:group", "REM " <> bobA) #> ("e5", "B:group", ERR $ BCAST B_NOT_FOUND)
|
||||
alice #: ("e6", team, "REM C:unknown") #> ("e6", team, ERR $ CONN NOT_FOUND)
|
||||
alice #: ("e7", team, "REM " <> tomA) #> ("e7", team, ERR $ CONN NOT_FOUND)
|
||||
-- delete broadcast
|
||||
alice #: ("7", team, "DEL") #> ("7", team, OK)
|
||||
alice #: ("8", team, "SEND 11\ntry sending") #> ("8", team, ERR $ BCAST B_NOT_FOUND)
|
||||
-- commands with errors
|
||||
alice #: ("e8", team, "DEL") #> ("e8", team, ERR $ BCAST B_NOT_FOUND)
|
||||
alice #: ("e9", "B:group", "DEL") #> ("e9", "B:group", ERR $ BCAST B_NOT_FOUND)
|
||||
client <# ("", "conn1", END)
|
||||
|
||||
testIntroduction :: forall c. Transport c => TProxy c -> c -> c -> c -> IO ()
|
||||
testIntroduction _ alice bob tom = do
|
||||
@@ -262,21 +166,21 @@ testIntroduction _ alice bob tom = do
|
||||
(alice, "alice") `connect` (bob, "bob")
|
||||
(alice, "alice") `connect` (tom, "tom")
|
||||
-- send introduction of tom to bob
|
||||
alice #: ("1", "C:bob", "INTRO C:tom 8\nmeet tom") #> ("1", "C:bob", OK)
|
||||
("", "C:alice", Req invId1 "meet tom") <- (bob <#:)
|
||||
bob #: ("2", "C:tom_via_alice", "ACPT C:" <> invId1 <> " 7\nI'm bob") #> ("2", "C:tom_via_alice", OK)
|
||||
("", "C:alice", Req invId2 "I'm bob") <- (tom <#:)
|
||||
alice #: ("1", "bob", "INTRO tom 8\nmeet tom") #> ("1", "bob", OK)
|
||||
("", "alice", Right (REQ invId1 "meet tom")) <- (bob <#:)
|
||||
bob #: ("2", "tom_via_alice", "ACPT " <> invId1 <> " 7\nI'm bob") #> ("2", "tom_via_alice", OK)
|
||||
("", "alice", Right (REQ invId2 "I'm bob")) <- (tom <#:)
|
||||
-- TODO info "tom here" is not used, either JOIN command also should have eInfo parameter
|
||||
-- or this should be another command, not ACPT
|
||||
tom #: ("3", "C:bob_via_alice", "ACPT C:" <> invId2 <> " 8\ntom here") #> ("3", "C:bob_via_alice", OK)
|
||||
tom <# ("", "C:bob_via_alice", CON)
|
||||
bob <# ("", "C:tom_via_alice", CON)
|
||||
alice <# ("", "C:bob", ICON (IE (Conn "tom")))
|
||||
tom #: ("3", "bob_via_alice", "ACPT " <> invId2 <> " 8\ntom here") #> ("3", "bob_via_alice", OK)
|
||||
tom <# ("", "bob_via_alice", CON)
|
||||
bob <# ("", "tom_via_alice", CON)
|
||||
alice <# ("", "bob", ICON "tom")
|
||||
-- they can message each other now
|
||||
tom #: ("4", "C:bob_via_alice", "SEND :hello") #> ("4", "C:bob_via_alice", SENT 1)
|
||||
bob <#= \case ("", "C:tom_via_alice", Msg "hello") -> True; _ -> False
|
||||
bob #: ("5", "C:tom_via_alice", "SEND 9\nhello too") #> ("5", "C:tom_via_alice", SENT 2)
|
||||
tom <#= \case ("", "C:bob_via_alice", Msg "hello too") -> True; _ -> False
|
||||
tom #: ("4", "bob_via_alice", "SEND :hello") #> ("4", "bob_via_alice", SENT 1)
|
||||
bob <#= \case ("", "tom_via_alice", Msg "hello") -> True; _ -> False
|
||||
bob #: ("5", "tom_via_alice", "SEND 9\nhello too") #> ("5", "tom_via_alice", SENT 2)
|
||||
tom <#= \case ("", "bob_via_alice", Msg "hello too") -> True; _ -> False
|
||||
|
||||
testIntroductionRandomIds :: forall c. Transport c => TProxy c -> c -> c -> c -> IO ()
|
||||
testIntroductionRandomIds _ alice bob tom = do
|
||||
@@ -285,17 +189,17 @@ testIntroductionRandomIds _ alice bob tom = do
|
||||
(aliceT, tomA) <- alice `connect'` tom
|
||||
-- send introduction of tom to bob
|
||||
alice #: ("1", bobA, "INTRO " <> tomA <> " 8\nmeet tom") #> ("1", bobA, OK)
|
||||
("", aliceB', Req invId1 "meet tom") <- (bob <#:)
|
||||
("", aliceB', Right (REQ invId1 "meet tom")) <- (bob <#:)
|
||||
aliceB' `shouldBe` aliceB
|
||||
("2", tomB, Right (APartyCmd OK)) <- bob #: ("2", "C:", "ACPT C:" <> invId1 <> " 7\nI'm bob")
|
||||
("", aliceT', Req invId2 "I'm bob") <- (tom <#:)
|
||||
("2", tomB, Right OK) <- bob #: ("2", "C:", "ACPT " <> invId1 <> " 7\nI'm bob")
|
||||
("", aliceT', Right (REQ invId2 "I'm bob")) <- (tom <#:)
|
||||
aliceT' `shouldBe` aliceT
|
||||
-- TODO info "tom here" is not used, either JOIN command also should have eInfo parameter
|
||||
-- or this should be another command, not ACPT
|
||||
("3", bobT, Right (APartyCmd OK)) <- tom #: ("3", "C:", "ACPT C:" <> invId2 <> " 8\ntom here")
|
||||
("3", bobT, Right OK) <- tom #: ("3", "", "ACPT " <> invId2 <> " 8\ntom here")
|
||||
tom <# ("", bobT, CON)
|
||||
bob <# ("", tomB, CON)
|
||||
alice <# ("", bobA, ICON . IE . Conn $ B.drop 2 tomA)
|
||||
alice <# ("", bobA, ICON tomA)
|
||||
-- they can message each other now
|
||||
tom #: ("4", bobT, "SEND :hello") #> ("4", bobT, SENT 1)
|
||||
bob <#= \case ("", c, Msg "hello") -> c == tomB; _ -> False
|
||||
@@ -304,16 +208,16 @@ testIntroductionRandomIds _ alice bob tom = do
|
||||
|
||||
connect :: forall c. Transport c => (c, ByteString) -> (c, ByteString) -> IO ()
|
||||
connect (h1, name1) (h2, name2) = do
|
||||
("c1", _, Inv qInfo) <- h1 #: ("c1", "C:" <> name2, "NEW")
|
||||
("c1", _, Right (INV qInfo)) <- h1 #: ("c1", name2, "NEW")
|
||||
let qInfo' = serializeSmpQueueInfo qInfo
|
||||
h2 #: ("c2", "C:" <> name1, "JOIN " <> qInfo') #> ("", "C:" <> name1, CON)
|
||||
h1 <# ("", "C:" <> name2, CON)
|
||||
h2 #: ("c2", name1, "JOIN " <> qInfo') #> ("", name1, CON)
|
||||
h1 <# ("", name2, CON)
|
||||
|
||||
connect' :: forall c. Transport c => c -> c -> IO (ByteString, ByteString)
|
||||
connect' h1 h2 = do
|
||||
("c1", conn2, Inv qInfo) <- h1 #: ("c1", "C:", "NEW")
|
||||
("c1", conn2, Right (INV qInfo)) <- h1 #: ("c1", "", "NEW")
|
||||
let qInfo' = serializeSmpQueueInfo qInfo
|
||||
("", conn1, Right (APartyCmd CON)) <- h2 #: ("c2", "C:", "JOIN " <> qInfo')
|
||||
("", conn1, Right CON) <- h2 #: ("c2", "", "JOIN " <> qInfo')
|
||||
h1 <# ("", conn2, CON)
|
||||
pure (conn1, conn2)
|
||||
|
||||
@@ -322,25 +226,24 @@ samplePublicKey = "rsa:MIIBoDANBgkqhkiG9w0BAQEFAAOCAY0AMIIBiAKCAQEAtn1NI2tPoOGSG
|
||||
|
||||
syntaxTests :: forall c. Transport c => TProxy c -> Spec
|
||||
syntaxTests t = do
|
||||
it "unknown command" $ ("1", "C:5678", "HELLO") >#> ("1", "C:5678", "ERR CMD SYNTAX")
|
||||
it "unknown command" $ ("1", "5678", "HELLO") >#> ("1", "5678", "ERR CMD SYNTAX")
|
||||
describe "NEW" do
|
||||
describe "valid" do
|
||||
-- TODO: ERROR no connection alias in the response (it does not generate it yet if not provided)
|
||||
-- TODO: add tests with defined connection alias
|
||||
xit "without parameters" $ ("211", "C:", "NEW") >#>= \case ("211", "C:", "INV" : _) -> True; _ -> False
|
||||
it "without parameters" $ ("211", "", "NEW") >#>= \case ("211", _, "INV" : _) -> True; _ -> False
|
||||
describe "invalid" do
|
||||
-- TODO: add tests with defined connection alias
|
||||
it "with parameters" $ ("222", "C:", "NEW hi") >#> ("222", "C:", "ERR CMD SYNTAX")
|
||||
it "with parameters" $ ("222", "", "NEW hi") >#> ("222", "", "ERR CMD SYNTAX")
|
||||
|
||||
describe "JOIN" do
|
||||
describe "valid" do
|
||||
-- TODO: ERROR no connection alias in the response (it does not generate it yet if not provided)
|
||||
-- TODO: add tests with defined connection alias
|
||||
it "using same server as in invitation" $
|
||||
("311", "C:a", "JOIN smp::localhost:5000::1234::" <> samplePublicKey) >#> ("311", "C:a", "ERR SMP AUTH")
|
||||
("311", "a", "JOIN smp::localhost:5000::1234::" <> samplePublicKey) >#> ("311", "a", "ERR SMP AUTH")
|
||||
describe "invalid" do
|
||||
-- TODO: JOIN is not merged yet - to be added
|
||||
it "no parameters" $ ("321", "C:", "JOIN") >#> ("321", "C:", "ERR CMD SYNTAX")
|
||||
it "no parameters" $ ("321", "", "JOIN") >#> ("321", "", "ERR CMD SYNTAX")
|
||||
where
|
||||
-- simple test for one command with the expected response
|
||||
(>#>) :: ARawTransmission -> ARawTransmission -> Expectation
|
||||
|
||||
Reference in new issue
Block a user