Implementation of sourceTimer in Conduit 0.5.x

sourceTimer.hs

import Control.Concurrent           (forkIO,
                                     killThread,
                                     threadDelay)
import Control.Monad                (void)
import Control.Concurrent.STM       (atomically)
import Control.Monad.Trans          (MonadIO(..))
import Control.Concurrent.STM.TChan (TChan, newTChan, writeTChan, readTChan)
import Control.Monad.Trans.Resource (MonadResource, allocate)
import Data.Conduit                 (Source, awaitE)
import Data.Conduit.Internal        (Pipe(..))


sourceTChan :: MonadIO m => TChan a -> Source m a
sourceTChan ch = src
  where
    src = PipeM pull
    pull = do
      a <- liftIO $ atomically $ readTChan ch
      return $ HaveOutput src (return ()) a

sourceTimer :: MonadResource m => Int -> Source m Int
sourceTimer micro = do
    chan <- liftIO . atomically $ newTChan
    void $ allocate (forkIO $ timerThread 0 chan) (killThread)
    sourceTChan chan
  where
    timerThread i chan = do
      threadDelay micro
      liftIO . atomically $ writeTChan chan i
      timerThread (i + 1) chan