fold-debounce-conduit (empty) → 0.1.0.0
raw patch · 8 files changed
+380/−0 lines, 8 filesdep +basedep +conduitdep +fold-debouncesetup-changed
Dependencies added: base, conduit, fold-debounce, fold-debounce-conduit, hspec, resourcet, stm, transformers, transformers-base
Files
- ChangeLog.md +5/−0
- LICENSE +30/−0
- README.md +13/−0
- Setup.hs +2/−0
- fold-debounce-conduit.cabal +53/−0
- src/Data/Conduit/FoldDebounce.hs +159/−0
- test/Data/Conduit/FoldDebounceSpec.hs +117/−0
- test/Spec.hs +1/−0
+ ChangeLog.md view
@@ -0,0 +1,5 @@+# Revision history for fold-debounce-conduit++## 0.1.0.0 -- 2015-06-22++* First version. Released on an unsuspecting world.
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2015, Toshio Ito++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 Toshio Ito 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.
+ README.md view
@@ -0,0 +1,13 @@+# fold-debounce-conduit++Regulate input traffic from conduit Source with Control.FoldDebounce++## How to run tests++```+cabal configure --enable-tests && cabal build && cabal test+```++## Author++Toshio Ito <debug.ito [at] gmail.com>
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ fold-debounce-conduit.cabal view
@@ -0,0 +1,53 @@+name: fold-debounce-conduit+version: 0.1.0.0+author: Toshio Ito <debug.ito@gmail.com>+maintainer: Toshio Ito <debug.ito@gmail.com>+license: BSD3+license-file: LICENSE+synopsis: Regulate input traffic from conduit Source with Control.FoldDebounce+description: Regulate input traffic from conduit Source with Control.FoldDebounce. See "Data.Conduit.FoldDebounce"+category: Conduit+cabal-version: >= 1.10+build-type: Simple+extra-source-files: README.md, ChangeLog.md+homepage: https://github.com/debug-ito/fold-debounce-conduit+bug-reports: https://github.com/debug-ito/fold-debounce-conduit/issues++library+ default-language: Haskell2010+ hs-source-dirs: src+ ghc-options: -Wall+ exposed-modules: Data.Conduit.FoldDebounce+ -- other-modules:+ default-extensions: FlexibleContexts+ build-depends: base >=4.6.0 && <4.9,+ conduit >=1.2.4 && <1.3,+ fold-debounce >=0.2.0 && <0.3,+ resourcet >=1.1.5 && <1.2,+ stm >=2.4.4 && <2.5,+ transformers >=0.3.0 && <0.5,+ transformers-base >=0.4.4 && <0.5++-- executable fold-debounce-conduit+-- default-language: Haskell2010+-- hs-source-dirs: src+-- main-is: Main.hs+-- ghc-options: -Wall+-- -- other-modules: +-- -- other-extensions: +-- build-depends: base >=4 && <5++test-suite spec+ type: exitcode-stdio-1.0+ default-language: Haskell2010+ hs-source-dirs: test+ ghc-options: -Wall+ main-is: Spec.hs+ other-modules: Data.Conduit.FoldDebounceSpec+ build-depends: base, fold-debounce-conduit, stm,+ conduit, transformers, resourcet,+ hspec >=2.1.7 && <2.2++source-repository head+ type: git+ location: https://github.com/debug-ito/fold-debounce-conduit.git
+ src/Data/Conduit/FoldDebounce.hs view
@@ -0,0 +1,159 @@+-- |+-- Module: Data.Conduit.FoldDebounce+-- Description: Regulate input traffic from conduit Source with Control.FoldDebounce+-- Maintainer: Toshio Ito <debug.ito@gmail.com>+-- +-- Synopsis:+--+-- > module Main (main) where+-- > +-- > import Data.Conduit (Source, Sink, yield, ($$))+-- > import qualified Data.Conduit.List as CL+-- > import Control.Concurrent (threadDelay)+-- > import Control.Monad.IO.Class (liftIO)+-- > import Control.Monad.Trans.Resource (ResourceT, runResourceT)+-- > +-- > import qualified Data.Conduit.FoldDebounce as F+-- > +-- > fastSource :: Int -> Source (ResourceT IO) Int+-- > fastSource max_num = fastStream' 0 where+-- > fastStream' count = do+-- > yield count+-- > if count >= max_num+-- > then return ()+-- > else do+-- > liftIO $ threadDelay 100000+-- > fastStream' (count + 1)+-- > +-- > printSink :: Show a => Sink a (ResourceT IO) ()+-- > printSink = CL.mapM_ (liftIO . putStrLn . show)+-- > +-- > main :: IO ()+-- > main = do+-- > putStrLn "-- Before debounce"+-- > runResourceT $ fastSource 10 $$ printSink+-- > let debouncer = F.debounce F.Args { F.cb = undefined, -- anything will do+-- > F.fold = (\list num -> list ++ [num]),+-- > F.init = [] }+-- > F.def { F.delay = 500000 }+-- > putStrLn "-- After debounce"+-- > runResourceT $ debouncer (fastSource 10) $$ printSink+--+-- Result:+--+-- > -- Before debounce+-- > 0+-- > 1+-- > 2+-- > 3+-- > 4+-- > 5+-- > 6+-- > 7+-- > 8+-- > 9+-- > 10+-- > -- After debounce+-- > [0,1,2,3,4]+-- > [5,6,7,8,9]+-- > [10]+--+-- This module regulates (slows down) data stream from conduit+-- 'Source' using "Control.FoldDebounce".+--+-- The data from the original 'Source' (type @i@) are pulled and+-- folded together to create an output data (type @o@). The output+-- data then comes out of the debounced 'Source' in a predefined+-- interval (specified by 'delay' option).+--+-- See "Control.FoldDebounce" for detail.+module Data.Conduit.FoldDebounce (+ debounce,+ -- * Re-exports+ Args(..),+ Opts,+ def,+ -- ** Accessors for 'Opts'+ delay,+ alwaysResetTimer,+ -- * Preset parameters+ forStack, forMonoid, forVoid+) where++import Prelude hiding (init)+import Control.Monad (void)+import Data.Monoid (Monoid)++import Control.FoldDebounce (Args(Args,cb,fold,init),+ Opts, delay, alwaysResetTimer, def)+import qualified Control.FoldDebounce as F+import Data.Conduit (Source, Sink, await, yieldOr, ($$))+import Control.Monad.Trans.Resource (MonadResource, MonadBaseControl,+ allocate, register, release, resourceForkIO, runResourceT)+import Control.Monad.Trans.Class (lift)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Concurrent.STM (newTChanIO, writeTChan, readTChan,+ atomically,+ TVar, readTVar, newTVarIO, writeTVar)++-- | Debounce conduit 'Source' with "Control.FoldDebounce". The data+-- stream from the original 'Source' (type @i@) is debounced and+-- folded into the data stream of the type @o@.+--+-- Note that the original 'Source' is connected to a 'Sink' in another+-- thread. You may need some synchronization if the original 'Source'+-- has side-effects.+debounce :: (MonadResource m, MonadBaseControl IO m)+ => Args i o -- ^ mandatory argument for FoldDebounce. 'cb'+ -- field is ignored, so you can set anything+ -- to that.+ -> Opts i o -- ^ optional argument for FoldDebounce+ -> Source m i -- ^ original 'Source'+ -> Source m o -- ^ debounced 'Source'+debounce args opts src = do+ out_chan <- liftIO $ newTChanIO+ (out_termed_key, out_termed) <- allocate (liftIO $ newTVarIO False) (liftIO . atomically . flip writeTVar True)+ let retSource = do+ mgot <- liftIO $ atomically $ readTChan out_chan+ case mgot of+ OutFinished -> return ()+ OutData got -> yieldOr got (release out_termed_key) >> retSource+ lift $ runResourceT $ do+ void $ register $ atomically $ writeTChan out_chan OutFinished+ (_, trig) <- allocate (F.new args { F.cb = atomically . writeTChan out_chan . OutData }+ opts)+ (F.close)+ void $ resourceForkIO $ lift (src $$ trigSink trig out_termed)+ retSource++-- | Internal data type for output channel.+data OutData o = OutData o+ | OutFinished++trigSink :: (MonadIO m) => F.Trigger i o -> TVar Bool -> Sink i m ()+trigSink trig out_termed = trigSink' where+ trigSink' = do+ mgot <- await+ termed <- liftIO $ atomically $ readTVar out_termed+ case (termed, mgot) of+ (True, _) -> return ()+ (False, Nothing) -> return ()+ (False, Just got) -> do+ liftIO $ F.send trig got+ trigSink'+++-- | 'Args' for stacks. Input events are accumulated in a stack, i.e.,+-- the last event is at the head of the list.+forStack :: Args i [i]+forStack = F.forStack undefined++-- | 'Args' for monoids. Input events are appended to the tail.+forMonoid :: Monoid i => Args i i+forMonoid = F.forMonoid undefined++-- | 'Args' that discards input events. The data stream from the+-- debounced 'Source' indicates the presence of data from the original+-- 'Source'.+forVoid :: Args i ()+forVoid = F.forVoid undefined
+ test/Data/Conduit/FoldDebounceSpec.hs view
@@ -0,0 +1,117 @@+module Data.Conduit.FoldDebounceSpec (main, spec) where++import Test.Hspec++import Data.Maybe (isJust)+import Data.Monoid (Monoid)+import Control.Concurrent (threadDelay, myThreadId)+import Control.Monad (forM_, void)+import System.Timeout (timeout)+import qualified Data.Conduit.FoldDebounce as F+import Data.Conduit (Source, ConduitM, ($$), yield, addCleanup)+import Control.Monad.IO.Class (liftIO)+import qualified Data.Conduit.List as CL+import Control.Concurrent.STM (atomically, TVar, newTVarIO, writeTVar, readTVar)+import Control.Monad.Trans.Resource (ResourceT, runResourceT, register)++main :: IO ()+main = hspec spec++delayedSource :: [(Int, a)] -> Source (ResourceT IO) a+delayedSource [] = return ()+delayedSource ((delay, item):rest) = do+ liftIO $ threadDelay delay+ yield item+ delayedSource rest++periodicSource :: Int -> [a] -> Source (ResourceT IO) a+periodicSource interval items = delayedSource $ zip (repeat interval) items++terminationDetector :: IO (TVar Bool, (ConduitM i o (ResourceT IO) r -> ConduitM i o (ResourceT IO) r))+terminationDetector = do+ terminated <- newTVarIO False+ return (terminated,+ addCleanup (\completed -> if not completed then liftIO $ atomically $ writeTVar terminated True else return () ) )++attachResource :: Source (ResourceT IO) a -> IO (TVar Bool, Source (ResourceT IO) a)+attachResource src = do+ released <- newTVarIO False+ let src' = do+ void $ register $ atomically $ writeTVar released True+ src+ return (released, src')++debSum :: Int -> Source (ResourceT IO) Int -> Source (ResourceT IO) Int+debSum delay = F.debounce F.Args { F.init = 0, F.fold = (+), F.cb = undefined } F.def { F.delay = delay }++debMonoid :: Monoid i => Int -> Source (ResourceT IO) i -> Source (ResourceT IO) i+debMonoid delay = F.debounce F.forMonoid F.def { F.delay = delay }++spec :: Spec+spec = do+ describe "debounce" $ do+ it "should fold inputs" $ do+ ret <- runResourceT $ debSum 500000 (periodicSource 10000 [1..10]) $$ CL.consume+ ret `shouldBe` [sum [1..10]]+ it "should debounce source" $ do+ let s = delayedSource [(1000, "a"), (1000, "b"), (200000, "c"), (1000, "d"), (1000, "e"), (200000, "f")]+ ret <- runResourceT $ debMonoid 100000 s $$ CL.consume+ ret `shouldBe` ["ab", "cde", "f"]+ it "should terminate debounced Source immediately if the original Source terminates immediately" $ do+ ret <- timeout 50000000 $ runResourceT $ debMonoid 60000000 (CL.sourceList ["A", "B", "C", "D", "E"]) $$ CL.consume+ ret `shouldBe` Just ["ABCDE"]+ it "should terminate the Sink for the original Source if the Sink for the debounced Source terminates" $ do+ (terminated, detector) <- terminationDetector+ let orig_source = detector $ periodicSource 10000 (repeat "a")+ ret <- runResourceT $ debMonoid 50000 orig_source $$ CL.take 4+ length ret `shouldBe` 4+ forM_ ret (`shouldContain` "aaa")+ threadDelay 20000+ atomically (readTVar terminated) `shouldReturn` True+ it "should terminate the debounced Source gracefully if the original Source throws exception" $ do+ let s = (periodicSource 1000 ["a", "b"]) >> error "Exception in origSource" >> (periodicSource 1000 ["c", "d"])+ ret <- runResourceT $ debMonoid 100000 s $$ CL.consume+ ret `shouldBe` ["ab"]+ it "should terminated the original Source gracefully if the downstream Sink throws exception" $ do+ (terminated, detector) <- terminationDetector+ let orig_source = detector $ periodicSource 10000 (repeat "a")+ ret_sink = do+ taken <- CL.take 4+ _ <- error "Exception in retSink"+ return taken+ runResourceT (debMonoid 50000 orig_source $$ ret_sink) `shouldThrow` errorCall "Exception in retSink"+ threadDelay 20000+ atomically (readTVar terminated) `shouldReturn` True+ it "should connect the original Source in another thread" $ do+ this_thread <- myThreadId+ orig_thread_t <- newTVarIO Nothing+ let orig_source = do+ liftIO $ atomically . writeTVar orig_thread_t . Just =<< myThreadId+ yield 10+ return ()+ ret <- runResourceT $ debSum 10000 orig_source $$ CL.consume+ ret `shouldBe` [10]+ orig_thread <- atomically (readTVar orig_thread_t)+ orig_thread `shouldSatisfy` isJust+ orig_thread `shouldSatisfy` (/= Just this_thread)+ it "should release the resource in the original Source when the original Source finishes" $ do+ (released, orig_source) <- attachResource $ periodicSource 10000 [1,2,3,4]+ ret <- runResourceT $ debSum 500000 orig_source $$ CL.consume+ ret `shouldBe` [10]+ atomically (readTVar released) `shouldReturn` True+ it "should release the resource in the original Source when the downstream Sink throws exception" $ do+ (released, orig_source) <- attachResource $ periodicSource 10000 $ repeat "a"+ let sink = do+ taken <- CL.take 4+ _ <- error "Exception in downstream"+ return taken+ runResourceT (debMonoid 500000 orig_source $$ sink) `shouldThrow` errorCall "Exception in downstream"+ threadDelay 20000+ atomically (readTVar released) `shouldReturn` True+ it "should release the resource in the original Source when the original Source throws exception" $ do+ (released, orig_source) <- attachResource (periodicSource 10000 ["a", "b"] >> error "Exception in source")+ ret <- runResourceT $ debMonoid 500000 orig_source $$ CL.consume+ ret `shouldBe` ["ab"]+ atomically (readTVar released) `shouldReturn` True++
+ test/Spec.hs view
@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}