hinotify-conduit 0.1.0.0 → 0.1.0.1

raw patch · 4 files changed

+497/−109 lines, 4 filesdep +asyncdep +containersdep +directorydep −mtldep ~basedep ~bytestringdep ~conduitPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: async, containers, directory, filepath, hspec, temporary, unix

Dependencies removed: mtl

Dependency ranges changed: base, bytestring, conduit, filepath-bytestring, hinotify, resourcet, stm, stm-chans, stm-conduit

API changes (from Hackage documentation)

Files

• ChangeLog.md +6/−0
• hinotify-conduit.cabal +59/−58
• src/Data/Conduit/INotify.hs +140/−50
• test/Spec.hs +292/−1

@@ -1,3 +1,9 @@ # Changelog for hinotify-conduit  ## Unreleased changes++## 0.1.0.1++- fix `sourceFileFollowModifyRotateWithSeek` not following after replace+- add `inotifyEventsSourceRotateMultiple`+- fix `sourceFileFollowModifyRotateWithSeek` loop on EOF

@@ -1,68 +1,69 @@-cabal-version: 1.12---- This file has been generated from package.yaml by hpack version 0.34.4.------ see: https://github.com/sol/hpack+cabal-version:      1.12+name:               hinotify-conduit+version:            0.1.0.1+license:            BSD3+license-file:       LICENSE+copyright:          2021 Jiri Marsicek+maintainer:         jiri.marsicek@gmail.com+author:             Jiri Marsicek+homepage:           https://github.com/j1r1k/hinotify-conduit#readme+bug-reports:        https://github.com/j1r1k/hinotify-conduit/issues+synopsis:           inotify conduit sources+description:+    Please see the README on GitHub at <https://github.com/j1r1k/hinotify-conduit#readme> -name:           hinotify-conduit-version:        0.1.0.0-synopsis:       inotify conduit sources-description:    Please see the README on GitHub at <https://github.com/j1r1k/hinotify-conduit#readme>-category:       System,Conduit-homepage:       https://github.com/j1r1k/hinotify-conduit#readme-bug-reports:    https://github.com/j1r1k/hinotify-conduit/issues-author:         Jiri Marsicek-maintainer:     jiri.marsicek@gmail.com-copyright:      2021 Jiri Marsicek-license:        BSD3-license-file:   LICENSE-build-type:     Simple+category:           System,Conduit+build-type:         Simple extra-source-files:     README.md     ChangeLog.md  source-repository head-  type: git-  location: https://github.com/j1r1k/hinotify-conduit+    type:     git+    location: https://github.com/j1r1k/hinotify-conduit  library-  exposed-modules:-      Data.Conduit.INotify-  other-modules:-      Paths_hinotify_conduit-  hs-source-dirs:-      src-  build-depends:-      base >=4.7 && <5-    , bytestring-    , conduit-    , filepath-bytestring-    , hinotify-    , mtl-    , resourcet-    , stm-    , stm-chans-    , stm-conduit-  default-language: Haskell2010+    exposed-modules:  Data.Conduit.INotify+    hs-source-dirs:   src+    other-modules:    Paths_hinotify_conduit+    default-language: Haskell2010+    ghc-options:      -Wall+    build-depends:+        base >=4.9 && <5,+        bytestring >=0.10.12.0,+        conduit >=1.3.4.1,+        containers >=0.6.2.1,+        directory >=1.3.6.0,+        filepath >=1.4.2.1,+        filepath-bytestring >=1.4.2.1.8,+        hinotify >=0.4.1,+        resourcet >=1.2.4.3,+        stm >=2.5.0.0,+        stm-chans >=3.0.0.4,+        stm-conduit >=4.0.1  test-suite hinotify-conduit-test-  type: exitcode-stdio-1.0-  main-is: Spec.hs-  other-modules:-      Paths_hinotify_conduit-  hs-source-dirs:-      test-  ghc-options: -threaded -rtsopts -with-rtsopts=-N-  build-depends:-      base >=4.7 && <5-    , bytestring-    , conduit-    , filepath-bytestring-    , hinotify-    , hinotify-conduit-    , mtl-    , resourcet-    , stm-    , stm-chans-    , stm-conduit-  default-language: Haskell2010+    type:             exitcode-stdio-1.0+    main-is:          Spec.hs+    hs-source-dirs:   test+    other-modules:    Paths_hinotify_conduit+    default-language: Haskell2010+    ghc-options:      -threaded -rtsopts -with-rtsopts=-N -Wall+    build-depends:+        async >=2.2.3,+        base >=4.9 && <5,+        bytestring >=0.10.12.0,+        conduit >=1.3.4.1,+        containers >=0.6.2.1,+        directory >=1.3.6.0,+        filepath >=1.4.2.1,+        filepath-bytestring >=1.4.2.1.8,+        hinotify >=0.4.1,+        hinotify-conduit -any,+        hspec >=2.7.10,+        resourcet >=1.2.4.3,+        stm >=2.5.0.0,+        stm-chans >=3.0.0.4,+        stm-conduit >=4.0.1,+        temporary >=1.3,+        unix >=2.7.2.2

