@@ -1,28 +1,108 @@
11module Katip.Scribes.RotatingHandler where
22
3−import Control.Concurrent.STM
4−import Control.DeepSeq
5−import Data.ByteString.Lazy as BL
3+import Control.Applicative
4+import Control.Concurrent.Async
5+import Control.Concurrent.STM
6+import Control.DeepSeq
7+import Control.Exception
8+import Control.Monad
9+import Data.ByteString.Lazy as BL
10+import Data.IORef
11+import GHC.IO.Handle
12+import GHC.IO.Handle.FD
13+import GHC.IO.IOMode
614
715-- | File owner struct writing data to the file
816-- sequentially. Automatic buffer flushing and log rotating.
917data FileOwner = FileOwner
10− { foDataQueune :: TBQueue BL.ByteString
18+ { foDataQueune :: TBQueue BL.ByteString
19+ , foControlQueue :: TBQueue ControlMsg
20+ , foAsync :: Async ()
21+ -- ^ Wait for to be sure that worker thread is closed.
22+ }
23+
24+data FileOwnerSettings = FileOwnerSettings
25+ { fosDebounceFreq :: Maybe Int
26+ , fosDataQueueLen :: Int
27+ , fosControlQueueLen :: Int
28+ }
1129
30+defaultFileOwnerSettings :: FileOwnerSettings
31+defaultFileOwnerSettings = FileOwnerSettings
32+ { fosDebounceFreq = 200000 -- every 200ms
33+ , fosDataQueueLen = 1000
34+ , fosControlQueueLen = 100
1235 }
1336
1437data ControlMsg
1538 = CloseMsg
1639 | FlushMsg
1740 | ReopenMsg
41+ deriving (Eq, Ord, Show)
42+
43+newFileOwner :: FilePath -> FileOwnerSettings -> IO FileOwner
44+newFileOwner fp s = do
45+ dqueue <- newTBQueueIO $ fosDataQueueLen s
46+ cqueue <- newTBQueueIO $ fosControlQueueLen s
47+ let
48+ newResource = openBinaryFile fp AppendMode
49+ ack = newResource >>= newIORef
50+ release ref = do
51+ h <- readIORef ref
52+ hFlush h
53+ hClose h
54+ go ref = do
55+ let
56+ readAllData = do
57+ a <- readTBQueue dqueue
58+ -- flushTBQueue never retries
59+ as <- flushTBQueue dqueue
60+ return $ a:as
61+ readMsg
62+ = (Left <$> readTBQueue cqueue)
63+ <|> (Right <$> readAllData)
64+ atomically readMsg >>= \case
65+ Right bs -> do
66+ h <- readIORef ref
67+ BL.hPutStr h $ mconcat bs
68+ -- flush ref
69+ go ref
70+ Left c -> case c of
71+ CloseMsg -> return ()
72+ FlushMsg -> do
73+ h <- readIORef ref
74+ hFlush h
75+ go ref
76+ ReopenMsg -> do
77+ newH <- newResource
78+ oldH <- atomicModifyIORef' ref (\oldH -> (newH, oldH))
79+ hFlush oldH
80+ hClose oldH
81+ go ref
82+ worker :: IO ()
83+ worker = bracket ack release go
84+ asyncRet <- async worker
85+ return $ FileOwner
86+ { foDataQueune = dqueue
87+ , foControlQueue = cqueue
88+ , foAsync = asyncRet
89+ }
90+
1891
19−newFileOwner :: FilePath -> Maybe Int -> IO FileOwner
20−newFileOwner = error "newFileOwner not implemented"
2192
2293fileOwnerControl :: FileOwner -> ControlMsg -> IO ()
23−fileOwnerControl = error "fileOwnerControl not implemented"
94+fileOwnerControl fo msg = do
95+ atomically $ writeTBQueue (foControlQueue fo) msg
96+ when (msg == CloseMsg) $ do
97+ -- Wait for worker thread to finish
98+ void $ waitCatch $ foAsync fo
2499
25100writeFileOwner :: FileOwner -> BL.ByteString -> IO ()
26101writeFileOwner fo bs = do
27102 a <- return $!! bs
103+ -- The deepseq here is to be sure that writing thread will not
104+ -- calculate thunks and will not get into the blackhole or
105+ -- something. Precalculating is the responsibility of the sending
106+ -- thread, because there may be several sending threads and only one
107+ -- writing.
28108 atomically $ writeTBQueue (foDataQueune fo) $!! a