iteratee-stm (empty) → 0.1

raw patch · 4 files changed

+157/−0 lines, 4 filesdep +basedep +iterateedep +stmsetup-changed

Dependencies added: base, iteratee, stm, stm-chans, transformers

Files

• LICENSE +30/−0
• Setup.hs +2/−0
• iteratee-stm.cabal +23/−0
• src/Data/Iteratee/STM.hs +102/−0

@@ -0,0 +1,30 @@+Copyright John W. Lato 2011++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 John W. Lato 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,2 @@+import Distribution.Simple+main = defaultMain

@@ -0,0 +1,23 @@+Name:                iteratee-stm+Version:             0.1+Synopsis:            Concurrent iteratees using STM+Description:         Enumerators and iteratees which read from/write to STM+                     channels.  This allows for processes with dedicated IO+                     threads.+Homepage:            http://www.tiresiaspress.us/~jwlato/haskell/iteratee-stm+License:             BSD3+License-file:        LICENSE+Author:              John W. Lato+Maintainer:          jwlato@gmail.com+Category:            Data+Build-type:          Simple+Cabal-version:       >=1.6++Library+  Hs-Source-Dirs:     src+  Exposed-modules:    Data.Iteratee.STM+  Build-depends:      base         >= 3      && < 5+                     ,iteratee     == 0.8.*+                     ,stm          == 2.2.*+                     ,stm-chans    == 1.2.*+                     ,transformers >= 0.2

@@ -0,0 +1,102 @@+module Data.Iteratee.STM (+  -- * Channel enumerator/iteratee primitives+  iterChan+ ,enumChan+  -- ** Channel control functions+ ,iterCloseChan+ ,enumCloseChan+  -- * Forking combinators+ ,forkIter+ ,forkEnum+)++where++import Control.Concurrent.STM.TBMChan+import Data.Iteratee as I++import Control.Applicative+import Control.Concurrent+import Control.Concurrent.STM+import Control.Exception (finally)+import Control.Monad+import Control.Monad.IO.Class++-- | return all available values from a TBMChan+drainChan :: TBMChan a -> STM [a]+drainChan chan = go []+  where+    go acc = tryReadTBMChan chan >>= \res -> case res of+      Just (Just a) -> go (a:acc)+      _             -> return acc++-- | Close a channel.+-- +iterCloseChan :: (Nullable s, MonadIO m) => TBMChan s -> Iteratee s m ()+iterCloseChan chan = liftIO . atomically $ closeTBMChan chan++-- | An iteratee which writes all its data to a TBMChan.+-- +-- The iteratee moves to a complete state when the channel is closed.+iterChan :: (Nullable s, MonadIO m) => TBMChan s -> Iteratee s m ()+iterChan chan = do+   stream_eof <- isFinished+   unless stream_eof $ do+     chnk <- getChunk+     wrote_chnk <- liftIO . atomically $ do+       isClosed <- isClosedTBMChan chan+       if isClosed then return False else writeTBMChan chan chnk >> return True+     if wrote_chnk+       then iterChan chan+       else idone () (Chunk chnk)++-- | Enumerate over data provided by a TBMChan.+enumChan :: (NullPoint s, MonadIO m) => TBMChan s -> Enumerator s m a+enumChan chan = enumFromCallback cb ()+ where+   cb () = do+       mres <- liftIO . atomically $ readTBMChan chan+       case mres of+           Nothing -> return $ Right ((False, ()), I.empty)+           Just s    -> return $ Right ((True, ()), s)++-- | An enumerator which closes the provided channel and sends EOF to the iteratee.+enumCloseChan :: (MonadIO m) => TBMChan s -> Enumerator s m a+enumCloseChan chan iter = do+  liftIO . atomically $ closeTBMChan chan+  enumEof iter++-- | Fork an enumerator to run in a separate thread, with a @sz@ upper bound on the+-- channel size.+-- +-- The current thread will wait for the forked thread to terminate+forkEnum+  :: (MonadIO m, Nullable s, NullPoint s)+  => Int+  -> Enumerator s IO ()+  -> Enumerator s m a+forkEnum sz enum iter = do+   chan <- liftIO $ newTBMChanIO sz+   mvar <- liftIO $ newEmptyMVar+   liftIO . forkIO $ ((enum >>> enumCloseChan chan) (iterChan chan) >>= run)+                     `finally` putMVar mvar ()+   i2 <- enumChan chan iter+   liftIO $ readMVar mvar+   return i2++-- | Fork an iteratee to run in a separate thread, with a @sz@ upper bound on the+-- channel size.+-- +-- The current thread will wait for the forked thread to finish before returning.+forkIter+  :: (Nullable s, NullPoint s, MonadIO m)+  => Int+  -> Iteratee s IO ()+  -> Iteratee s m ()+forkIter sz iter = do+  chan <- liftIO $ newTBMChanIO sz+  mvar <- liftIO $ newEmptyMVar+  liftIO . forkIO $ (enumChan chan iter >>= run) `finally` putMVar mvar ()+  iterChan chan+  iterCloseChan chan+  liftIO $ readMVar mvar