write-buffer-core (empty) → 0.1.0.0

raw patch · 5 files changed

+145/−0 lines, 5 filesdep +basedep +bytestringdep +clocksetup-changed

Dependencies added: base, bytestring, clock, dlist, exceptions, lifted-async, lifted-base, monad-control, mtl, stm, stm-chans

Files

• LICENSE +30/−0
• README.md +4/−0
• Setup.hs +2/−0
• src/WriteBuffer.hs +70/−0
• write-buffer-core.cabal +39/−0

@@ -0,0 +1,30 @@+Copyright Author name here (c) 2017++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of Author name here nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

@@ -0,0 +1,4 @@+# WriteBuffer++Do you have some costly action that's faster if you can do it in bulk?+`WriteBuffer` is the tool for you!

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

@@ -0,0 +1,70 @@+{-# LANGUAGE FlexibleContexts      #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RecordWildCards       #-}++module WriteBuffer+    ( module WriteBuffer+    , module X+    ) where++import           Control.Concurrent.Lifted+import           Control.Concurrent.STM          as X+import           Control.Concurrent.STM.TBMQueue as X+import           Control.Monad.Catch+import           Control.Monad.Reader+import           Control.Monad.Trans.Control+import           Data.ByteString                 (ByteString)+import           Data.DList+import           Data.Int+import           Data.IORef.Lifted+import           System.Clock+import           System.IO+import           System.Timeout++data WriteBufferOpts rec m+    = WriteBufferOpts+    { maxBufferSize    :: Int+    , maxTimeToWait    :: Integer+    , bufferInputQueue :: TBMQueue rec+    , onError          :: SomeException -> m ()+    , saveRecords      :: [rec] -> m ()+    }++makeBufferOpts :: MonadIO m => TBMQueue rec -> ([rec] -> m ()) -> WriteBufferOpts rec m+makeBufferOpts bufferInputQueue saveRecords = WriteBufferOpts {..}+  where+    maxBufferSize = 1000+    maxTimeToWait = 5 * 10^9+    onError = liftIO . print++runWriteBuffer :: (MonadCatch m, MonadBaseControl IO m, MonadIO m) => WriteBufferOpts rec m -> m ThreadId+runWriteBuffer WriteBufferOpts{..} = fork $ do+    timeRef <- newIORef =<< liftIO (getTime Monotonic)+    forever $ loop empty 0 timeRef+  where+    loop inputs len timeRef+        | len >= maxBufferSize = do+            saveRecords (toList inputs)+            loop empty 0 timeRef+        | otherwise = do+            earlier <- liftIO $ readIORef timeRef+            now <- liftIO $ getTime Monotonic+            let diff = diffTimeSpec now earlier+            if toNanoSecs diff >= maxTimeToWait then do+                liftIO $ writeIORef timeRef now+                saveRecords (toList inputs) `catch` onError+                loop empty 0 timeRef+            else do+                mminput <- liftIO+                    $ timeout (fromIntegral maxTimeToWait `div` 1000)+                    $ atomically+                    $ readTBMQueue bufferInputQueue+                case mminput of+                    Nothing ->  -- timeout+                        loop inputs len timeRef+                    Just minput ->+                        case minput of+                            Nothing -> -- closed queue+                                pure ()+                            Just input ->+                                loop (inputs `snoc` input) (len + 1) timeRef

@@ -0,0 +1,39 @@+-- This file has been generated from package.yaml by hpack version 0.17.0.+--+-- see: https://github.com/sol/hpack++name:           write-buffer-core+version:        0.1.0.0+synopsis:       Buffer your writes, transparently+description:    Core for the write-buffer set of libraries+category:       Web+homepage:       https://github.com/parsonsmatt/write-buffer#readme+author:         Matt Parsons+maintainer:     parsonsmatt@gmail.com+copyright:      2017 Matt Parsons+license:        BSD3+license-file:   LICENSE+build-type:     Simple+cabal-version:  >= 1.10++extra-source-files:+    README.md++library+  hs-source-dirs:+      src+  build-depends:+      base >= 4.7 && < 5+    , mtl+    , exceptions+    , stm+    , stm-chans+    , bytestring+    , clock+    , dlist+    , lifted-async+    , lifted-base+    , monad-control+  exposed-modules:+      WriteBuffer+  default-language: Haskell2010