mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-08-28 11:34:36 +00:00
cleanup
This commit is contained in:
Evgeny Poberezkin
@@ -38,6 +38,7 @@ module Simplex.Messaging.Server.MsgStore.Journal
|
||||
msgQueueStatePath,
|
||||
readQueueState,
|
||||
newMsgQueueState,
|
||||
getJournalQueueMessages,
|
||||
newJournalId,
|
||||
appendState,
|
||||
queueLogFileName,
|
||||
@@ -47,9 +48,6 @@ module Simplex.Messaging.Server.MsgStore.Journal
|
||||
#if defined(dbServerPostgres)
|
||||
postgresQueueStore,
|
||||
#endif
|
||||
exportJournalMessages,
|
||||
getJournalQueueMessages,
|
||||
encodeMessages,
|
||||
)
|
||||
where
|
||||
|
||||
@@ -59,16 +57,14 @@ import Control.Logger.Simple
|
||||
import Control.Monad
|
||||
import Control.Monad.Trans.Except
|
||||
import qualified Data.Attoparsec.ByteString.Char8 as A
|
||||
import qualified Data.ByteString.Builder as BLD
|
||||
import Data.ByteString.Char8 (ByteString)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
import Data.Char (ord)
|
||||
import Data.Either (fromRight, partitionEithers)
|
||||
import Data.Functor (($>))
|
||||
import Data.Int (Int64)
|
||||
import Data.List (sort, sortBy)
|
||||
import Data.List (sort)
|
||||
import qualified Data.Map.Strict as M
|
||||
import Data.Maybe (catMaybes, fromMaybe, isJust, isNothing, mapMaybe)
|
||||
import Data.Maybe (fromMaybe, isJust, isNothing, mapMaybe)
|
||||
import Data.Text (Text)
|
||||
import qualified Data.Text as T
|
||||
import Data.Text.Encoding (decodeLatin1)
|
||||
@@ -80,7 +76,6 @@ import Simplex.Messaging.Agent.Client (getMapLock)
|
||||
import Simplex.Messaging.Agent.Lock
|
||||
import Simplex.Messaging.Encoding.String
|
||||
import Simplex.Messaging.Protocol
|
||||
import Simplex.Messaging.Server.MsgStore
|
||||
import Simplex.Messaging.Server.MsgStore.Journal.SharedLock
|
||||
import Simplex.Messaging.Server.MsgStore.Types
|
||||
import Simplex.Messaging.Server.QueueStore
|
||||
@@ -93,9 +88,8 @@ import Simplex.Messaging.TMap (TMap)
|
||||
import qualified Simplex.Messaging.TMap as TM
|
||||
import Simplex.Messaging.Util (ifM, tshow, whenM, ($>>=), (<$$>))
|
||||
import System.Directory
|
||||
import System.Exit (exitFailure)
|
||||
import System.FilePath (takeFileName, (</>))
|
||||
import System.IO (BufferMode (..), Handle, IOMode (..), SeekMode (..), stdout)
|
||||
import System.IO (BufferMode (..), Handle, IOMode (..), SeekMode (..))
|
||||
import qualified System.IO as IO
|
||||
import System.Random (StdGen, genByteString, newStdGen)
|
||||
|
||||
@@ -492,7 +486,7 @@ instance MsgStoreClass (JournalMsgStore s) where
|
||||
where
|
||||
newQ = do
|
||||
let dir = msgQueueDirectory ms rId
|
||||
statePath = msgQueueStatePath dir $ B.unpack (strEncode rId)
|
||||
statePath = msgQueueStatePath dir rId
|
||||
queue = JMQueue {queueDirectory = dir, statePath}
|
||||
q <- ifM (doesDirectoryExist dir) (openMsgQueue ms queue forWrite) (createQ queue)
|
||||
atomically $ writeTVar msgQueue' $ Just q
|
||||
@@ -820,8 +814,8 @@ msgQueueDirectory JournalMsgStore {config = JournalStoreConfig {storePath, pathP
|
||||
let (seg, s') = B.splitAt 2 s
|
||||
in seg : splitSegments (n - 1) s'
|
||||
|
||||
msgQueueStatePath :: FilePath -> String -> FilePath
|
||||
msgQueueStatePath dir queueId = dir </> (queueLogFileName <> "." <> queueId <> logFileExt)
|
||||
msgQueueStatePath :: FilePath -> RecipientId -> FilePath
|
||||
msgQueueStatePath dir rId = dir </> (queueLogFileName <> "." <> B.unpack (strEncode rId) <> logFileExt)
|
||||
|
||||
createNewJournal :: FilePath -> ByteString -> IO Handle
|
||||
createNewJournal dir journalId = do
|
||||
@@ -1045,104 +1039,29 @@ hClose h =
|
||||
closeOnException :: Handle -> IO a -> IO a
|
||||
closeOnException h a = a `E.onException` hClose h
|
||||
|
||||
exportJournalMessages :: Bool -> JournalMsgStore s -> Handle -> IO Int
|
||||
exportJournalMessages tty ms@JournalMsgStore {config} h = ifM (doesDirectoryExist storePath) exportStore (pure 0)
|
||||
where
|
||||
exportStore = do
|
||||
(!qCount, !msgCount) <- foldQueues 0 exportQueueMessages (0, 0) ("", storePath)
|
||||
putStrLn $ progress qCount
|
||||
pure msgCount
|
||||
JournalStoreConfig {storePath, pathParts} = config
|
||||
exportQueueMessages :: (Int, Int) -> (String, FilePath) -> IO (Int, Int)
|
||||
exportQueueMessages (!i, !count) (queueId, dir) = do
|
||||
let i' = i + 1
|
||||
when (tty && i' `mod` 100000 == 0) $ putStr (progress i' <> "\r") >> IO.hFlush stdout
|
||||
case strDecode $ B.pack queueId of
|
||||
Right rId -> do
|
||||
let statePath = msgQueueStatePath dir queueId
|
||||
msgs <-
|
||||
readQueueState ms statePath >>= \case
|
||||
(Just MsgQueueState {readState = rs, writeState = ws, size}, _)
|
||||
| size == 0 -> pure []
|
||||
| journalId rs == journalId ws ->
|
||||
getMsgs rs (bytePos rs) (bytePos ws - bytePos rs)
|
||||
| otherwise ->
|
||||
(++)
|
||||
<$> getMsgs rs (byteCount rs - bytePos rs) (bytePos rs)
|
||||
<*> getMsgs ws (0 :: Int64) (bytePos ws)
|
||||
where
|
||||
getMsgs :: JournalState t -> Int64 -> Int64 -> IO [Message]
|
||||
getMsgs js pos sz
|
||||
| sz > 0 = IO.withFile f ReadWriteMode $ \h' -> do
|
||||
IO.hSeek h' AbsoluteSeek $ fromIntegral pos
|
||||
parseMsgs f =<< B.hGet h' (fromIntegral sz)
|
||||
| otherwise = pure []
|
||||
where
|
||||
f = journalFilePath dir $ journalId js
|
||||
parseMsgs f s = do
|
||||
let (errs, msgs) = partitionEithers $ map (strDecode @Message) $ B.lines s
|
||||
unless (null errs) $ putStrLn $ "Error reading " <> show (length errs) <> " messages from " <> f
|
||||
pure msgs
|
||||
_ -> pure []
|
||||
unless (null msgs) $ BLD.hPutBuilder h $ encodeMessages rId msgs
|
||||
pure (i', count + length msgs)
|
||||
Left e -> do
|
||||
logError $ "STORE: exportQueueMessages, message queue directory " <> T.pack dir <> " is invalid, " <> tshow e
|
||||
exitFailure
|
||||
progress i = "Processed: " <> show i <> " queues"
|
||||
foldQueues depth f acc (queueId, path) = do
|
||||
let f' = if depth == pathParts - 1 then f else foldQueues (depth + 1) f
|
||||
listDirs >>= foldM f' acc
|
||||
where
|
||||
listDirs = fmap catMaybes . mapM queuePath . sortBy compareB64 =<< listDirectory path
|
||||
compareB64 [] [] = EQ
|
||||
compareB64 [] _ = LT
|
||||
compareB64 _ [] = GT
|
||||
compareB64 (c : cs) (c' : cs')
|
||||
| c == c' = compareB64 cs cs'
|
||||
| otherwise = compare (charValue c) (charValue c')
|
||||
charValue '-' = 62
|
||||
charValue '_' = 63
|
||||
charValue c
|
||||
| c >= 'A' && c <= 'Z' = ord c - ord 'A'
|
||||
| c >= 'a' && c <= 'z' = 26 + ord c - ord 'a'
|
||||
| c >= '0' && c <= '9' = 52 + ord c - ord '0'
|
||||
| otherwise = -1
|
||||
queuePath dir = do
|
||||
let !path' = path </> dir
|
||||
!queueId' = queueId <> dir
|
||||
ifM
|
||||
(doesDirectoryExist path')
|
||||
(pure $ Just (queueId', path'))
|
||||
(Nothing <$ putStrLn ("Error: path " <> path' <> " is not a directory, skipping"))
|
||||
|
||||
getJournalQueueMessages :: JournalMsgStore s -> JournalQueue s -> IO [Message]
|
||||
getJournalQueueMessages ms q = do
|
||||
let rId = recipientId' q
|
||||
dir = msgQueueDirectory ms rId
|
||||
statePath = msgQueueStatePath dir $ B.unpack (strEncode rId)
|
||||
readQueueState ms statePath >>= \case
|
||||
getJournalQueueMessages ms q =
|
||||
readQueueState ms (msgQueueStatePath dir rId) >>= \case
|
||||
(Just MsgQueueState {readState = rs, writeState = ws, size}, _)
|
||||
| size == 0 -> pure []
|
||||
| journalId rs == journalId ws -> do
|
||||
let f = journalFilePath dir $ journalId rs
|
||||
s <- B.readFile f
|
||||
parseMsgs f $ B.take (bytePos' ws - bytePos' rs) $ B.drop (bytePos' rs) s
|
||||
| otherwise -> do
|
||||
let rf = journalFilePath dir $ journalId rs
|
||||
wf = journalFilePath dir $ journalId ws
|
||||
r <- B.readFile rf
|
||||
w <- B.readFile wf
|
||||
rMsgs <- parseMsgs rf $ B.take (fromIntegral $ byteCount rs) $ B.drop (bytePos' rs) r
|
||||
wMsgs <- parseMsgs wf $ B.take (bytePos' ws) w
|
||||
pure $ rMsgs ++ wMsgs
|
||||
msgs <- getMsgs rs (bytePos rs) (byteCount rs - bytePos rs)
|
||||
if journalId rs == journalId ws
|
||||
then pure msgs
|
||||
else (msgs ++) <$> getMsgs ws (0 :: Int64) (bytePos ws)
|
||||
_ -> pure []
|
||||
where
|
||||
bytePos' = fromIntegral . bytePos
|
||||
parseMsgs f s = do
|
||||
let (errs, msgs) = partitionEithers $ map strDecode $ B.lines s
|
||||
unless (null errs) $ putStrLn $ "Error reading " <> show (length errs) <> " messages from " <> f
|
||||
pure msgs
|
||||
|
||||
encodeMessages :: RecipientId -> [Message] -> BLD.Builder
|
||||
encodeMessages rId = mconcat . map (\msg -> BLD.byteString (strEncode $ MLRv3 rId msg) <> BLD.char8 '\n')
|
||||
rId = recipientId' q
|
||||
dir = msgQueueDirectory ms rId
|
||||
getMsgs :: JournalState t -> Int64 -> Int64 -> IO [Message]
|
||||
getMsgs js pos sz
|
||||
| sz > 0 = IO.withFile f ReadWriteMode $ \h' -> do
|
||||
IO.hSeek h' AbsoluteSeek $ fromIntegral pos
|
||||
parseMsgs =<< B.hGet h' (fromIntegral sz)
|
||||
| otherwise = pure []
|
||||
where
|
||||
f = journalFilePath dir $ journalId js
|
||||
parseMsgs s = do
|
||||
let (errs, msgs) = partitionEithers $ map (strDecode @Message) $ B.lines s
|
||||
unless (null errs) $ putStrLn $ "Error reading " <> show (length errs) <> " messages from " <> f
|
||||
pure msgs
|
||||
|
||||
@@ -84,6 +84,7 @@ data MessageStats = MessageStats
|
||||
expiredMsgsCount :: Int,
|
||||
storedQueues :: Int
|
||||
}
|
||||
deriving (Show)
|
||||
|
||||
instance Monoid MessageStats where
|
||||
mempty = MessageStats 0 0 0
|
||||
|
||||
Reference in New Issue
Block a user