fold-debounce-conduit 0.1.0.5 → 0.2.0.0

raw patch · 4 files changed

+94/−67 lines, 4 filesdep ~basedep ~conduitdep ~hspecPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: base, conduit, hspec, resourcet, stm, transformers

API changes (from Hackage documentation)

Files

• ChangeLog.md +10/−0
• fold-debounce-conduit.cabal +7/−7
• src/Data/Conduit/FoldDebounce.hs +46/−41
• test/Data/Conduit/FoldDebounceSpec.hs +31/−19

@@ -1,5 +1,15 @@ # Revision history for fold-debounce-conduit +## 0.2.0.0  -- 2018-02-03++* Support `conduit-1.3.0` and `resourcet-1.2.0`. This causes major+  update of this package, too.+* Drop support for old GHCs and old dependency packages, to support+  the latest `conduit` and `resourcet`.+* Drop version upper bounds for `hspec`, because it's stable in most+  part.++ ## 0.1.0.5  -- 2017-01-24  * Confirmed test with `hspec-2.4.0`

@@ -1,5 +1,5 @@ name:                   fold-debounce-conduit-version:                0.1.0.5+version:                0.2.0.0 author:                 Toshio Ito <debug.ito@gmail.com> maintainer:             Toshio Ito <debug.ito@gmail.com> license:                BSD3@@ -20,12 +20,12 @@   exposed-modules:      Data.Conduit.FoldDebounce   -- other-modules:   default-extensions:   FlexibleContexts-  build-depends:        base >=4.6.0 && <5.0,-                        conduit >=1.2.4 && <1.3,+  build-depends:        base >=4.9.0 && <4.11,+                        conduit >=1.3.0 && <1.4,                         fold-debounce >=0.2.0 && <0.3,-                        resourcet >=1.1.5 && <1.2,-                        stm >=2.4.2 && <2.5,-                        transformers >=0.3.0 && <0.6,+                        resourcet >=1.2.0 && <1.3,+                        stm >=2.4.4.1 && <2.5,+                        transformers >=0.5.2 && <0.6,                         transformers-base >=0.4.4 && <0.5  -- executable fold-debounce-conduit@@ -46,7 +46,7 @@   other-modules:        Data.Conduit.FoldDebounceSpec   build-depends:        base, fold-debounce-conduit, stm,                         conduit, transformers, resourcet,-                        hspec >=2.1.7 && <2.5+                        hspec >=2.1.7  source-repository head   type:                 git