@@ -1,49 +1,54 @@+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE TupleSections #-}+ module Data.Conduit.INotify where  import Conduit (ConduitT, MonadIO, bracketP, lift, liftIO, (.|)) import qualified Conduit as C (await, awaitForever, mapInput, sourceHandle, yield)-import Control.Concurrent.STM (TVar, newTVar, newTVarIO, readTVarIO, writeTVar)+import Control.Concurrent.STM (TVar, modifyTVar, newTVar, newTVarIO, readTVarIO, writeTVar) import Control.Concurrent.STM.TMQueue (TMQueue, closeTMQueue, newTMQueue, writeTMQueue)-import Control.Exception (tryJust)-import Control.Monad.Except (guard) import Control.Monad.STM (STM, atomically) import Control.Monad.Trans.Resource (MonadResource) import Data.ByteString (ByteString)-import qualified Data.ByteString as BS (hGetSome)+import qualified Data.ByteString as BS (hGetSome, null) import qualified Data.ByteString.Lazy.Internal as BS (defaultChunkSize) import qualified Data.Conduit.List as C (catMaybes, map, mapMaybe) import Data.Conduit.TQueue (sourceTMQueue) import Data.Foldable (traverse_)+import Data.List.NonEmpty (NonEmpty ((:|)))+import qualified Data.List.NonEmpty as NonEmpty (toList)+import Data.Map (Map)+import qualified Data.Map as Map (delete, fromList, insert, notMember)+import System.Directory (canonicalizePath)+import System.FilePath (takeDirectory, takeFileName, (</>)) import System.FilePath.ByteString (encodeFilePath)-import qualified System.INotify as INotify (Event (DeletedSelf, Modified), EventVariety (DeleteSelf, Modify), INotify, WatchDescriptor, addWatch, initINotify, killINotify, removeWatch)+import System.FilePath.Posix.ByteString (decodeFilePath)+import System.INotify (filePath)+import qualified System.INotify as INotify (Event (Ignored, Modified, MovedIn), EventVariety (DeleteSelf, Modify, MoveIn), INotify, WatchDescriptor, addWatch, removeWatch) import qualified System.IO as IO (Handle, IOMode (ReadMode), SeekMode (AbsoluteSeek), hClose, hSeek, hTell, openFile)-import qualified System.IO.Error as IO (isEOFError) --- | Run 'ConduitT' with 'INotify'-withINotify :: MonadResource m => (INotify.INotify -> ConduitT a b m r) -> ConduitT a b m r-withINotify = bracketP INotify.initINotify INotify.killINotify- -- | Watch INotify events for given file -- Does not support file rotation. -- Once the watched file is removed, it will not emit any additional events and needs to be terminated via handle. inotifyEventsSource ::   (MonadResource m, Monad m) =>+  -- | INotify+  INotify.INotify ->   -- | events to watch for-  [INotify.EventVariety] ->+  NonEmpty INotify.EventVariety ->   -- | path to file to be watched   FilePath ->   -- | returns (source, handle to terminate the watch)   STM (ConduitT () INotify.Event m (), STM ())-inotifyEventsSource events fp = do+inotifyEventsSource i events fp = do   q <- newTMQueue-  return (withINotify (\i -> bracketP (initialize i q) cleanup (inside q)), closeTMQueue q)+  return (bracketP (initialize q) cleanup (inside q), closeTMQueue q)   where-    initialize i q = INotify.addWatch i events (encodeFilePath fp) (atomically . writeTMQueue q)+    initialize q = INotify.addWatch i (NonEmpty.toList events) (encodeFilePath fp) (atomically . writeTMQueue q)     cleanup = INotify.removeWatch     inside q _ = sourceTMQueue q --- | Stream contents of a 'IO.Handle' as binary data.--- Will yield Nothing after EOF is reached+-- | Stream contents of a 'IO.Handle' as binary data and yield Nothing after EOF is reached sourceHandleEof :: MonadIO m => IO.Handle -> ConduitT () (Maybe ByteString) m () sourceHandleEof h = C.sourceHandle h .| C.map Just <> C.yield Nothing @@ -55,13 +60,15 @@ -- Does not support file rotations. For version supporing rotations see 'sourceFileFollowModifyRotateWithSeek' sourceFileFollowModify ::   (MonadResource m, MonadIO m) =>-  -- patch to file to be followed+  -- | INotify+  INotify.INotify ->+  -- | path to file to be followed   FilePath ->-  -- returns (source of binary data from file, handle to terminate the follow)+  -- | returns (source of binary data from file, handle to terminate the follow)   STM (ConduitT () (Maybe ByteString) m (), STM ())-sourceFileFollowModify fp =+sourceFileFollowModify i fp =   do-    (eventsSource, closeWatch) <- inotifyEventsSource [INotify.Modify] fp+    (eventsSource, closeWatch) <- inotifyEventsSource i (INotify.Modify :| [] {- TODO singleton -}) fp     return (bracketP (IO.openFile fp IO.ReadMode) IO.hClose (inside eventsSource), closeWatch)   where     inside :: MonadIO m => ConduitT () INotify.Event m () -> IO.Handle -> ConduitT () (Maybe ByteString) m ()@@ -71,22 +78,22 @@         <> sourceHandleEof h -- read to the end of the file after the watch ends  -- | Version of 'sourceFileFollowModify' not notifying about EOF-sourceFileFollowModify' :: (MonadResource m, MonadIO m) => FilePath -> STM (ConduitT () ByteString m (), STM ())-sourceFileFollowModify' fp = do-  (source, close) <- sourceFileFollowModify fp+sourceFileFollowModify' :: (MonadResource m, MonadIO m) => INotify.INotify -> FilePath -> STM (ConduitT () ByteString m (), STM ())+sourceFileFollowModify' i fp = do+  (source, close) <- sourceFileFollowModify i fp   return (source .| C.catMaybes, close)  -- | Like 'bracketP', but resource can be released within 'in-between' computation. -- Resource is recreated after release if needed replacableBracketP ::   MonadResource m =>-  -- acquire resource computation+  -- | acquire resource computation   IO a ->-  -- release resource computation+  -- | release resource computation   (a -> IO ()) ->-  -- computation to run in-between.-  -- first: acquires the resource if not available, otherwise just gets it-  -- second: releases the resource+  -- | computation to run in-between.+  -- | first: acquires the resource if not available, otherwise just gets it+  -- | second: releases the resource   ((m a, m ()) -> ConduitT i o m ()) ->   ConduitT i o m () replacableBracketP initialize cleanup inside =@@ -109,16 +116,16 @@  -- | Watch INotify events for given file. -- Interprets file removal as file rotation and tries to recreate the watch again.-inotifyEventsSourceRotate :: MonadResource m => [INotify.EventVariety] -> FilePath -> STM (ConduitT () INotify.Event m (), STM ())-inotifyEventsSourceRotate events fp = do+inotifyEventsSourceRotate :: MonadResource m => INotify.INotify -> NonEmpty INotify.EventVariety -> FilePath -> STM (ConduitT () INotify.Event m (), STM ())+inotifyEventsSourceRotate i events fp = do   q <- newTMQueue-  let c = sourceTMQueue q .| withINotify (\i -> replacableBracketP (initialize i q) cleanup inside)+  let c = sourceTMQueue q .| replacableBracketP (initialize q) cleanup inside   return (c, closeTMQueue q)   where     -- WatchDescriptior is stored within TVar because it destroys itself when the watched file is deleted-    initialize :: INotify.INotify -> TMQueue INotify.Event -> IO (TVar (Maybe INotify.WatchDescriptor))-    initialize i q = do-      w <- INotify.addWatch i (INotify.DeleteSelf : events) (encodeFilePath fp) (atomically . writeTMQueue q)+    initialize :: TMQueue INotify.Event -> IO (TVar (Maybe INotify.WatchDescriptor))+    initialize q = do+      w <- INotify.addWatch i (INotify.DeleteSelf : NonEmpty.toList events) (encodeFilePath fp) (atomically . writeTMQueue q)       newTVarIO $ Just w      cleanup :: TVar (Maybe INotify.WatchDescriptor) -> IO ()@@ -130,7 +137,7 @@       event <- C.await        case event of-        Just e@INotify.DeletedSelf {} -> do+        Just e@INotify.Ignored -> do           C.yield e           liftIO $ atomically $ writeTVar var Nothing -- WatchDescriptor is deleted implicitly           lift unset@@ -141,6 +148,43 @@         Nothing ->           return () +inotifyEventsSourceRotateMultiple :: MonadResource m => INotify.INotify -> [(NonEmpty INotify.EventVariety, FilePath)] -> STM (ConduitT () (INotify.Event, FilePath) m (), STM ())+inotifyEventsSourceRotateMultiple i eventsToFp = do+  q <- newTMQueue+  let c = sourceTMQueue q .| replacableBracketP (initialize q) cleanup (inside q)+  return (c, closeTMQueue q)+  where+    initializeSingle :: TMQueue (INotify.Event, FilePath) -> NonEmpty INotify.EventVariety -> FilePath -> IO INotify.WatchDescriptor+    initializeSingle q events fp =+      INotify.addWatch i (NonEmpty.toList events) (encodeFilePath fp) (atomically . writeTMQueue q . (,fp))++    initialize :: TMQueue (INotify.Event, FilePath) -> IO (TVar (Map FilePath INotify.WatchDescriptor))+    initialize q = do+      ws <- traverse (\(events, fp) -> (fp,) <$> initializeSingle q events fp) eventsToFp+      newTVarIO $ Map.fromList ws++    cleanup :: TVar (Map FilePath INotify.WatchDescriptor) -> IO ()+    cleanup var = readTVarIO var >>= traverse_ INotify.removeWatch++    inside :: MonadIO m => TMQueue (INotify.Event, FilePath) -> (m (TVar (Map FilePath INotify.WatchDescriptor)), m ()) -> ConduitT (INotify.Event, FilePath) (INotify.Event, FilePath) m ()+    inside q (getOrInit, unset) = do+      var <- lift getOrInit+      ws <- liftIO $ readTVarIO var+      -- initialize all WatchDescriptors that might be missing+      liftIO $ traverse_ (\(events, fp) -> if Map.notMember fp ws then initializeSingle q events fp >>= atomically . modifyTVar var . Map.insert fp else pure ()) eventsToFp+      event <- C.await++      case event of+        Just e@(INotify.Ignored, fp) -> do+          C.yield e+          liftIO $ atomically $ modifyTVar var (Map.delete fp) -- WatchDescriptor is deleted implicitly+          inside q (getOrInit, unset)+        Just other -> do+          C.yield other+          inside q (getOrInit, unset)+        Nothing ->+          return ()+ data FollowFileEvent = Replaced | Modified deriving (Eq, Show)  -- | Stream contents of a file as binary data.@@ -149,19 +193,30 @@ -- -- Interprets file removal as file rotation and tries to recreate the watch and continue to follow the file from last position (expects just rotation that resembles append to file). -- Source emits 'Nothing' when EOF is reached. For version emitting just data see 'sourceFileFollowModifyRotateWithSeek\''-sourceFileFollowModifyRotateWithSeek :: (MonadResource m, MonadIO m) => FilePath -> STM (ConduitT () (Maybe ByteString) m (), STM ())-sourceFileFollowModifyRotateWithSeek fp = do-  (eventsSource, closeWatch) <- inotifyEventsSourceRotate [INotify.Modify] fp+--+-- Since the handle prevents the file from deleting, it is watching a parent directory for 'INotify.MoveIn' events and interprets them as rotations+sourceFileFollowModifyRotateWithSeek ::+  (MonadResource m, MonadIO m) =>+  -- | INotify+  INotify.INotify ->+  -- | path to parent directory+  FilePath ->+  -- | file name relative to parent directory+  FilePath ->+  -- | (source, handle to terminate the watch)+  STM (ConduitT () (Maybe ByteString) m (), STM ())+sourceFileFollowModifyRotateWithSeek i parent fp = do+  (eventsSource, closeWatch) <- inotifyEventsSourceRotateMultiple i [(INotify.MoveIn :| [], parent), (INotify.DeleteSelf :| [INotify.Modify], parent </> fp)]   positionVar <- newTVar Nothing   return (eventsSource .| C.mapMaybe handleINotifyEvent .| replacableBracketP (initialize positionVar) cleanup (inside positionVar), closeWatch)   where-    handleINotifyEvent INotify.Modified {} = Just Modified-    handleINotifyEvent INotify.DeletedSelf {} = Just Replaced+    handleINotifyEvent (INotify.Modified {}, fp') = if fp' == parent </> fp then Just Modified else Nothing+    handleINotifyEvent (INotify.MovedIn {filePath = fp'}, parent') = if parent == parent' && fp == decodeFilePath fp' then Just Replaced else Nothing     handleINotifyEvent _ = Nothing      initialize :: TVar (Maybe Integer) -> IO IO.Handle     initialize positionVar = do-      newHandle <- liftIO $ IO.openFile fp IO.ReadMode+      newHandle <- IO.openFile (parent </> fp) IO.ReadMode       maybePosition <- readTVarIO positionVar       traverse_ (IO.hSeek newHandle IO.AbsoluteSeek) maybePosition -- seek to original position       return newHandle@@ -172,13 +227,10 @@     inside :: MonadIO m => TVar (Maybe Integer) -> (m IO.Handle, m ()) -> ConduitT FollowFileEvent (Maybe ByteString) m ()     inside positionVar (getOrInit, unset) = do       handle <- lift getOrInit-      line <- liftIO $ tryJust (guard . IO.isEOFError) $ BS.hGetSome handle BS.defaultChunkSize+      line <- liftIO $ BS.hGetSome handle BS.defaultChunkSize -      case line of-        Right l -> do-          C.yield $ Just l-          inside positionVar (getOrInit, unset)-        Left _ -> do+      if BS.null line+        then do           -- eof reached           C.yield Nothing           event <- C.await@@ -195,9 +247,47 @@             Nothing ->               -- read the file until EOF after the watch is terminated               C.mapInput (const ()) (const event) (sourceHandleEof handle)+        else do+          C.yield $ Just line+          inside positionVar (getOrInit, unset)  -- | Version of 'sourceFileFollowModifyRotateWithSeek' not notifying about EOF-sourceFileFollowModifyRotateWithSeek' :: (MonadResource m, MonadIO m) => FilePath -> STM (ConduitT () ByteString m (), STM ())-sourceFileFollowModifyRotateWithSeek' fp = do-  (source, close) <- sourceFileFollowModifyRotateWithSeek fp+sourceFileFollowModifyRotateWithSeek' ::+  (MonadResource m, MonadIO m) =>+  -- | INotify+  INotify.INotify ->+  -- | path to parent directory+  FilePath ->+  -- | file name relative to parent directory+  FilePath ->+  -- | (source, handle to terminate the watch)+  STM (ConduitT () ByteString m (), STM ())+sourceFileFollowModifyRotateWithSeek' i parent fp = do+  (source, close) <- sourceFileFollowModifyRotateWithSeek i parent fp+  return (source .| C.catMaybes, close)++-- | Version of 'sourceFileFollowModifyRotateWithSeek' that determines parent directory+sourceFileFollowModifyRotateWithSeekIO ::+  (MonadResource m, MonadIO m) =>+  -- | INotify+  INotify.INotify ->+  -- | file name+  FilePath ->+  -- | (source, handle to terminate the watch)+  IO (ConduitT () (Maybe ByteString) m (), STM ())+sourceFileFollowModifyRotateWithSeekIO i fp = do+  absoluteFp <- canonicalizePath fp+  atomically $ sourceFileFollowModifyRotateWithSeek i (takeDirectory absoluteFp) (takeFileName absoluteFp)++-- | Version of 'sourceFileFollowModifyRotateWithSeek\'' that determines parent directory+sourceFileFollowModifyRotateWithSeekIO' ::+  (MonadResource m, MonadIO m) =>+  -- | INotify+  INotify.INotify ->+  -- | file name+  FilePath ->+  -- | (source, handle to terminate the watch)+  IO (ConduitT () ByteString m (), STM ())+sourceFileFollowModifyRotateWithSeekIO' i fp = do+  (source, close) <- sourceFileFollowModifyRotateWithSeekIO i fp   return (source .| C.catMaybes, close)

