packages feed

tail (empty) → 0.1.0.0

raw patch · 6 files changed

+513/−0 lines, 6 filesdep +asyncdep +basedep +directory

Dependencies added: async, base, directory, filepath, hspec, hspec-discover, process, relude, stm, stm-chans, tail, temporary, text, which

Files

+ src/Control/Concurrent/STM/CircularBuffer.hs view
@@ -0,0 +1,106 @@+{- |+Circular buffer implementation using STM TBMQueue with drop-oldest semantics.++This module provides a circular buffer built on top of "Control.Concurrent.STM.TBMQueue"+that automatically drops the oldest items when the buffer becomes full.++Example usage:++@+import Control.Concurrent.STM.CircularBuffer qualified as CB++main :: IO ()+main = atomically $ do+  buf <- CB.'new' 3+  CB.'add' "first" buf+  CB.'add' "second" buf+  CB.'add' "third" buf+  CB.'add' "fourth" buf  -- Drops "first"+  items <- CB.'drain' buf+  -- items will be Just ("second" :| ["third", "fourth"])+@+-}+module Control.Concurrent.STM.CircularBuffer (+  CircularBuffer,+  new,+  add,+  clone,+  drain,+  close,+) where++import Control.Concurrent.STM.TBMQueue (TBMQueue, closeTBMQueue, isFullTBMQueue, newTBMQueue, readTBMQueue, tryReadTBMQueue, writeTBMQueue)++-- | Circular buffer using TBMQueue - when full, oldest items are dropped+data CircularBuffer a = CircularBuffer (TBMQueue a) Int++{- | Create a new circular buffer with the given capacity.++The buffer will hold at most @capacity@ items. When full, adding new items+will cause the oldest items to be dropped.+-}+new :: Int -> STM (CircularBuffer a)+new cap = CircularBuffer <$> newTBMQueue cap <*> pure cap++{- | Add an element to the circular buffer.++If the buffer is full, the oldest item will be dropped to make room+for the new item.+-}+add :: a -> CircularBuffer a -> STM ()+add item (CircularBuffer queue _) = do+  whenM (isFullTBMQueue queue) $ do+    void $ readTBMQueue queue -- Drop oldest item+  writeTBMQueue queue item++{- | Clone the contents of a circular buffer into a new buffer with the same capacity.++This operation drains all items from the source buffer, copies them to a new buffer,+and then puts them back into the source buffer. The items maintain their order.++Throws an error if the source buffer is closed.+-}+clone :: (HasCallStack) => CircularBuffer a -> STM (CircularBuffer a)+clone sourceBuffer@(CircularBuffer _ capacity) = do+  newBuffer <- new capacity+  -- Use drain to get all items without blocking if empty+  drain sourceBuffer >>= \case+    Nothing -> error "Cannot clone a closed CircularBuffer"+    Just (toList -> items) -> do+      -- Add all items to new buffer and put them back in source+      mapM_ (`add` newBuffer) items+      -- Put items back in source buffer (FIFO order)+      mapM_ (`add` sourceBuffer) items+      pure newBuffer++{- | Read all currently available items, blocking for the first item.++This function will block until at least one item is available, then drain+all remaining available items without blocking.++Returns 'Nothing' if the queue is closed, or @'Just' items@ if at least+one item is available.+-}+drain :: CircularBuffer a -> STM (Maybe (NonEmpty a))+drain (CircularBuffer queue _) = do+  -- First, block for one item (same behavior as readTBMQueue)+  readTBMQueue queue >>= \case+    Nothing -> pure Nothing -- Queue is closed+    Just x -> do+      -- Got first item, now collect all other available items without blocking+      let go acc = do+            item <- tryReadTBMQueue queue+            case item of+              Nothing -> pure (reverse acc) -- Queue is closed+              Just Nothing -> pure (reverse acc) -- No more items available+              Just (Just y) -> go (y : acc) -- Got another item+      rest <- go []+      pure $ Just (x :| rest)++{- | Close the buffer to signal end-of-stream.++Once closed, no more items can be added to the buffer, and 'drain'+will return 'Nothing' after all remaining items have been drained.+-}+close :: CircularBuffer a -> STM ()+close (CircularBuffer queue _) = closeTBMQueue queue
+ src/System/Tail.hs view
@@ -0,0 +1,141 @@+{-# LANGUAGE OverloadedRecordDot #-}+{-# LANGUAGE RecordWildCards #-}++{- |+File tailing library with multi-subscriber support.++This library provides a Haskell API for @tail -f@ style file streaming+using 'STM', 'Control.Concurrent.Async', and system processes.++Example usage:++@+import System.Tail++main :: IO ()+main = do+  tail <- 'tailFile' \"\/var\/log\/app.log\"+  subscriber <- 'tailSubscribe' tail+  -- Read from subscriber...+  'tailStop' tail+@+-}+module System.Tail (+  -- * Core Types+  Tail,++  -- * Operations+  tailFile,+  tailStop,+  tailSubscribe,+) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async (Async, async)+import Control.Concurrent.STM.CircularBuffer (CircularBuffer)+import Control.Concurrent.STM.CircularBuffer qualified as CB+import System.Directory (doesFileExist)+import System.IO (hGetLine)+import System.Process (CreateProcess (..), ProcessHandle, StdStream (..), createProcess, proc, terminateProcess, waitForProcess)++-- | Represent `tail -f`'ing a file in Haskell+data Tail = Tail+  { filePath :: FilePath+  -- ^ The file being tailed+  , stop :: TMVar ()+  -- ^ Signal to stop tailing+  , tailProcess :: TVar (Maybe (ProcessHandle, Async ()))+  -- ^ The tail process handle and async reader+  , queues :: TVar [CircularBuffer Text]+  -- ^ Active subscriber queues+  , ringBuffer :: CircularBuffer Text+  -- ^ Ring buffer storing last N lines for new subscribers+  }+  deriving stock (Generic)++{- | Create a new 'Tail' handle for the given file path with specified buffer size.++The tail process starts immediately and begins reading from the file.+New subscribers will receive a ring buffer containing the last @bufferSize@ lines.+-}+tailFile :: (HasCallStack) => Int -> FilePath -> IO Tail+tailFile bufferSize filePath = do+  unlessM (doesFileExist filePath) $ error $ "File does not exist: " <> toText filePath+  queues <- newTVarIO mempty+  stop <- newEmptyTMVarIO+  tailProcess <- newTVarIO Nothing+  ringBuffer <- atomically $ CB.new bufferSize+  let t = Tail {..}+  -- Start the tail process immediately+  void $ async $ tailRun t+  pure t++{- | Signal the tail process to stop reading the file.++This will terminate the underlying @tail@ process and close all subscriber queues.+-}+tailStop :: Tail -> IO ()+tailStop t = do+  atomically $ putTMVar t.stop ()++tailRun :: Tail -> IO ()+tailRun t = do+  -- Start the tail -F process (show entire file from beginning)+  let createProc = (proc "tail" ["-F", "-n", "+1", t.filePath]) {std_out = CreatePipe}+  (_, Just hout, _, ph) <- createProcess createProc++  -- Start async reader that reads from tail process and distributes to queues+  readerAsync <- async $ readAndDistribute hout++  -- Store the process and reader+  atomically $ writeTVar t.tailProcess (Just (ph, readerAsync))++  -- Wait for stop signal+  atomically $ takeTMVar t.stop++  -- Clean up: terminate process and wait for reader to finish at EOF+  threadDelay 1_000_000 -- Give `tail -f` a second to flush any remaining lines+  terminateProcess ph+  void $ waitForProcess ph++  -- Reader will naturally stop when it reaches EOF++  -- Close all queues so readers can detect end of stream+  atomically $ do+    qs <- readTVar t.queues+    mapM_ CB.close qs++  atomically $ writeTVar t.tailProcess Nothing+  where+    readAndDistribute :: Handle -> IO ()+    readAndDistribute h = do+      let readLines = do+            hIsEOF h >>= \case+              True -> pass -- EOF reached, stop reading+              False -> do+                line <- toText <$> hGetLine h+                -- Add to ring buffer and distribute to all queues+                atomically $ do+                  -- Update ring buffer+                  CB.add line t.ringBuffer+                  -- Distribute to subscriber queues+                  qs <- readTVar t.queues+                  forM_ qs $ \q ->+                    CB.add line q+                readLines+      readLines++{- | Subscribe to tail output and receive a 'CircularBuffer' for reading lines.++The returned buffer will contain any previously read lines from the ring buffer,+plus all new lines as they are read from the file.++Use 'Control.Concurrent.STM.CircularBuffer.drain' to read from the buffer.+-}+tailSubscribe :: Tail -> IO (CircularBuffer Text)+tailSubscribe t = atomically $ do+  -- Clone ring buffer as CircularBuffer with buffered lines+  queue <- CB.clone t.ringBuffer+  -- Add to active subscribers+  modifyTVar' t.queues (queue :)+  pure queue
+ tail.cabal view
@@ -0,0 +1,130 @@+cabal-version: 2.0++-- This file has been generated from package.yaml by hpack version 0.37.0.+--+-- see: https://github.com/sol/hpack++name:           tail+version:        0.1.0.0+synopsis:       Haskell API for tail -f streaming+description:    A Haskell library providing an API for tail -f style file streaming using STM, process, and async.+category:       System+homepage:       https://github.com/juspay/vira+author:         Sridhar Ratnakumar+maintainer:     srid@srid.ca+copyright:      2025 Sridhar Ratnakumar+license:        MIT+build-type:     Simple++library+  exposed-modules:+      Control.Concurrent.STM.CircularBuffer+      System.Tail+  other-modules:+      Paths_tail+  autogen-modules:+      Paths_tail+  hs-source-dirs:+      src+  default-extensions:+      DataKinds+      DeriveDataTypeable+      DeriveGeneric+      DeriveTraversable+      DerivingStrategies+      DerivingVia+      ExplicitForAll+      FlexibleContexts+      FlexibleInstances+      GeneralizedNewtypeDeriving+      ImportQualifiedPost+      LambdaCase+      MultiParamTypeClasses+      MultiWayIf+      NamedFieldPuns+      NoStarIsType+      NumericUnderscores+      OverloadedStrings+      ScopedTypeVariables+      StrictData+      TypeApplications+      TypeFamilies+      TypeOperators+      TypeSynonymInstances+      ViewPatterns+  ghc-options: -Wall -optP-Wno-nonportable-include-path -Wincomplete-record-updates -Wincomplete-uni-patterns -Wmissing-deriving-strategies -Wunused-foralls -fprint-explicit-foralls -fprint-explicit-kinds+  build-depends:+      async+    , base ==4.*+    , directory+    , filepath+    , process+    , relude >=1.0+    , stm+    , stm-chans+    , text+    , which+  mixins:+      base hiding (Prelude)+    , relude (Relude as Prelude, Relude.Container.One)+    , relude +  default-language: GHC2021++test-suite tail-test+  type: exitcode-stdio-1.0+  main-is: Spec.hs+  other-modules:+      Control.Concurrent.STM.CircularBufferSpec+      System.TailSpec+      Paths_tail+  autogen-modules:+      Paths_tail+  hs-source-dirs:+      test+  default-extensions:+      DataKinds+      DeriveDataTypeable+      DeriveGeneric+      DeriveTraversable+      DerivingStrategies+      DerivingVia+      ExplicitForAll+      FlexibleContexts+      FlexibleInstances+      GeneralizedNewtypeDeriving+      ImportQualifiedPost+      LambdaCase+      MultiParamTypeClasses+      MultiWayIf+      NamedFieldPuns+      NoStarIsType+      NumericUnderscores+      OverloadedStrings+      ScopedTypeVariables+      StrictData+      TypeApplications+      TypeFamilies+      TypeOperators+      TypeSynonymInstances+      ViewPatterns+  ghc-options: -Wall -optP-Wno-nonportable-include-path -Wincomplete-record-updates -Wincomplete-uni-patterns -Wmissing-deriving-strategies -Wunused-foralls -fprint-explicit-foralls -fprint-explicit-kinds+  build-depends:+      async+    , base ==4.*+    , directory+    , filepath+    , hspec+    , hspec-discover+    , process+    , relude >=1.0+    , stm+    , stm-chans+    , tail+    , temporary+    , text+    , which+  mixins:+      base hiding (Prelude)+    , relude (Relude as Prelude, Relude.Container.One)+    , relude +  default-language: GHC2021
+ test/Control/Concurrent/STM/CircularBufferSpec.hs view
@@ -0,0 +1,56 @@+module Control.Concurrent.STM.CircularBufferSpec where++import Control.Concurrent.STM.CircularBuffer qualified as CB+import Test.Hspec++spec :: Spec+spec = describe "CircularBuffer" $ do+  it "adds and drains items" $ do+    items <- atomically $ do+      buf <- CB.new 3+      CB.add ("a" :: Text) buf+      CB.add "b" buf+      CB.drain buf+    fmap toList items `shouldBe` Just ["a", "b"]++  it "drops oldest when full" $ do+    items <- atomically $ do+      buf <- CB.new 2+      CB.add ("a" :: Text) buf+      CB.add "b" buf+      CB.add "c" buf -- Should drop "a"+      CB.drain buf+    fmap toList items `shouldBe` Just ["b", "c"]++  it "clones buffer contents" $ do+    clonedItems <- atomically $ do+      buf <- CB.new 2+      CB.add ("x" :: Text) buf+      CB.add "y" buf+      cloned <- CB.clone buf+      CB.drain cloned+    fmap toList clonedItems `shouldBe` Just ["x", "y"]++  it "drains remaining items when closed" $ do+    result <- atomically $ do+      buf <- CB.new 2+      CB.add ("test" :: Text) buf+      CB.close buf+      CB.drain buf+    fmap toList result `shouldBe` Just ["test"]++  it "returns Nothing when closed and empty" $ do+    result <- atomically $ do+      buf <- CB.new 2+      CB.close buf+      CB.drain buf+    result `shouldBe` (Nothing :: Maybe (NonEmpty Text))++  it "throws error when cloning closed buffer" $ do+    atomically+      ( do+          buf <- CB.new 2+          CB.close buf+          CB.clone buf+      )+      `shouldThrow` anyErrorCall
+ test/Spec.hs view
@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}
+ test/System/TailSpec.hs view
@@ -0,0 +1,79 @@+module System.TailSpec where++import Control.Concurrent (forkIO, threadDelay)+import Control.Concurrent.STM.CircularBuffer (CircularBuffer)+import Control.Concurrent.STM.CircularBuffer qualified as CB+import GHC.IO.Handle (hClose)+import System.Directory (doesFileExist)+import System.IO.Temp (withSystemTempFile)+import System.Process (system)+import System.Tail qualified as Tail+import Test.Hspec++-- Helper to read all items from CircularBuffer until closed+drainAll :: CircularBuffer a -> IO [a]+drainAll q = do+  let go acc = do+        maybeItems <- atomically $ CB.drain q+        case maybeItems of+          Nothing -> pure acc -- Queue is closed+          Just items -> go (acc ++ toList items)+  go []++spec :: Spec+spec = describe "System.Tail" $ do+  it "hello" $ do+    t <- Tail.tailFile 100 "/dev/null"+    Tail.tailStop t+  it "streams a static file" $ do+    thisFile <- findThisFile+    t <- Tail.tailFile 100 thisFile+    q <- Tail.tailSubscribe t+    threadDelay 1_000_000 -- Wait 1 second for tail to start and read content+    Tail.tailStop t+    ls <- drainAll q+    viaNonEmpty head ls `shouldBe` Just "module System.TailSpec where"+    -- Find the last line from the lines we got+    let lastLine = viaNonEmpty last ls+    lastLine `shouldBe` Just "-- End of file."+  it "streams a log file being appended to by another process" $ do+    -- Create a file under a temp directory. Then spawn an external process that writes to it lines over time.+    withSystemTempFile "tail-spec" $ \tempFile h -> do+      hClose h+      void $ system $ "echo 'Hello' >> " <> tempFile+      t <- Tail.tailFile 100 tempFile+      q <- Tail.tailSubscribe t+      void $ forkIO $ do+        threadDelay 1_000_000+        void $ system $ "echo 'World' >> " <> tempFile+        Tail.tailStop t+      -- Read everything and compare output+      threadDelay 3_000_000+      ls <- drainAll q+      ls `shouldBe` ["Hello", "World"]+  it "ring buffer provides last lines to new subscribers" $ do+    withSystemTempFile "ring-buffer-spec" $ \tempFile h -> do+      hClose h+      void $ system $ "echo 'Line1' >> " <> tempFile+      t <- Tail.tailFile 100 tempFile+      threadDelay 500_000 -- Let tail read existing lines+      q <- Tail.tailSubscribe t+      void $ system $ "echo 'Line2' >> " <> tempFile -- Write *after* subscribing+      Tail.tailStop t+      ls <- drainAll q+      ls `shouldBe` ["Line1", "Line2"]++-- | Find this test file by trying common paths+findThisFile :: IO FilePath+findThisFile = do+  let paths =+        [ "test/System/TailSpec.hs" -- cabal test / nix build+        , "packages/tail/test/System/TailSpec.hs" -- ghcid from root+        ]+  let findFirst [] = error "TailSpec.hs not found"+      findFirst (p : ps) = do+        exists <- doesFileExist p+        if exists then pure p else findFirst ps+  findFirst paths++-- End of file.