diff --git a/ChangeLog.md b/ChangeLog.md new file mode 100644 --- /dev/null +++ b/ChangeLog.md @@ -0,0 +1,5 @@ +# Revision history for fold-debounce-conduit + +## 0.1.0.0 -- 2015-06-22 + +* First version. Released on an unsuspecting world. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -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 diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/fold-debounce-conduit.cabal b/fold-debounce-conduit.cabal new file mode 100644 --- /dev/null +++ b/fold-debounce-conduit.cabal @@ -0,0 +1,53 @@ +name: fold-debounce-conduit +version: 0.1.0.0 +author: Toshio Ito +maintainer: Toshio Ito +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 diff --git a/src/Data/Conduit/FoldDebounce.hs b/src/Data/Conduit/FoldDebounce.hs new file mode 100644 --- /dev/null +++ b/src/Data/Conduit/FoldDebounce.hs @@ -0,0 +1,159 @@ +-- | +-- Module: Data.Conduit.FoldDebounce +-- Description: Regulate input traffic from conduit Source with Control.FoldDebounce +-- Maintainer: Toshio Ito +-- +-- 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 diff --git a/test/Data/Conduit/FoldDebounceSpec.hs b/test/Data/Conduit/FoldDebounceSpec.hs new file mode 100644 --- /dev/null +++ b/test/Data/Conduit/FoldDebounceSpec.hs @@ -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 + + diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -0,0 +1,1 @@ +{-# OPTIONS_GHC -F -pgmF hspec-discover #-}