@@ -1,2 +1,293 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# OPTIONS_GHC -Wno-type-defaults #-}++import Conduit (ConduitT, ResourceT, ZipSink (ZipSink), getZipSink, runConduit, runResourceT, (.|))+import qualified Conduit as C+import Control.Concurrent.Async (Async, async)+import qualified Control.Concurrent.Async as Async (wait)+import Control.Concurrent.STM (TQueue, atomically, newTQueueIO, readTQueue)+import Control.Exception (bracket)+import Data.ByteString (ByteString)+import Data.Conduit.INotify (inotifyEventsSource, inotifyEventsSourceRotate, sourceFileFollowModify, sourceFileFollowModifyRotateWithSeekIO)+import qualified Data.Conduit.List as C (mapMaybe)+import Data.Conduit.TQueue (sinkTQueue)+import Data.Functor (void)+import Data.List.NonEmpty (NonEmpty ((:|)))+import GHC.Conc (threadDelay)+import GHC.IO.Exception (IOErrorType (NoSuchThing))+import System.Directory (copyFile, removeFile)+import System.INotify (Event (isDirectory, maybeFilePath), withINotify)+import qualified System.INotify as INotify (Event (DeletedSelf, Ignored, Modified), EventVariety (AllEvents, Modify))+import System.IO.Error (ioeGetErrorType)+import System.IO.Temp (emptySystemTempFile, withSystemTempFile)+import System.Posix.IO (OpenFileFlags (append))+import qualified System.Posix.IO as IO (OpenMode (WriteOnly), closeFd, defaultFileFlags, fdWrite, openFd)+import System.Timeout (timeout)+import Test.Hspec (describe, hspec, it, shouldReturn, shouldThrow)++isNoSuchThingErrorType :: IOErrorType -> Bool+isNoSuchThingErrorType NoSuchThing = True+isNoSuchThingErrorType _ = False++data AsyncEvent a = AsyncStarted | AsyncEvent a deriving (Eq, Show)++fromAsyncEvent :: AsyncEvent a -> Maybe a+fromAsyncEvent (AsyncEvent a) = Just a+fromAsyncEvent AsyncStarted = Nothing++tempFileTemplate :: String+tempFileTemplate = "hinotify-conduit.spec"++withEmptySystemTempFile :: String -> (FilePath -> IO a) -> IO a+withEmptySystemTempFile template = bracket (emptySystemTempFile template) removeFile++appendToFile :: FilePath -> String -> IO ()+appendToFile fp str =+  bracket+    (IO.openFd fp IO.WriteOnly Nothing IO.defaultFileFlags {append = True})+    IO.closeFd+    (\fd -> void $ IO.fdWrite fd str)++readTQueueTimeout :: TQueue a -> IO (Maybe a)+readTQueueTimeout = timeout (floor 50e5) . atomically . readTQueue++runSource :: ConduitT () a (ResourceT IO) () -> IO (Async [a], TQueue (AsyncEvent a))+runSource source = do+  q <- newTQueueIO+  let sink = getZipSink $ const <$> ZipSink (C.mapMaybe fromAsyncEvent .| C.sinkList) <*> ZipSink (sinkTQueue q)+  asyncHandle <- async $ runResourceT $ runConduit $ source .| (C.yield AsyncStarted >> C.awaitForever (C.yield . AsyncEvent)) .| sink+  -- wait for AsyncStarted event+  Just AsyncStarted <- readTQueueTimeout q+  threadDelay 1000+  return (asyncHandle, q)+ main :: IO ()-main = putStrLn "Test suite not yet implemented"+main = hspec $+  describe "Data.Conduit.INotify" $ do+    describe "inotifyEventsSource" $ do+      it "fails when file does not exist" $+        let computation :: IO () = withINotify $ \i ->+              do+                (source, _) <- atomically $ inotifyEventsSource i (INotify.AllEvents :| []) "somefile"+                runResourceT $ runConduit $ source .| C.sinkNull+         in computation `shouldThrow` (isNoSuchThingErrorType . ioeGetErrorType)++      it "is possible to interrupt the source via handle" $+        let computation :: IO [INotify.Event] = withINotify $ \i -> withSystemTempFile tempFileTemplate $ \fp _ -> do+              (source, close) <- atomically $ inotifyEventsSource i (INotify.Modify :| []) fp+              (asyncHandle, _) <- runSource source+              atomically close+              Async.wait asyncHandle+         in computation `shouldReturn` mempty++      it "emits event on file modification" $+        let computation :: IO [INotify.Event] = withINotify $ \i -> withSystemTempFile tempFileTemplate $ \fp _ -> do+              (source, close) <- atomically $ inotifyEventsSource i (INotify.Modify :| []) fp+              (asyncHandle, q) <- runSource source++              appendToFile fp "String"+              _ <- readTQueueTimeout q++              atomically close+              Async.wait asyncHandle+         in computation `shouldReturn` [INotify.Modified {isDirectory = False, maybeFilePath = Nothing}]++    describe "inotifyEventsSourceRotate" $ do+      it "fails when file does not exist" $+        let computation :: IO () = withINotify $ \i -> do+              (source, _) <- atomically $ inotifyEventsSourceRotate i (INotify.AllEvents :| []) "somefile"+              runResourceT $ runConduit $ source .| C.sinkNull+         in computation `shouldThrow` (isNoSuchThingErrorType . ioeGetErrorType)++      it "is possible to interrupt the source via handle" $+        let computation :: IO [INotify.Event] = withINotify $ \i -> withSystemTempFile tempFileTemplate $ \fp _ -> do+              (source, close) <- atomically $ inotifyEventsSourceRotate i (INotify.Modify :| []) fp+              (asyncHandle, _) <- runSource source+              atomically close+              Async.wait asyncHandle+         in computation `shouldReturn` mempty++      it "emits event on file modification" $+        let computation :: IO [INotify.Event] = withINotify $ \i -> withSystemTempFile tempFileTemplate $ \fp _ -> do+              (source, close) <- atomically $ inotifyEventsSourceRotate i (INotify.Modify :| []) fp+              (asyncHandle, q) <- runSource source++              appendToFile fp "String"+              _ <- readTQueueTimeout q++              atomically close+              Async.wait asyncHandle+         in computation `shouldReturn` [INotify.Modified {isDirectory = False, maybeFilePath = Nothing}]++      it "emits event on file modification after rotate" $+        let computation :: IO [INotify.Event] = withINotify $ \i ->+              withEmptySystemTempFile tempFileTemplate $ \fp1 ->+                withEmptySystemTempFile tempFileTemplate $ \fp2 -> do+                  (source, close) <- atomically $ inotifyEventsSourceRotate i (INotify.Modify :| []) fp1+                  (asyncHandle, q) <- runSource source++                  -- append to watched file+                  appendToFile fp1 "String"+                  _ <- readTQueueTimeout q++                  -- replace watched file with modified copy+                  copyFile fp2 fp1+                  _ <- readTQueueTimeout q+                  _ <- readTQueueTimeout q++                  -- delay needed to initialize watch in async+                  threadDelay 10000++                  -- append to watched file+                  appendToFile fp1 "String"+                  _ <- readTQueueTimeout q++                  threadDelay 10000++                  atomically close+                  Async.wait asyncHandle+         in computation+              `shouldReturn` [ INotify.Modified {isDirectory = False, maybeFilePath = Nothing},+                               INotify.DeletedSelf,+                               INotify.Ignored,+                               INotify.Modified {isDirectory = False, maybeFilePath = Nothing}+                             ]++    describe "sourceFileFollowModify" $ do+      it "fails when file does not exist" $+        let computation :: IO () = withINotify $ \i -> do+              (source, _) <- atomically $ sourceFileFollowModify i "somefile"+              runResourceT $ runConduit $ source .| C.sinkNull+         in computation `shouldThrow` (isNoSuchThingErrorType . ioeGetErrorType)++      it "is possible to interrupt the source via handle" $+        let computation :: IO [Maybe ByteString] = withINotify $ \i -> withEmptySystemTempFile tempFileTemplate $ \fp -> do+              (source, close) <- atomically $ sourceFileFollowModify i fp+              asyncHandle <- async $ runResourceT $ runConduit $ source .| C.sinkList+              atomically close+              Async.wait asyncHandle+         in computation+              `shouldReturn` [ Nothing, --initial eof+                               Nothing -- eof after handle close+                             ]++      it "file is followed after modification is made" $+        let computation :: IO [Maybe ByteString] = withINotify $ \i -> withEmptySystemTempFile tempFileTemplate $ \fp -> do+              appendToFile fp "Initial"++              (source, close) <- atomically $ sourceFileFollowModify i fp+              (asyncHandle, q) <- runSource source+              -- wait for initial eof+              _ <- readTQueueTimeout q++              appendToFile fp "String"++              -- wait for read after inotify event+              _ <- readTQueueTimeout q++              -- wait for eof+              _ <- readTQueueTimeout q++              atomically close+              -- wait for eof after watch is closed+              _ <- readTQueueTimeout q++              Async.wait asyncHandle+         in computation+              `shouldReturn` [ Just "Initial", -- initial data+                               Nothing, -- initial eof+                               Just "String", -- data after append+                               Nothing, -- eof after append+                               Nothing -- eof after watch is closed+                             ]++    describe "sourceFileFollowModifyRotateWithSeekIO" $ do+      it "fails when file does not exist" $+        let computation :: IO () = withINotify $ \i -> do+              (source, _) <- sourceFileFollowModifyRotateWithSeekIO i "somefile"+              runResourceT $ runConduit $ source .| C.sinkNull+         in computation `shouldThrow` (isNoSuchThingErrorType . ioeGetErrorType)++      it "is possible to interrupt the source via handle" $+        let computation :: IO [Maybe ByteString] = withINotify $ \i -> withEmptySystemTempFile tempFileTemplate $ \fp -> do+              (source, close) <- sourceFileFollowModifyRotateWithSeekIO i fp+              asyncHandle <- async $ runResourceT $ runConduit $ source .| C.sinkList+              atomically close+              Async.wait asyncHandle+         in computation+              `shouldReturn` [ Nothing, --initial eof+                               Nothing -- eof after handle close+                             ]++      it "file is followed after modification is made" $+        let computation :: IO [Maybe ByteString] = withINotify $ \i -> withEmptySystemTempFile tempFileTemplate $ \fp -> do+              appendToFile fp "Initial"++              (source, close) <- sourceFileFollowModifyRotateWithSeekIO i fp+              (asyncHandle, q) <- runSource source+              -- wait for initial eof+              _ <- readTQueueTimeout q++              appendToFile fp "String"++              -- wait for read after inotify event+              _ <- readTQueueTimeout q++              -- wait for eof+              _ <- readTQueueTimeout q++              atomically close+              -- wait for eof after watch is closed+              _ <- readTQueueTimeout q++              Async.wait asyncHandle+         in computation+              `shouldReturn` [ Just "Initial", -- initial data+                               Nothing, -- initial eof+                               Just "String", -- data after append+                               Nothing, -- eof after append+                               Nothing -- eof after watch is closed+                             ]++      it "file is followed after rotation is made" $+        let computation :: IO [Maybe ByteString] = withINotify $ \i -> withEmptySystemTempFile tempFileTemplate $ \fp1 ->+              withEmptySystemTempFile tempFileTemplate $ \fp2 -> do+                appendToFile fp1 "Initial"++                (source, close) <- sourceFileFollowModifyRotateWithSeekIO i fp1+                (asyncHandle, q) <- runSource source+                -- wait for initial eof+                _ <- readTQueueTimeout q++                -- make a copy watched file+                copyFile fp1 fp2++                -- append to a copy+                appendToFile fp2 "String"++                -- replace watched file with modified copy+                copyFile fp2 fp1++                -- wait for read after inotify event+                _ <- readTQueueTimeout q++                -- wait for eof+                _ <- readTQueueTimeout q++                -- delay needed to process watch cleanup+                threadDelay 1000++                atomically close+                -- wait for eof after watch is closed+                _ <- readTQueueTimeout q++                Async.wait asyncHandle+         in computation+              `shouldReturn` [ Just "Initial", -- initial data+                               Nothing, -- initial eof+                               Just "String", -- data after rotate+                               Nothing, -- eof after append+                               Nothing -- eof after watch is closed+                             ]