haskell - 无限读取文件

Question

我正在尝试从文件中读取一些不规则的输入(例如，可能会不时出现的命令)。例如。最初的源文件为空，并且我的程序已启动。然后，将一些字符串附加到文件，我的程序必须读取该字符串。

第一个朴素的实现:

import System.IO
import Control.Monad

listen :: Handle -> IO ()
listen file = forever $ do
    ineof <- hIsEOF file
    if ineof
        then do
            s <- hGetLine file
            putStrLn s
        else
            return ()

但这当然不能正常工作(首先是因为性能问题)。如何正确实现此功能(也许使用管道)？

Answer 1

我在下面汇总了实现此示例。基本思想是:

使用fsnotify软件包监视文件更改。

使用sourceFileRange可以流式传输文件中以前未使用的部分。

使用MVar使fsnotify回调向Source发出信号以继续读取。

这假定源文件仅被添加，从未删除或缩短。

import           Control.Concurrent        (forkIO, threadDelay)
import           Control.Concurrent.MVar   (MVar, newEmptyMVar, putMVar,
                                            takeMVar)
import           Control.Exception         (IOException, try)
import           Control.Monad             (forever, void, when)
import           Control.Monad.IO.Class    (liftIO)
import           Data.ByteString           (ByteString)
import qualified Data.ByteString           as S
import           Data.Conduit              (MonadResource, Source, bracketP,
                                            runResourceT, ($$), ($=))
import           Data.Conduit.Binary       (sourceFileRange)
import qualified Data.Conduit.List         as CL
import           Data.IORef                (IORef, modifyIORef, newIORef,
                                            readIORef)
import           Data.Time                 (getCurrentTime)
import           Filesystem                (canonicalizePath)
import           Filesystem.Path.CurrentOS (decodeString, directory)
import           System.FSNotify           (Event (..), startManager,
                                            stopManager, watchDir)

tryIO :: IO a -> IO (Either IOException a)
tryIO = try

sourceFileForever :: MonadResource m => FilePath -> Source m ByteString
sourceFileForever fp' = bracketP startManager stopManager $ \manager -> do
    fp <- liftIO $ canonicalizePath $ decodeString fp'
    baton <- liftIO newEmptyMVar
    liftIO $ watchDir manager (directory fp) (const True) $ \event -> void $ tryIO $ do
        fpE <- canonicalizePath $
            case event of
                Added x _ -> x
                Modified x _ -> x
                Removed x _ -> x
        when (fpE == fp) $ putMVar baton ()
    consumedRef <- liftIO $ newIORef 0
    loop baton consumedRef
  where
    loop :: MonadResource m => MVar () -> IORef Integer -> Source m ByteString
    loop baton consumedRef = forever $ do
        consumed <- liftIO $ readIORef consumedRef
        sourceFileRange fp' (Just consumed) Nothing $= CL.iterM counter
        liftIO $ takeMVar baton
      where
        counter bs = liftIO $ modifyIORef consumedRef (+ fromIntegral (S.length bs))

main :: IO ()
main = do
    let fp = "foo.txt"
    writeFile fp "Hello World!"
    _ <- forkIO $ runResourceT $ sourceFileForever fp $$ CL.mapM_ (liftIO . print)
    forever $ do
        now <- getCurrentTime
        appendFile fp $ show now ++ "\n"
        threadDelay 1000000