@@ -7,15 +7,16 @@ -- -- > module Main (main) where -- > --- > import Data.Conduit (Source, Sink, yield, ($$))+-- > import Data.Conduit (ConduitT, yield, runConduit, (.|)) -- > import qualified Data.Conduit.List as CL+-- > import Data.Void (Void) -- > 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 :: Int -> ConduitT () Int (ResourceT IO) () -- > fastSource max_num = fastStream' 0 where -- >   fastStream' count = do -- >     yield count@@ -25,20 +26,20 @@ -- >         liftIO $ threadDelay 100000 -- >         fastStream' (count + 1) -- > --- > printSink :: Show a => Sink a (ResourceT IO) ()+-- > printSink :: Show a => ConduitT a Void (ResourceT IO) () -- > printSink = CL.mapM_ (liftIO . putStrLn . show) -- >  -- > main :: IO () -- > main = do -- >   putStrLn "-- Before debounce"--- >   runResourceT $ fastSource 10 $$ printSink+-- >   runResourceT $ runConduit $ 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---+-- >   runResourceT $ runConduit $ debouncer (fastSource 10) .| printSink+--  -- Result: -- -- > -- Before debounce@@ -58,13 +59,13 @@ -- > [5,6,7,8,9] -- > [10] ----- This module regulates (slows down) data stream from conduit--- 'Source' using "Control.FoldDebounce".+-- 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).+-- 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 (@@ -83,12 +84,13 @@ import Prelude hiding (init) import Control.Monad (void) import Data.Monoid (Monoid)+import Data.Void (Void)  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,+import Data.Conduit (ConduitT, await, (.|), bracketP, yield, runConduit)+import Control.Monad.Trans.Resource (MonadResource, MonadUnliftIO,                                      allocate, register, release, resourceForkIO, runResourceT) import Control.Monad.Trans.Class (lift) import Control.Monad.IO.Class (MonadIO, liftIO)@@ -96,41 +98,44 @@                                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@.+-- | 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'+-- 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)+debounce :: (MonadResource m, MonadUnliftIO 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-+            -> ConduitT () i m () -- ^ original source+            -> ConduitT () o m () -- ^ debounced source+debounce args opts src = bracketP initOutTermed finishOutTermed debounceWith+  where+    initOutTermed = newTVarIO False+    finishOutTermed = atomically . flip writeTVar True+    debounceWith out_termed = do+      out_chan <- liftIO $ newTChanIO+      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 $ runConduit (src .| trigSink trig out_termed)+      keepYield out_chan+    keepYield out_chan = do+      mgot <- liftIO $ atomically $ readTChan out_chan+      case mgot of+       OutFinished -> return ()+       OutData got -> yield got >> keepYield out_chan+       -- | 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 :: (MonadIO m) => F.Trigger i o -> TVar Bool -> ConduitT i Void m () trigSink trig out_termed = trigSink' where   trigSink' = do     mgot <- await@@ -153,7 +158,7 @@ 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'.+-- debounced source indicates the presence of data from the original+-- source. forVoid :: Args i () forVoid = F.forVoid undefined

@@ -8,7 +8,7 @@ import Control.Monad (forM_, void) import Control.Monad.IO.Class (liftIO) import Control.Monad.Trans.Resource (ResourceT, runResourceT, register)-import Data.Conduit (Source, ConduitM, ($$), yield, addCleanup)+import Data.Conduit (ConduitT, (.|), yield, bracketP, runConduit) import qualified Data.Conduit.FoldDebounce as F import qualified Data.Conduit.List as CL import Data.Maybe (isJust)@@ -18,23 +18,33 @@ main :: IO () main = hspec spec -delayedSource :: [(Int, a)] -> Source (ResourceT IO) a+delayedSource :: [(Int, a)] -> ConduitT () a (ResourceT IO) () delayedSource [] = return () delayedSource ((delay, item):rest) = do   liftIO $ threadDelay delay   yield item   delayedSource rest -periodicSource :: Int -> [a] -> Source (ResourceT IO) a+periodicSource :: Int -> [a] -> ConduitT () a (ResourceT IO) () 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 :: IO (TVar Bool, (ConduitT i o (ResourceT IO) r -> ConduitT 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 () ) )+  return (terminated, makeDetector terminated)+  where+    makeDetector terminated orig = bracketP initCompletion setTermination runAction+      where+        initCompletion = newTVarIO False+        runAction v_completed = do+          ret <- orig+          liftIO $ atomically $ writeTVar v_completed True+          return ret+        setTermination v_completed = atomically $ do+          completed <- readTVar v_completed+          writeTVar terminated $ not completed -attachResource :: Source (ResourceT IO) a -> IO (TVar Bool, Source (ResourceT IO) a)+attachResource :: ConduitT () a (ResourceT IO) () -> IO (TVar Bool, ConduitT () a (ResourceT IO) ()) attachResource src = do   released <- newTVarIO False   let src' = do@@ -42,10 +52,10 @@         src   return (released, src') -debSum :: Int -> Source (ResourceT IO) Int -> Source (ResourceT IO) Int+debSum :: Int -> ConduitT () Int (ResourceT IO) () -> ConduitT () Int (ResourceT IO) () 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 :: Monoid i => Int -> ConduitT () i (ResourceT IO) () -> ConduitT () i (ResourceT IO) () debMonoid delay = F.debounce F.forMonoid F.def { F.delay = delay }  shouldSatisfyEventually :: Int -> TVar a -> (a -> Bool) -> IO ()@@ -64,19 +74,20 @@ spec = do   describe "debounce" $ do     it "should fold inputs" $ do-      ret <- runResourceT $ debSum 500000 (periodicSource 10000 [1..10]) $$ CL.consume+      ret <- runResourceT $ runConduit $ 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 <- runResourceT $ runConduit $ 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 <- timeout 50000000 $ runResourceT $ runConduit+             $ 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+      ret <- runResourceT $ runConduit $ debMonoid 50000 orig_source .| CL.take 4       length ret `shouldBe` 4       forM_ ret (`shouldContain` "a")       terminated `shouldSatisfyEventually'` (== True)@@ -85,7 +96,7 @@       -- i.e., we just let it terminate the thread. So we'll see the       -- error message while running the test.       let s = (periodicSource 1000 ["a", "b"]) >> error "Exception in origSource" >> (periodicSource 1000 ["c", "d"])-      ret <- runResourceT $ debMonoid 100000 s $$ CL.consume+      ret <- runResourceT $ runConduit $ 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@@ -94,7 +105,7 @@             taken <- CL.take 4             _ <- error "Exception in retSink"             return taken-      runResourceT (debMonoid 50000 orig_source $$ ret_sink) `shouldThrow` errorCall "Exception in retSink"+      (runResourceT $ runConduit $ debMonoid 50000 orig_source .| ret_sink) `shouldThrow` errorCall "Exception in retSink"       terminated `shouldSatisfyEventually'` (== True)     it "should connect the original Source in another thread" $ do       this_thread <- myThreadId@@ -103,14 +114,14 @@             liftIO $ atomically . writeTVar orig_thread_t . Just =<< myThreadId             yield 10             return ()-      ret <- runResourceT $ debSum 10000 orig_source $$ CL.consume+      ret <- runResourceT $ runConduit $ 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 <- runResourceT $ runConduit $ debSum 500000 orig_source .| CL.consume       ret `shouldBe` [10]       released `shouldSatisfyEventually'` (== True)     it "should release the resource in the original Source when the downstream Sink throws exception" $ do@@ -119,14 +130,15 @@             taken <- CL.take 4             _ <- error "Exception in downstream"             return taken-      runResourceT (debMonoid 500000 orig_source $$ sink) `shouldThrow` errorCall "Exception in downstream"+      (runResourceT $ runConduit $ debMonoid 500000 orig_source .| sink)+        `shouldThrow` errorCall "Exception in downstream"       released `shouldSatisfyEventually'` (== True)     it "should release the resource in the original Source when the original Source throws exception" $ do       -- Because the error is not handled, we'll see the error message       -- while running the test. (See above for the case "original       -- Source throwing exception")       (released, orig_source) <- attachResource (periodicSource 10000 ["a", "b"] >> error "Exception in source")-      ret <- runResourceT $ debMonoid 500000 orig_source $$ CL.consume+      ret <- runResourceT $ runConduit $ debMonoid 500000 orig_source .| CL.consume       ret `shouldBe` ["ab"]       released `shouldSatisfyEventually'` (== True)