stm-firehose 0.1.2 → 0.1.3

raw patch · 3 files changed

+34/−6 lines, 3 files

Files

• Control/Concurrent/STM/Firehose.hs +13/−1
• Data/Conduit/Network/Firehose.hs +20/−4
• stm-firehose.cabal +1/−1

@@ -1,3 +1,14 @@+{-|+A subscription based messaging system, with non-blocking bounded write.++> fh <- atomically newFirehose+> -- the following line will not block, even though nobody subscribed+> atomically (mapM_ (writeEvent fh) [1..100])+> -- let's subscribe a single client+> sub <- atomically (subscribe 10 fh)+> forkIO (forever (atomically (readEvent fh) >>= print))+> writeEvent fh 1+-} module Control.Concurrent.STM.Firehose (Firehose, Subscription, newFirehose, writeEvent, subscribe, unsubscribe, readEvent, getQueue) where  import Control.Concurrent.STM@@ -27,7 +38,8 @@  -- | Unsubscribe from the fire hose. Subsequent calls to 'readEvent' will -- return 'Nothing'. This runs in O(n), where n is the current number of--- subscriptions.+-- subscriptions. Please contact the maintainer if you need better+-- performance. unsubscribe :: Subscription a -> STM () unsubscribe (Subscription q (Firehose lst)) = do     closeTBMQueue q

@@ -1,4 +1,11 @@ {-# LANGUAGE OverloadedStrings #-}+{-| This module is here to let you easily build firehose systems. The+'firehoseApp' application is a standard 'Application' that will stream the events to clients. The 'firehoseConduit' function will spawn a web server on the given port, and let+the data-flow in a conduit be examined this way.++For an example implementation, with a JSON encodable data type, see <http://hackage.haskell.org/package/hslogstash/docs/src/Data-Conduit-FireHose.html#fireHose>.++-} module Data.Conduit.Network.Firehose (firehoseApp, firehoseConduit) where  import Control.Concurrent (forkIO)@@ -16,9 +23,16 @@ import Control.Monad (void) import Control.Monad.IO.Class --- | A firehose application, suitable for use in a wai-compatible server.+{-| A firehose application, suitable for use in a wai-compatible server.+A typical usage is with JSON encodable data, where the serialization function can be :++> -- encode to JSON, turn into a Builder, then append a newline.+> (<> fromByteString "\n") . fromLazyByteString . encode++The filtering function has a type that let you create it based on the 'Request'. That means you can use the query string to build the proper filters.+-} firehoseApp :: Int -- ^ Buffer size for the fire hose threads-            -> (Request -> a ->  Bool) -- ^ A filtering function for fire hose messages. Only messages that match this functions will be passed. The request can be used to build the filter.+            -> (Request -> a ->  Bool) -- ^ A filtering function for fire hose messages. Only messages that match this function will be passed. The request can be used to build the filter.             -> (a -> Builder) -- ^ The serialization function             -> Firehose a             -> Application@@ -28,8 +42,10 @@         -- Subscription a -> IO (Status, ResponseHeaders, Source IO (Flush Builder))         runFirehose sub = return (status200, [], sourceTBMQueue (getQueue sub) $= CL.filter filtering' $= CL.concatMap (\e -> [Chunk (serialize e), Flush]) ) --- | A fire hose conduit creator, that can be inserted in your conduits as--- firehose entry points. Will run Warp on the specified port.+{-| A fire hose conduit creator, that can be inserted in your conduits as+firehose entry points. Will run Warp on the specified port. Please not+that the connection will timeout after an hour.+-} firehoseConduit :: (Monad m, MonadIO m)                 => Int -- ^ Port to listen on                 -> Int -- ^ Buffer size for the fire hose threads

@@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/  name:                stm-firehose-version:             0.1.2+version:             0.1.3 synopsis:            Conduits and STM operations for fire hoses. description:         A fire hose is a component in a message passing system that let clients tap into the message flow. This module provides low level (built on STM channels) and high level (based on conduits) building blocks. It should work with a fixed amount of memory, and has non blocking write operations. license:             BSD3