smp server: batch commands (#1559)

* protocol: refactor types and encoding

* clean

* smp server: batch commands (#1560)

* smp server: batch commands verification into one DB transaction

* ghc 8.10.7

* flatten transmission tuples

* diff

* only use batch logic if there is more than one transmission

* func

* reset NTF service when adding notifier

* version

* Revert "smp server: use separate database pool for reading queues and creating service records (#1561)"

This reverts commit 3df2425.

* version

* Revert "version"

This reverts commit d80a6b7.

Author: epoberezkin

src/Simplex/FileTransfer/Client.hs

@@ -204,7 +204,7 @@ sendXFTPTransmission XFTPClient {config, thParams, http2Client} t chunkSpec_ = d
   HTTP2Response {respBody = body@HTTP2Body {bodyHead}} <- withExceptT xftpClientError . ExceptT $ sendRequest http2Client req (Just reqTimeout)
   when (B.length bodyHead /= xftpBlockSize) $ throwE $ PCEResponseError BLOCK
   -- TODO validate that the file ID is the same as in the request?
-  (_, _, (_, _fId, respOrErr)) <- liftEither . first PCEResponseError $ xftpDecodeTransmission thParams bodyHead
+  (_, _fId, respOrErr) <-liftEither $ first PCEResponseError $ xftpDecodeTClient thParams bodyHead
   case respOrErr of
     Right r -> case protocolError r of
       Just e -> throwE $ PCEProtocolError e

src/Simplex/FileTransfer/Protocol.hs

@@ -44,16 +44,18 @@ import Simplex.Messaging.Protocol
     EntityId (..),
     RecipientId,
     SenderId,
+    RawTransmission,
     SentRawTransmission,
-    SignedTransmission,
+    SignedTransmissionOrError,
     SndPublicAuthKey,
     Transmission,
     TransmissionForAuth (..),
     CorrId (..),
     encodeTransmission,
     encodeTransmissionForAuth,
     messageTagP,
-    tDecodeParseValidate,
+    tDecodeServer,
+    tDecodeClient,
     tEncodeBatch1,
     tParse,
   )
@@ -197,7 +199,7 @@ instance FilePartyI p => ProtocolEncoding XFTPVersion XFTPErrorType (FileCommand
   fromProtocolError = fromProtocolError @XFTPVersion @XFTPErrorType @FileResponse
   {-# INLINE fromProtocolError #-}
 
-  checkCredentials (auth, _, EntityId fileId, _) cmd = case cmd of
+  checkCredentials auth (EntityId fileId) cmd = case cmd of
     -- FNEW must not have signature and chunk ID
     FNEW {}
       | isNothing auth -> Left $ CMD NO_AUTH
@@ -231,7 +233,7 @@ instance ProtocolEncoding XFTPVersion XFTPErrorType FileCmd where
   fromProtocolError = fromProtocolError @XFTPVersion @XFTPErrorType @FileResponse
   {-# INLINE fromProtocolError #-}
 
-  checkCredentials t (FileCmd p c) = FileCmd p <$> checkCredentials t c
+  checkCredentials tAuth entId (FileCmd p c) = FileCmd p <$> checkCredentials tAuth entId c
   {-# INLINE checkCredentials #-}
 
 instance Encoding FileInfo where
@@ -310,7 +312,7 @@ instance ProtocolEncoding XFTPVersion XFTPErrorType FileResponse where
     PEBlock -> BLOCK
   {-# INLINE fromProtocolError #-}
 
-  checkCredentials (_, _, EntityId entId, _) cmd = case cmd of
+  checkCredentials _ (EntityId entId) cmd = case cmd of
     FRSndIds {} -> noEntity
     -- ERR response does not always have entity ID
     FRErr _ -> Right cmd
@@ -335,25 +337,35 @@ checkParty' c = case testEquality (sFileParty @p) (sFileParty @p') of
   Just Refl -> Just c
   _ -> Nothing
 
-xftpEncodeAuthTransmission :: ProtocolEncoding XFTPVersion e c => THandleParams XFTPVersion 'TClient -> C.APrivateAuthKey -> Transmission c -> Either TransportError ByteString
-xftpEncodeAuthTransmission thParams@THandleParams {thAuth} pKey (corrId, fId, msg) = do
-  let TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth thParams (corrId, fId, msg)
+xftpEncodeAuthTransmission :: ProtocolEncoding XFTPVersion XFTPErrorType c => THandleParams XFTPVersion 'TClient -> C.APrivateAuthKey -> Transmission c -> Either TransportError ByteString
+xftpEncodeAuthTransmission thParams@THandleParams {thAuth} pKey t@(corrId, _, _) = do
+  let TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth thParams t
   xftpEncodeBatch1 . (,tToSend) =<< authTransmission thAuth False (Just pKey) (C.cbNonce $ bs corrId) tForAuth
 
-xftpEncodeTransmission :: ProtocolEncoding XFTPVersion e c => THandleParams XFTPVersion p -> Transmission c -> Either TransportError ByteString
-xftpEncodeTransmission thParams (corrId, fId, msg) = do
-  let t = encodeTransmission thParams (corrId, fId, msg)
-  xftpEncodeBatch1 (Nothing, t)
+xftpEncodeTransmission :: ProtocolEncoding XFTPVersion XFTPErrorType c => THandleParams XFTPVersion p -> Transmission c -> Either TransportError ByteString
+xftpEncodeTransmission thParams t = xftpEncodeBatch1 (Nothing, encodeTransmission thParams t)
 
 -- this function uses batch syntax but puts only one transmission in the batch
 xftpEncodeBatch1 :: SentRawTransmission -> Either TransportError ByteString
 xftpEncodeBatch1 t = first (const TELargeMsg) $ C.pad (tEncodeBatch1 False t) xftpBlockSize
 
-xftpDecodeTransmission :: ProtocolEncoding XFTPVersion e c => THandleParams XFTPVersion p -> ByteString -> Either XFTPErrorType (SignedTransmission e c)
-xftpDecodeTransmission thParams t = do
+xftpDecodeTServer :: THandleParams XFTPVersion 'TServer -> ByteString -> Either XFTPErrorType (SignedTransmissionOrError XFTPErrorType FileCmd)
+xftpDecodeTServer = xftpDecodeTransmission tDecodeServer
+{-# INLINE xftpDecodeTServer #-}
+
+xftpDecodeTClient :: THandleParams XFTPVersion 'TClient -> ByteString -> Either XFTPErrorType (Transmission (Either XFTPErrorType FileResponse))
+xftpDecodeTClient = xftpDecodeTransmission tDecodeClient
+{-# INLINE xftpDecodeTClient #-}
+
+xftpDecodeTransmission ::
+  (THandleParams XFTPVersion p -> Either TransportError RawTransmission -> r) ->
+  THandleParams XFTPVersion p ->
+  ByteString ->
+  Either XFTPErrorType r
+xftpDecodeTransmission tDecode thParams t = do
   t' <- first (const BLOCK) $ C.unPad t
   case tParse thParams t' of
-    t'' :| [] -> Right $ tDecodeParseValidate thParams t''
+    t'' :| [] -> Right $ tDecode thParams t''
     _ -> Left BLOCK
 
 $(J.deriveJSON (enumJSON $ dropPrefix "F") ''FileParty)

src/Simplex/FileTransfer/Server.hs

@@ -53,7 +53,7 @@ import qualified Simplex.Messaging.Crypto as C
 import qualified Simplex.Messaging.Crypto.Lazy as LC
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Protocol (CorrId (..), BlockingInfo, EntityId (..), RcvPublicAuthKey, RcvPublicDhKey, RecipientId, TAuthorizations, pattern NoEntity)
+import Simplex.Messaging.Protocol (BlockingInfo, EntityId (..), RcvPublicAuthKey, RcvPublicDhKey, RecipientId, SignedTransmission, pattern NoEntity)
 import Simplex.Messaging.Server (dummyVerifyCmd, verifyCmdAuthorization)
 import Simplex.Messaging.Server.Control (CPClientRole (..))
 import Simplex.Messaging.Server.Expiration
@@ -317,22 +317,20 @@ data ServerFile = ServerFile
 processRequest :: XFTPTransportRequest -> M ()
 processRequest XFTPTransportRequest {thParams, reqBody = body@HTTP2Body {bodyHead}, sendResponse}
   | B.length bodyHead /= xftpBlockSize = sendXFTPResponse ("", NoEntity, FRErr BLOCK) Nothing
-  | otherwise = do
-      case xftpDecodeTransmission thParams bodyHead of
-        Right (sig_, signed, (corrId, fId, cmdOrErr)) ->
-          case cmdOrErr of
-            Right cmd -> do
-              let THandleParams {thAuth} = thParams
-              verifyXFTPTransmission ((,C.cbNonce (bs corrId)) <$> thAuth) sig_ signed fId cmd >>= \case
-                VRVerified req -> uncurry send =<< processXFTPRequest body req
-                VRFailed e -> send (FRErr e) Nothing
-            Left e -> send (FRErr e) Nothing
+  | otherwise =
+      case xftpDecodeTServer thParams bodyHead of
+        Right (Right t@(_, _, (corrId, fId, _))) -> do
+          let THandleParams {thAuth} = thParams
+          verifyXFTPTransmission thAuth t >>= \case
+            VRVerified req -> uncurry send =<< processXFTPRequest body req
+            VRFailed e -> send (FRErr e) Nothing
           where
             send resp = sendXFTPResponse (corrId, fId, resp)
+        Right (Left (corrId, fId, e)) -> sendXFTPResponse (corrId, fId, FRErr e) Nothing
         Left e -> sendXFTPResponse ("", NoEntity, FRErr e) Nothing
   where
-    sendXFTPResponse (corrId, fId, resp) serverFile_ = do
-      let t_ = xftpEncodeTransmission thParams (corrId, fId, resp)
+    sendXFTPResponse t' serverFile_ = do
+      let t_ = xftpEncodeTransmission thParams t'
 #ifdef slow_servers
       randomDelay
 #endif
@@ -361,8 +359,8 @@ randomDelay = do
 
 data VerificationResult = VRVerified XFTPRequest | VRFailed XFTPErrorType
 
-verifyXFTPTransmission :: Maybe (THandleAuth 'TServer, C.CbNonce) -> Maybe TAuthorizations -> ByteString -> XFTPFileId -> FileCmd -> M VerificationResult
-verifyXFTPTransmission auth_ tAuth authorized fId cmd =
+verifyXFTPTransmission :: Maybe (THandleAuth 'TServer) -> SignedTransmission FileCmd -> M VerificationResult
+verifyXFTPTransmission thAuth (tAuth, authorized, (corrId, fId, cmd)) =
   case cmd of
     FileCmd SFSender (FNEW file rcps auth') -> pure $ XFTPReqNew file rcps auth' `verifyWith` sndKey file
     FileCmd SFRecipient PING -> pure $ VRVerified XFTPReqPing
@@ -381,9 +379,9 @@ verifyXFTPTransmission auth_ tAuth authorized fId cmd =
                 EntityBlocked info -> VRFailed $ BLOCKED info
                 EntityOff -> noFileAuth
           Left _ -> pure noFileAuth
-        noFileAuth = maybe False (dummyVerifyCmd Nothing authorized) tAuth `seq` VRFailed AUTH
+        noFileAuth = dummyVerifyCmd thAuth tAuth authorized corrId `seq` VRFailed AUTH
     -- TODO verify with DH authorization
-    req `verifyWith` k = if verifyCmdAuthorization auth_ tAuth authorized k then VRVerified req else VRFailed AUTH
+    req `verifyWith` k = if verifyCmdAuthorization thAuth tAuth authorized corrId k then VRVerified req else VRFailed AUTH
 
 processXFTPRequest :: HTTP2Body -> XFTPRequest -> M (FileResponse, Maybe ServerFile)
 processXFTPRequest HTTP2Body {bodyPart} = \case

src/Simplex/Messaging/Agent/Client.hs

@@ -33,6 +33,7 @@ module Simplex.Messaging.Agent.Client
     withConnLocks,
     withInvLock,
     withLockMap,
+    withLocksMap,
     getMapLock,
     ipAddressProtected,
     closeAgentClient,
@@ -1004,16 +1005,16 @@ withInvLock' AgentClient {invLocks} = withLockMap invLocks
 {-# INLINE withInvLock' #-}
 
 withConnLocks :: AgentClient -> Set ConnId -> Text -> AM' a -> AM' a
-withConnLocks AgentClient {connLocks} = withLocksMap_ connLocks
+withConnLocks AgentClient {connLocks} = withLocksMap connLocks
 {-# INLINE withConnLocks #-}
 
 withLockMap :: (Ord k, MonadUnliftIO m) => TMap k Lock -> k -> Text -> m a -> m a
 withLockMap = withGetLock . getMapLock
 {-# INLINE withLockMap #-}
 
-withLocksMap_ :: (Ord k, MonadUnliftIO m) => TMap k Lock -> Set k -> Text -> m a -> m a
-withLocksMap_ = withGetLocks . getMapLock
-{-# INLINE withLocksMap_ #-}
+withLocksMap :: (Ord k, MonadUnliftIO m) => TMap k Lock -> Set k -> Text -> m a -> m a
+withLocksMap = withGetLocks . getMapLock
+{-# INLINE withLocksMap #-}
 
 getMapLock :: Ord k => TMap k Lock -> k -> STM Lock
 getMapLock locks key = TM.lookup key locks >>= maybe newLock pure

src/Simplex/Messaging/Client.hs

@@ -183,7 +183,7 @@ data PClient v err msg = PClient
     clientCorrId :: TVar ChaChaDRG,
     sentCommands :: TMap CorrId (Request err msg),
     sndQ :: TBQueue (Maybe (Request err msg), ByteString),
-    rcvQ :: TBQueue (NonEmpty (SignedTransmission err msg)),
+    rcvQ :: TBQueue (NonEmpty (Transmission (Either err msg))),
     msgQ :: Maybe (TBQueue (ServerTransmissionBatch v err msg))
   }
 
@@ -615,7 +615,7 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
 
     receive :: Transport c => ProtocolClient v err msg -> THandle v c 'TClient -> IO ()
     receive ProtocolClient {client_ = PClient {rcvQ, lastReceived, timeoutErrorCount}} h = forever $ do
-      tGet h >>= atomically . writeTBQueue rcvQ
+      tGetClient h >>= atomically . writeTBQueue rcvQ
       getCurrentTime >>= atomically . writeTVar lastReceived
       atomically $ writeTVar timeoutErrorCount 0
 
@@ -642,14 +642,14 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
     process :: ProtocolClient v err msg -> IO ()
     process c = forever $ atomically (readTBQueue $ rcvQ $ client_ c) >>= processMsgs c
 
-    processMsgs :: ProtocolClient v err msg -> NonEmpty (SignedTransmission err msg) -> IO ()
+    processMsgs :: ProtocolClient v err msg -> NonEmpty (Transmission (Either err msg)) -> IO ()
     processMsgs c ts = do
       ts' <- catMaybes <$> mapM (processMsg c) (L.toList ts)
       forM_ msgQ $ \q ->
         mapM_ (atomically . writeTBQueue q . serverTransmission c) (L.nonEmpty ts')
 
-    processMsg :: ProtocolClient v err msg -> SignedTransmission err msg -> IO (Maybe (EntityId, ServerTransmission err msg))
-    processMsg ProtocolClient {client_ = PClient {sentCommands}} (_, _, (corrId, entId, respOrErr))
+    processMsg :: ProtocolClient v err msg -> Transmission (Either err msg) -> IO (Maybe (EntityId, ServerTransmission err msg))
+    processMsg ProtocolClient {client_ = PClient {sentCommands}} (corrId, entId, respOrErr)
       | B.null $ bs corrId = sendMsg $ STEvent clientResp
       | otherwise =
           TM.lookupIO corrId sentCommands >>= \case
@@ -767,7 +767,7 @@ createSMPQueue ::
   -- Maybe NewNtfCreds ->
   ExceptT SMPClientError IO QueueIdsKeys
 createSMPQueue c nonce_ (rKey, rpKey) dhKey auth subMode qrd =
-  sendProtocolCommand_ c nonce_ Nothing (Just rpKey) NoEntity (Cmd SRecipient $ NEW $ NewQueueReq rKey dhKey auth subMode (Just qrd)) >>= \case
+  sendProtocolCommand_ c nonce_ Nothing (Just rpKey) NoEntity (Cmd SCreator $ NEW $ NewQueueReq rKey dhKey auth subMode (Just qrd)) >>= \case
     IDS qik -> pure qik
     r -> throwE $ unexpectedResponse r
 
@@ -848,7 +848,7 @@ nsubResponse_ = \case
   r' -> Left $ unexpectedResponse r'
 {-# INLINE nsubResponse_ #-}
 
-subscribeService :: forall p. (PartyI p, SubscriberParty p) => SMPClient -> SParty p -> ExceptT SMPClientError IO Int64
+subscribeService :: forall p. (PartyI p, ServiceParty p) => SMPClient -> SParty p -> ExceptT SMPClientError IO Int64
 subscribeService c party = case smpClientService c of
   Just THClientService {serviceId, serviceKey} -> do
     liftIO $ enablePings c
@@ -858,8 +858,8 @@ subscribeService c party = case smpClientService c of
     where
       subCmd :: Command p
       subCmd = case party of
-        SRecipient -> SUBS
-        SNotifier -> NSUBS
+        SRecipientService -> SUBS
+        SNotifierService -> NSUBS
   Nothing -> throwE PCEServiceUnavailable
 
 smpClientService :: SMPClient -> Maybe THClientService
@@ -1119,8 +1119,8 @@ proxySMPCommand c@ProtocolClient {thParams = proxyThParams, client_ = PClient {c
         -- server interaction errors are thrown directly
         t' <- liftEitherWith PCECryptoError $ C.cbDecrypt cmdSecret (C.reverseNonce nonce) er
         case tParse serverThParams t' of
-          t'' :| [] -> case tDecodeParseValidate serverThParams t'' of
-            (_auth, _signed, (_c, _e, cmd)) -> case cmd of
+          t'' :| [] -> case tDecodeClient serverThParams t'' of
+            (_, _, cmd) -> case cmd of
               Right (ERR e) -> throwE $ PCEProtocolError e -- this is the error from the destination relay
               Right r' -> pure $ Right r'
               Left e -> throwE $ PCEResponseError e

src/Simplex/Messaging/Client/Agent.hs

@@ -70,9 +70,9 @@ import Simplex.Messaging.Protocol
     QueueId,
     SMPServer,
     SParty (..),
-    SubscriberParty,
-    subscriberParty,
-    subscriberServiceRole
+    ServiceParty,
+    serviceParty,
+    partyServiceRole
   )
 import Simplex.Messaging.Session
 import Simplex.Messaging.TMap (TMap)
@@ -331,11 +331,11 @@ reconnectClient ca@SMPClientAgent {active, agentCfg, smpSubWorkers, workerSeq} s
 
 reconnectSMPClient :: forall p. SMPClientAgent p -> SMPServer -> (Maybe (ServiceId, Int64), Maybe (Map QueueId C.APrivateAuthKey)) -> ExceptT SMPClientError IO ()
 reconnectSMPClient ca@SMPClientAgent {agentCfg, agentParty} srv (sSub_, qSubs_) =
-  withSMP ca srv $ \smp -> liftIO $ case subscriberParty agentParty of
+  withSMP ca srv $ \smp -> liftIO $ case serviceParty agentParty of
     Just Dict -> resubscribe smp
     Nothing -> pure ()
   where
-    resubscribe :: (PartyI p, SubscriberParty p) => SMPClient -> IO ()
+    resubscribe :: (PartyI p, ServiceParty p) => SMPClient -> IO ()
     resubscribe smp = do
       mapM_ (smpSubscribeService ca smp srv) sSub_
       forM_ qSubs_ $ \qSubs -> do
@@ -394,22 +394,22 @@ withSMP ca srv action = (getSMPServerClient' ca srv >>= action) `catchE` logSMPE
       logInfo $ "SMP error (" <> safeDecodeUtf8 (strEncode $ host srv) <> "): " <> tshow e
       throwE e
 
-subscribeQueuesNtfs :: SMPClientAgent 'Notifier -> SMPServer -> NonEmpty (NotifierId, NtfPrivateAuthKey) -> IO ()
+subscribeQueuesNtfs :: SMPClientAgent 'NotifierService -> SMPServer -> NonEmpty (NotifierId, NtfPrivateAuthKey) -> IO ()
 subscribeQueuesNtfs = subscribeQueues_
 {-# INLINE subscribeQueuesNtfs #-}
 
-subscribeQueues_ :: SubscriberParty p => SMPClientAgent p -> SMPServer -> NonEmpty (QueueId, C.APrivateAuthKey) -> IO ()
+subscribeQueues_ :: ServiceParty p => SMPClientAgent p -> SMPServer -> NonEmpty (QueueId, C.APrivateAuthKey) -> IO ()
 subscribeQueues_ ca srv subs = do
   atomically $ addPendingSubs ca srv $ L.toList subs
   runExceptT (getSMPServerClient' ca srv) >>= \case
     Right smp -> smpSubscribeQueues ca smp srv subs
     Left _ -> pure () -- no call to reconnectClient - failing getSMPServerClient' does that
 
-smpSubscribeQueues :: SubscriberParty p => SMPClientAgent p -> SMPClient -> SMPServer -> NonEmpty (QueueId, C.APrivateAuthKey) -> IO ()
+smpSubscribeQueues :: ServiceParty p => SMPClientAgent p -> SMPClient -> SMPServer -> NonEmpty (QueueId, C.APrivateAuthKey) -> IO ()
 smpSubscribeQueues ca smp srv subs = do
   rs <- case agentParty ca of
-    SRecipient -> subscribeSMPQueues smp subs
-    SNotifier -> subscribeSMPQueuesNtfs smp subs
+    SRecipientService -> subscribeSMPQueues smp subs
+    SNotifierService -> subscribeSMPQueuesNtfs smp subs
   rs' <-
     atomically $
       ifM
@@ -454,18 +454,18 @@ smpSubscribeQueues ca smp srv subs = do
     notify_ :: (SMPServer -> NonEmpty a -> SMPClientAgentEvent) -> [a] -> IO ()
     notify_ evt qs = mapM_ (notify ca . evt srv) $ L.nonEmpty qs
 
-subscribeServiceNtfs :: SMPClientAgent 'Notifier -> SMPServer -> (ServiceId, Int64) -> IO ()
+subscribeServiceNtfs :: SMPClientAgent 'NotifierService -> SMPServer -> (ServiceId, Int64) -> IO ()
 subscribeServiceNtfs = subscribeService_
 {-# INLINE subscribeServiceNtfs #-}
 
-subscribeService_ :: (PartyI p, SubscriberParty p) => SMPClientAgent p -> SMPServer -> (ServiceId, Int64) -> IO ()
+subscribeService_ :: (PartyI p, ServiceParty p) => SMPClientAgent p -> SMPServer -> (ServiceId, Int64) -> IO ()
 subscribeService_ ca srv serviceSub = do
   atomically $ setPendingServiceSub ca srv $ Just serviceSub
   runExceptT (getSMPServerClient' ca srv) >>= \case
     Right smp -> smpSubscribeService ca smp srv serviceSub
     Left _ -> pure () -- no call to reconnectClient - failing getSMPServerClient' does that
 
-smpSubscribeService :: (PartyI p, SubscriberParty p) => SMPClientAgent p -> SMPClient -> SMPServer -> (ServiceId, Int64) -> IO ()
+smpSubscribeService :: (PartyI p, ServiceParty p) => SMPClientAgent p -> SMPClient -> SMPServer -> (ServiceId, Int64) -> IO ()
 smpSubscribeService ca smp srv serviceSub@(serviceId, _) = case smpClientService smp of
   Just service | serviceAvailable service -> subscribe
   _ -> notifyUnavailable
@@ -490,7 +490,7 @@ smpSubscribeService ca smp srv serviceSub@(serviceId, _) = case smpClientService
       setActiveServiceSub ca srv $ Just ((serviceId, n), sessId)
       setPendingServiceSub ca srv Nothing
     serviceAvailable THClientService {serviceRole, serviceId = serviceId'} =
-      serviceId == serviceId' && subscriberServiceRole (agentParty ca) == serviceRole
+      serviceId == serviceId' && partyServiceRole (agentParty ca) == serviceRole
     notifyUnavailable = do
       atomically $ setPendingServiceSub ca srv Nothing
       notify ca $ CAServiceUnavailable srv serviceSub -- this will resubscribe all queues directly