mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-08-30 05:28:22 +00:00
Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0eea37ede4 | ||
|
|
6c1e1f18d4 | ||
|
|
c71f0d7beb | ||
|
|
52aa164f9a |
@@ -1865,11 +1865,13 @@ withWork c doWork getWork action =
|
||||
noWork = liftIO $ noWorkToDo doWork
|
||||
notifyErr err e = atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ err $ show e)
|
||||
|
||||
withWorkItems :: AgentClient -> TMVar () -> (DB.Connection -> IO (Either StoreError [Either StoreError a])) -> (NonEmpty a -> AM ()) -> AM ()
|
||||
withWorkItems c doWork getWork action = do
|
||||
withWorkItems :: String -> AgentClient -> TMVar () -> (DB.Connection -> IO (Either StoreError [Either StoreError a])) -> (NonEmpty a -> AM ()) -> AM ()
|
||||
withWorkItems str c doWork getWork action = do
|
||||
withStore' c getWork >>= \case
|
||||
Right [] -> noWork
|
||||
Right rs -> do
|
||||
let (errs, items) = partitionEithers rs
|
||||
liftIO $ print $ "withWorkItems - " <> str <> " - length items = " <> show (length items) <> ", length errs = " <> show (length errs)
|
||||
case L.nonEmpty items of
|
||||
Just items' -> action items'
|
||||
Nothing -> do
|
||||
|
||||
@@ -59,6 +59,7 @@ runNtfSupervisor :: AgentClient -> AM' ()
|
||||
runNtfSupervisor c = do
|
||||
ns <- asks ntfSupervisor
|
||||
forever $ do
|
||||
liftIO $ print "#################### runNtfSupervisor - in forever loop"
|
||||
cmd <- atomically . readTBQueue $ ntfSubQ ns
|
||||
handleErr . agentOperationBracket c AONtfNetwork waitUntilActive $
|
||||
runExceptT (processNtfCmd c cmd) >>= \case
|
||||
@@ -196,13 +197,14 @@ withTokenServer action = lift getNtfToken >>= mapM_ (\NtfToken {ntfServer} -> ac
|
||||
runNtfWorker :: AgentClient -> NtfServer -> Worker -> AM ()
|
||||
runNtfWorker c srv Worker {doWork} =
|
||||
forever $ do
|
||||
liftIO $ print "#################### runNtfWorker - in forever loop"
|
||||
waitForWork doWork
|
||||
ExceptT $ agentOperationBracket c AONtfNetwork throwWhenInactive $ runExceptT runNtfOperation
|
||||
where
|
||||
runNtfOperation :: AM ()
|
||||
runNtfOperation = do
|
||||
ntfBatchSize <- asks $ ntfBatchSize . config
|
||||
withWorkItems c doWork (\db -> getNextNtfSubNTFActions db srv ntfBatchSize) $ \nextSubs -> do
|
||||
withWorkItems "runNtfWorker" c doWork (\db -> getNextNtfSubNTFActions db srv ntfBatchSize) $ \nextSubs -> do
|
||||
logInfo $ "runNtfWorker - length nextSubs = " <> tshow (length nextSubs)
|
||||
currTs <- liftIO getCurrentTime
|
||||
let (creates, checks, deletes, rotates) = splitActions currTs nextSubs
|
||||
@@ -364,15 +366,18 @@ runNtfWorker c srv Worker {doWork} =
|
||||
|
||||
runNtfSMPWorker :: AgentClient -> SMPServer -> Worker -> AM ()
|
||||
runNtfSMPWorker c srv Worker {doWork} = forever $ do
|
||||
ts <- liftIO getCurrentTime
|
||||
liftIO $ print $ "#################### runNtfSMPWorker - in forever loop - ts = " <> show ts
|
||||
waitForWork doWork
|
||||
ExceptT $ agentOperationBracket c AONtfNetwork throwWhenInactive $ runExceptT runNtfSMPOperation
|
||||
where
|
||||
runNtfSMPOperation :: AM ()
|
||||
runNtfSMPOperation = do
|
||||
ntfBatchSize <- asks $ ntfBatchSize . config
|
||||
withWorkItems c doWork (\db -> getNextNtfSubSMPActions db srv ntfBatchSize) $ \nextSubs -> do
|
||||
withWorkItems "runNtfSMPWorker" c doWork (\db -> getNextNtfSubSMPActions db srv ntfBatchSize) $ \nextSubs -> do
|
||||
logInfo $ "runNtfSMPWorker - length nextSubs = " <> tshow (length nextSubs)
|
||||
let (creates, deletes) = splitActions nextSubs
|
||||
liftIO $ print $ "runNtfSMPWorker - length nextSubs = " <> tshow (length nextSubs) <> ", length creates = " <> tshow (length creates) <> ", length deletes = " <> tshow (length deletes)
|
||||
retrySubActions c creates createNotifierKeys
|
||||
retrySubActions c deletes deleteNotifierKeys
|
||||
splitActions :: NonEmpty (NtfSubSMPAction, NtfSubscription) -> ([NtfSubscription], [NtfSubscription])
|
||||
|
||||
Reference in New Issue
Block a user