diff --git a/Control/Monad/Trans/Resource.hs b/Control/Monad/Trans/Resource.hs --- a/Control/Monad/Trans/Resource.hs +++ b/Control/Monad/Trans/Resource.hs @@ -43,12 +43,14 @@ , ResourceThrow (..) -- ** Low-level , HasRef (..) + , InvalidAccess (..) + , resourceActive ) where import Data.Typeable import Data.IntMap (IntMap) import qualified Data.IntMap as IntMap -import Control.Exception (SomeException) +import Control.Exception (SomeException, throw, Exception) import Control.Monad.Trans.Control ( MonadTransControl (..), MonadBaseControl (..) , ComposeSt, defaultLiftBaseWith, defaultRestoreM @@ -260,6 +262,7 @@ data ReleaseMap base = ReleaseMap !NextKey !RefCount !(IntMap (base ())) + | ReleaseMapClosed -- | The Resource transformer. This transformer keeps track of all registered -- actions, and calls them upon exit (via 'runResourceT'). Actions may be @@ -317,11 +320,26 @@ => Ref base (ReleaseMap base) -> base () -> base ReleaseKey -register' istate rel = atomicModifyRef' istate $ \(ReleaseMap key rf m) -> - ( ReleaseMap (key + 1) rf (IntMap.insert key rel m) - , ReleaseKey key - ) +register' istate rel = atomicModifyRef' istate $ \rm -> + case rm of + ReleaseMap key rf m -> + ( ReleaseMap (key + 1) rf (IntMap.insert key rel m) + , ReleaseKey key + ) + ReleaseMapClosed -> throw $ InvalidAccess "register'" +data InvalidAccess = InvalidAccess { functionName :: String } + deriving Typeable + +instance Show InvalidAccess where + show (InvalidAccess f) = concat + [ "Control.Monad.Trans.Resource." + , f + , ": The mutable state is being accessed after cleanup. Please contact the maintainers." + ] + +instance Exception InvalidAccess + -- | Call a release action early, and deregister it from the list of cleanup -- actions to be performed. release :: Resource m @@ -344,25 +362,30 @@ ( ReleaseMap next rf $ IntMap.delete key m , Just action ) + lookupAction ReleaseMapClosed = throw $ InvalidAccess "release'" stateAlloc :: HasRef m => Ref m (ReleaseMap m) -> m () stateAlloc istate = do - atomicModifyRef' istate $ \(ReleaseMap nk rf m) -> - (ReleaseMap nk (rf + 1) m, ()) + atomicModifyRef' istate $ \rm -> + case rm of + ReleaseMap nk rf m -> + (ReleaseMap nk (rf + 1) m, ()) + ReleaseMapClosed -> throw $ InvalidAccess "stateAlloc" stateCleanup :: HasRef m => Ref m (ReleaseMap m) -> m () stateCleanup istate = mask_ $ do - (rf, m) <- atomicModifyRef' istate $ \(ReleaseMap nk rf m) -> - (ReleaseMap nk (rf - 1) m, (rf - 1, m)) - if rf == minBound - then do + mm <- atomicModifyRef' istate $ \rm -> + case rm of + ReleaseMap nk rf m -> + let rf' = rf - 1 + in if rf' == minBound + then (ReleaseMapClosed, Just m) + else (ReleaseMap nk rf' m, Nothing) + ReleaseMapClosed -> throw $ InvalidAccess "stateCleanup" + case mm of + Just m -> mapM_ (\x -> try x >> return ()) $ IntMap.elems m - -- Trigger an exception consistently for one race condition: - -- let's put an undefined value in the state. If somehow - -- another thread is still able to access it, at least we get - -- clearer error messages. - writeRef' istate $ error "Control.Monad.Trans.Resource.stateCleanup: There is a bug in the implementation. The mutable state is being accessed after cleanup. Please contact the maintainers." - else return () + Nothing -> return () -- | Unwrap a 'ResourceT' transformer, and call all registered release actions. -- @@ -382,9 +405,9 @@ -- strip or add new transformers to a stack, e.g. to run a @ReaderT@. Note that -- the original and new monad must both have the same 'Base' monad. transResourceT :: (Base m ~ Base n) - => (m a -> n a) + => (m a -> n b) -> ResourceT m a - -> ResourceT n a + -> ResourceT n b transResourceT f (ResourceT mx) = ResourceT (\r -> f (mx r)) -------- All of our monad et al instances @@ -523,3 +546,13 @@ (return ()) (stateCleanup r) (restore $ f r)) + +-- | Determine if the current @ResourceT@ is still active. This is necessary +-- for such cases as lazy I\/O, where an unevaluated thunk may still refer to a +-- closed @ResourceT@. +resourceActive :: Resource m => ResourceT m Bool +resourceActive = ResourceT $ \rmMap -> do + rm <- resourceLiftBase $ readRef' rmMap + case rm of + ReleaseMapClosed -> return False + _ -> return True diff --git a/Data/Conduit/Lazy.hs b/Data/Conduit/Lazy.hs --- a/Data/Conduit/Lazy.hs +++ b/Data/Conduit/Lazy.hs @@ -8,20 +8,25 @@ import Data.Conduit import System.IO.Unsafe (unsafeInterleaveIO) -import Control.Monad.Trans.Control +import Control.Monad.Trans.Control (MonadBaseControl, liftBaseOp_) +import Control.Monad.Trans.Resource (resourceActive) -- | Use lazy I\/O to consume all elements from a @Source@. -- -- Since 0.2.0 -lazyConsume :: MonadBaseControl IO m => Source m a -> ResourceT m [a] +lazyConsume :: (Resource m, MonadBaseControl IO m) => Source m a -> ResourceT m [a] lazyConsume src0 = do go src0 where go src = liftBaseOp_ unsafeInterleaveIO $ do - res <- sourcePull src - case res of - Closed -> return [] - Open src' x -> do - y <- go src' - return $ x : y + ra <- resourceActive + if ra + then do + res <- sourcePull src + case res of + Closed -> return [] + Open src' x -> do + y <- go src' + return $ x : y + else return [] diff --git a/Data/Conduit/List.hs b/Data/Conduit/List.hs --- a/Data/Conduit/List.hs +++ b/Data/Conduit/List.hs @@ -18,6 +18,7 @@ , take , drop , head + , zip , peek , consume , sinkNull @@ -322,3 +323,20 @@ -- Since 0.2.0 sourceNull :: Resource m => Source m a sourceNull = mempty + +-- | Combines two sources. The new source will stop producing once either +-- source has been exhausted. +-- +-- Since 0.2.2 +zip :: Resource m => Source m a -> Source m b -> Source m (a, b) +zip sa sb = Source pull close + where + pull = do ra <- sourcePull sa + case ra of + Closed -> return Closed + Open ra' a -> do rb <- sourcePull sb + case rb of + Closed -> return Closed + Open rb' b -> return $ Open (zip ra' rb') (a, b) + close = sourceClose sa >> sourceClose sb + diff --git a/conduit.cabal b/conduit.cabal --- a/conduit.cabal +++ b/conduit.cabal @@ -1,5 +1,5 @@ Name: conduit -Version: 0.2.1 +Version: 0.2.2 Synopsis: Streaming data processing library. Description: Conduits are an approach to the streaming data problem. It is meant as an alternative to enumerators\/iterators, hoping to address the same issues with different trade-offs based on real-world experience with enumerators. For more information, see . diff --git a/test/main.hs b/test/main.hs --- a/test/main.hs +++ b/test/main.hs @@ -121,6 +121,11 @@ bs1 @=? bs2 bs1 @=? bs3 + describe "zipping" $ do + it "zipping two small lists" $ do + res <- runResourceT $ CL.zip (CL.sourceList [1..10]) (CL.sourceList [11..12]) C.$$ CL.consume + res @=? zip [1..10 :: Int] [11..12 :: Int] + describe "Monad instance for Sink" $ do it "binding" $ do x <- runResourceT $ CL.sourceList [1..10] C.$$ do @@ -242,6 +247,10 @@ ) nums <- CLazy.lazyConsume $ mconcat $ map incr [1..10] liftIO $ nums @?= [1..10] + + it' "returns nothing outside ResourceT" $ do + bss <- runResourceT $ CLazy.lazyConsume $ CB.sourceFile "test/main.hs" + bss @?= [] describe "sequence" $ do it "simple sink" $ do