diff --git a/pipes-misc.cabal b/pipes-misc.cabal
--- a/pipes-misc.cabal
+++ b/pipes-misc.cabal
@@ -1,5 +1,5 @@
 name:                pipes-misc
-version:             0.2.2.1
+version:             0.2.3.0
 synopsis:            Miscellaneous utilities for pipes, required by glazier-tutorial
 description:         Please see README.md
 homepage:            https://github.com/louispan/pipes-misc#readme
@@ -16,8 +16,15 @@
 library
   hs-source-dirs:      src
   exposed-modules:     Pipes.Misc
+                       Pipes.Misc.Concurrent
+                       Pipes.Misc.State.Lazy
+                       Pipes.Misc.State.Strict
+                       Pipes.Misc.Time
+                       Pipes.Misc.Util
   build-depends:       base >= 4.7 && < 5
+                     , clock >= 0.7 && < 1
                      , lens >= 4 && < 5
+                     , Decimal >= 0.4 && < 1
                      , mtl >= 2 && < 3
                      , pipes >= 4 && < 5
                      , pipes-category >= 0.2 && < 0.3
@@ -34,8 +41,11 @@
   main-is:             Spec.hs
   build-depends:       base >= 4.7 && < 5
                      , lens >=4 &&< 5
+                     , mmorph >= 1 && < 2
                      , pipes >= 4 && < 5
+                     , pipes-concurrency >= 2 && < 3
                      , pipes-misc
+                     , stm >= 2.4 && < 3
                      , transformers >= 0.4 && < 0.6
                      , hspec >= 2 && < 3
   ghc-options:         -Wall -threaded -rtsopts -with-rtsopts=-N
diff --git a/src/Pipes/Misc.hs b/src/Pipes/Misc.hs
--- a/src/Pipes/Misc.hs
+++ b/src/Pipes/Misc.hs
@@ -1,130 +1,11 @@
-{-# LANGUAGE FlexibleContexts #-}
-{-# LANGUAGE RankNTypes #-}
-
--- | Miscellaneous utilities for pipes, required by glazier-tutorial
-module Pipes.Misc where
-
-import Control.Applicative
-import Control.Arrow
-import Control.Concurrent
-import Control.Concurrent.STM
-import Control.Lens
-import Control.Monad
-import Control.Monad.State.Strict
-import Control.Monad.Trans.Maybe
-import Control.Monad.Except
-import qualified Data.List.NonEmpty as NE
-import qualified Pipes as P
-import qualified Pipes.Concurrent as PC
-import qualified Pipes.Prelude as PP
-import qualified Pipes.Shaft as PS
-
--- | Like Pipes.Concurrent.fromInput, but stays in STM.
--- Using @hoist atomically@ to convert to IO monad seems to work.
--- Do not use @unsafeHoist atomically@.
-fromInputSTM :: PC.Input a -> P.Producer' a STM ()
-fromInputSTM as = void $ runMaybeT $ forever $ do
-    a <- MaybeT $ lift $ PC.recv as
-    lift $ P.yield a
-
--- | Like Pipes.Concurrent.toOutput, but stays in STM.
--- Using @hoist atomically@ to convert to IO monad seems to work.
--- Do not use @unsafeHoist atomically@.
-toOutputSTM :: PC.Output a -> P.Consumer' a STM ()
-toOutputSTM output = void $ runMaybeT $ forever $ do
-    a <- lift P.await
-    p <- lift $ lift $ PC.send output a
-    guard p
-
--- | Reads as much as possible from an input and return a list of all unblocked values read.
--- Blocks if the first value read is blocked.
-batch :: PC.Input a -> PC.Input (NE.NonEmpty a)
-batch (PC.Input xs) = PC.Input $ do
-    x <- xs
-    case x of
-        Nothing -> pure Nothing
-        Just x' -> do
-            xs' <- runExceptT . tryNext $ x' NE.:| []
-            case xs' of
-                Left ys -> pure (Just ys)
-                Right ys -> pure (Just ys)
-  where
-      tryNext ys = do
-          ys' <- ExceptT $ (tryCons ys <$> xs) <|> pure (Left ys)
-          tryNext ys'
-      tryCons ys x = case x of
-          Nothing -> Left ys -- return successful reads so far
-          Just x' -> Right $ x' NE.<| ys
-
--- | Given a size and a initial tail, create a pipe that
--- will buffer the output of a producer.
--- This pipe is stateful, and will only buffer until the immediate connecting
--- producer is finished.
--- @
--- forever $ do
---   a <- await
---   yield a >-> buffer 2 [] -- will only ever result a producer of single 'a : []'.
--- @
--- @
--- (forever $ do
---   a <- await
---   yield a
--- ) >-> buffer 2 [] -- will buffer properly and produce '[latest, old]'
--- @
-buffer :: Monad m => Int -> [a] -> P.Pipe a [a] m r
-buffer n as = do
-  a <- P.await
-  let as' = take n $ a : as
-  case forceSpine as' of -- TODO: can we leave this lazy?
-    () -> do
-      P.yield as'
-      buffer n as'
- where
-  -- from https://ro-che.info/articles/2015-05-28-force-list
-  forceSpine = foldr (const id) ()
-
--- | Store the output of the pipe into a MonadState.
-store :: MonadState s m => Getter a b -> Setter' s b -> P.Pipe a a m r
-store v s = forever $ do
-  a <- P.await
-  s .= view v a
-  P.yield a
-
--- | Yields a view into the stored value.
-retrieve :: MonadState s m => Getter s b -> P.Pipe a (b, a) m r
-retrieve v = forever $ do
-  a <- P.await
-  s <- get
-  P.yield (view v s, a)
-
--- | Run a pipe in a larger stream, using view function and modify function
--- of the larger stream.
-locally ::
-  Monad m =>
-     (s -> a)
-  -> (b -> s -> t)
-  -> P.Pipe a b m r
-  -> P.Pipe s t m r
-locally viewf modifyf p =
-  PP.map (\s -> (s, s))
-  P.>-> PS.runShaft (first $ PS.Shaft $ PP.map viewf P.>-> p)
-  P.>-> PP.map (uncurry modifyf)
-
--- | Do something with the state everytime there is a yield.
-onState :: (MonadState s m) => (s -> m ()) -> P.Pipe a a m r
-onState f = PP.mapM $ \a -> do
-    s <- get
-    f s
-    pure a
+module Pipes.Misc
+    ( module Pipes.Misc.Concurrent
+    , module Pipes.Misc.State.Strict
+    , module Pipes.Misc.Time
+    , module Pipes.Misc.Util
+    ) where
 
--- | Add a delay after every yield
--- To avoid delaying the first yield use:
---
--- @
--- Pipes.pull () >> delay d
--- @
---
-delay :: MonadIO io => Int -> P.Pipe a a io ()
-delay i = PP.mapM $ \a -> do
-    liftIO $ threadDelay i
-    pure a
+import Pipes.Misc.Concurrent
+import Pipes.Misc.State.Strict
+import Pipes.Misc.Time
+import Pipes.Misc.Util
diff --git a/src/Pipes/Misc/Concurrent.hs b/src/Pipes/Misc/Concurrent.hs
new file mode 100644
--- /dev/null
+++ b/src/Pipes/Misc/Concurrent.hs
@@ -0,0 +1,55 @@
+{-# LANGUAGE RankNTypes #-}
+
+module Pipes.Misc.Concurrent where
+
+import Control.Concurrent
+import Control.Concurrent.STM
+import Control.Monad
+import Control.Monad.Trans.Class
+import Control.Monad.Trans.Maybe
+import qualified Pipes as P
+import qualified Pipes.Concurrent as PC
+
+-- | Like Pipes.Concurrent.fromInput, but stays in STM.
+-- Using @hoist atomically@ to convert to IO monad seems to work.
+-- Do not use @unsafeHoist atomically@.
+-- Each transaction is `atomically` scoped around each yield,
+-- so be careful when `Pipes.Prelude.filter` or similar pipes to remove yields
+-- as this results in larger transactions and it may cause BlockIndefinitelyOnSTM exceptions.
+-- Intead, use Monoids to yield mempty so that the STM state changes.
+fromInputSTM :: PC.Input a -> P.Producer' a STM ()
+fromInputSTM as = void $ runMaybeT $ forever $ do
+    a <- MaybeT $ lift $ PC.recv as
+    lift $ P.yield a
+{-# INLINABLE fromInputSTM #-}
+
+-- | Like Pipes.Concurrent.toOutput, but stays in STM.
+-- Using @hoist atomically@ to convert to IO monad seems to work.
+-- Do not use @unsafeHoist atomically@.
+toOutputSTM :: PC.Output a -> P.Consumer' a STM ()
+toOutputSTM output = void $ runMaybeT $ forever $ do
+    a <- lift P.await
+    p <- lift $ lift $ PC.send output a
+    guard p
+{-# INLINABLE toOutputSTM #-}
+
+-- | Convert PC.Output @a -> STM Bool@ to @a -> MaybeT STM ()@
+toOutputMaybeT :: PC.Output a -> a -> MaybeT STM ()
+toOutputMaybeT output = (MaybeT . fmap guard) <$> PC.send output
+{-# INLINABLE toOutputMaybeT #-}
+
+-- | Converts a Producer in IO monad to a producer in STM monad.
+mkProducerSTM :: PC.Buffer a -> P.Producer a IO () -> IO (P.Producer a STM ())
+mkProducerSTM b xs = do
+    (output, input) <- PC.spawn b
+    void . forkIO . void . forever . P.runEffect $ xs P.>-> PC.toOutput output
+    pure (fromInputSTM input)
+{-# INLINABLE mkProducerSTM #-}
+
+-- | Converts a Producer in IO monad to a producer in STM monad. Also returns the seal.
+mkProducerSTM' :: PC.Buffer a -> P.Producer a IO () -> IO (STM (), P.Producer a STM ())
+mkProducerSTM' b xs = do
+    (output, input, seal) <- PC.spawn' b
+    void . forkIO . void . forever . P.runEffect $ xs P.>-> PC.toOutput output
+    pure (seal, fromInputSTM input)
+{-# INLINABLE mkProducerSTM' #-}
diff --git a/src/Pipes/Misc/State/Lazy.hs b/src/Pipes/Misc/State/Lazy.hs
new file mode 100644
--- /dev/null
+++ b/src/Pipes/Misc/State/Lazy.hs
@@ -0,0 +1,40 @@
+{-# LANGUAGE RankNTypes #-}
+
+module Pipes.Misc.State.Lazy where
+
+import Control.Lens
+import Control.Monad.State.Lazy
+import qualified Pipes as P
+import qualified Pipes.Prelude as PP
+
+-- | Store the output of the pipe into a MonadState.
+store :: MonadState s m => Getter a b -> Setter' s b -> P.Pipe a a m r
+store v s = forever $ do
+  a <- P.await
+  s .= view v a
+  P.yield a
+{-# INLINABLE store #-}
+
+-- | Yields a view into the stored value.
+retrieve :: MonadState s m => Getter s b -> P.Pipe a (b, a) m r
+retrieve v = forever $ do
+  a <- P.await
+  s <- get
+  P.yield (view v s, a)
+{-# INLINABLE retrieve #-}
+
+-- | Yields a view into the stored value
+retrieve' :: MonadState s m => Getter s b -> P.Pipe () b m r
+retrieve' v = forever $ do
+  P.await
+  s <- get
+  P.yield (view v s)
+{-# INLINABLE retrieve' #-}
+
+-- | Do something with the state everytime there is a yield.
+onState :: (MonadState s m) => (s -> m ()) -> P.Pipe a a m r
+onState f = PP.mapM $ \a -> do
+    s <- get
+    f s
+    pure a
+{-# INLINABLE onState #-}
diff --git a/src/Pipes/Misc/State/Strict.hs b/src/Pipes/Misc/State/Strict.hs
new file mode 100644
--- /dev/null
+++ b/src/Pipes/Misc/State/Strict.hs
@@ -0,0 +1,40 @@
+{-# LANGUAGE RankNTypes #-}
+
+module Pipes.Misc.State.Strict where
+
+import Control.Lens
+import Control.Monad.State.Strict
+import qualified Pipes as P
+import qualified Pipes.Prelude as PP
+
+-- | Store the output of the pipe into a MonadState.
+store :: MonadState s m => Getter a b -> Setter' s b -> P.Pipe a a m r
+store v s = forever $ do
+  a <- P.await
+  s .= view v a
+  P.yield a
+{-# INLINABLE store #-}
+
+-- | Yields a view into the stored value.
+retrieve :: MonadState s m => Getter s b -> P.Pipe a (b, a) m r
+retrieve v = forever $ do
+  a <- P.await
+  s <- get
+  P.yield (view v s, a)
+{-# INLINABLE retrieve #-}
+
+-- | Yields a view into the stored value
+retrieve' :: MonadState s m => Getter s b -> P.Pipe () b m r
+retrieve' v = forever $ do
+  P.await
+  s <- get
+  P.yield (view v s)
+{-# INLINABLE retrieve' #-}
+
+-- | Do something with the state everytime there is a yield.
+onState :: (MonadState s m) => (s -> m ()) -> P.Pipe a a m r
+onState f = PP.mapM $ \a -> do
+    s <- get
+    f s
+    pure a
+{-# INLINABLE onState #-}
diff --git a/src/Pipes/Misc/Time.hs b/src/Pipes/Misc/Time.hs
new file mode 100644
--- /dev/null
+++ b/src/Pipes/Misc/Time.hs
@@ -0,0 +1,71 @@
+{-# LANGUAGE RankNTypes #-}
+
+-- | You can use the Arrow instance to get different types of ticker time. Eg:
+--
+-- @
+-- diffTimeEvery :: MonadIO io => C.Clock -> Int -> P.Producer' C.TimeSpec io r
+-- diffTimeEvery clock micros = always () P.>-> delay micros P.>-> ticker clock P.>-> diffTime
+--
+-- import Control.Arrow
+-- import qualified Control.Category as Cat
+-- import qualified Pipes.Shaft as PS
+--
+-- diffAndTickEvery :: MonadIO io => C.Clock -> Int -> P.Producer' (C.TimeSpec, C.TimeSpec) io r
+-- diffAndTickEvery clock micros = always () P.>-> delay micros P.>-> ticker clock P.>->
+--    PS.runShaft (PS.Shaft diffTime &&& Cat.id)
+--
+-- @
+--
+module Pipes.Misc.Time where
+
+import Control.Concurrent
+import Control.Monad.IO.Class
+import qualified Pipes as P
+import qualified Pipes.Prelude as PP
+import qualified System.Clock as C
+import qualified Data.Decimal as D
+import qualified Pipes.Misc.Util as PM
+
+-- | Add a delay after every await
+delay :: MonadIO io => Int -> P.Pipe a a io r
+delay i = PP.mapM $ \a -> do
+    liftIO $ threadDelay i
+    pure a
+{-# INLINABLE delay #-}
+
+-- | After the first await, add a delay after every subsequent await.
+delay' :: MonadIO io => Int -> P.Pipe a a io r
+delay' i = do
+    a <- P.await
+    P.yield a
+    delay i
+{-# INLINABLE delay' #-}
+
+-- | obtain the threadDelay given a fps
+fps :: D.Decimal -> Int
+fps x = truncate (D.roundTo 0 (D.Decimal 0 1000000 / x))
+{-# INLINABLE fps #-}
+
+-- | Continuously yield the clock time
+-- Use with delay to reduce the yield rate. Eg:
+--
+-- @
+-- tickEvery :: MonadIO io => C.Clock -> Int -> P.Producer' C.TimeSpec io r
+-- tickEvery clock micros = always () P.>-> delay micros P.>-> ticker clock
+-- @
+--
+ticker :: MonadIO io => C.Clock -> P.Pipe () C.TimeSpec io r
+ticker clock = P.for P.cat $ \() -> do
+    t <- liftIO $ C.getTime clock
+    P.yield t
+{-# INLINABLE ticker #-}
+
+-- | Converts a stream of times, into a stream of delta time. The first yield is zero.
+diffTime :: Monad m => P.Pipe C.TimeSpec C.TimeSpec m r
+diffTime = PM.compare C.diffTimeSpec (C.TimeSpec 0 0)
+{-# INLINABLE diffTime #-}
+
+-- | Converts a stream of epoch times into a stream of epoch time, where zero is the first yielded time.
+resetEpoch :: Monad m => P.Pipe C.TimeSpec C.TimeSpec m r
+resetEpoch = PM.compare' C.diffTimeSpec
+{-# INLINABLE resetEpoch #-}
diff --git a/src/Pipes/Misc/Util.hs b/src/Pipes/Misc/Util.hs
new file mode 100644
--- /dev/null
+++ b/src/Pipes/Misc/Util.hs
@@ -0,0 +1,120 @@
+module Pipes.Misc.Util where
+
+import Control.Applicative
+import Control.Arrow
+import Control.Monad.Except
+import qualified Data.List.NonEmpty as NE
+import qualified Pipes as P
+import qualified Pipes.Concurrent as PC
+import qualified Pipes.Prelude as PP
+import qualified Pipes.Shaft as PS
+import Pipes.Internal (Proxy(..))
+
+
+-- | Reads as much as possible from an input and return a list of all unblocked values read.
+-- Blocks if the first value read is blocked.
+batch :: PC.Input a -> PC.Input (NE.NonEmpty a)
+batch (PC.Input xs) = PC.Input $ do
+    x <- xs
+    case x of
+        Nothing -> pure Nothing
+        Just x' -> do
+            xs' <- runExceptT . tryNext $ x' NE.:| []
+            case xs' of
+                Left ys -> pure (Just ys)
+                Right ys -> pure (Just ys)
+  where
+      tryNext ys = do
+          ys' <- ExceptT $ (tryCons ys <$> xs) <|> pure (Left ys)
+          tryNext ys'
+      tryCons ys x = case x of
+          Nothing -> Left ys -- return successful reads so far
+          Just x' -> Right $ x' NE.<| ys
+{-# INLINABLE batch #-}
+
+-- | Given a size and a initial tail, create a pipe that
+-- will buffer the output of a producer.
+-- This pipe is stateful, and will only buffer until the immediate connecting
+-- producer is finished.
+-- @
+-- forever $ do
+--   a <- await
+--   yield a >-> buffer 2 [] -- will only ever result a producer of single 'a : []'.
+-- @
+-- @
+-- (forever $ do
+--   a <- await
+--   yield a
+-- ) >-> buffer 2 [] -- will buffer properly and produce '[latest, old]'
+-- @
+buffer :: Monad m => Int -> [a] -> P.Pipe a [a] m r
+buffer n as = do
+  a <- P.await
+  let as' = take n $ a : as
+  case forceSpine as' of -- TODO: can we leave this lazy?
+    () -> do
+      P.yield as'
+      buffer n as'
+ where
+  -- from https://ro-che.info/articles/2015-05-28-force-list
+  forceSpine = foldr (const id) ()
+{-# INLINABLE buffer #-}
+
+-- | Run a pipe in a larger stream, using view function and modify function
+-- of the larger stream.
+locally ::
+  Monad m =>
+     (s -> a)
+  -> (b -> s -> t)
+  -> P.Pipe a b m r
+  -> P.Pipe s t m r
+locally viewf modifyf p =
+  PP.map (\s -> (s, s))
+  P.>-> PS.runShaft (first $ PS.Shaft $ PP.map viewf P.>-> p)
+  P.>-> PP.map (uncurry modifyf)
+{-# INLINABLE locally #-}
+
+-- | Given comparison function and an initial value.
+-- yield the result of comparing the value await with the previously awaited value.
+compare :: Monad m => (a -> a -> b) -> a -> P.Pipe a b m r
+compare f i = do
+    a <- P.await
+    P.yield (f a i)
+    go a
+  where
+    go a = do
+        b <- P.await
+        P.yield (f b a)
+        go b
+{-# INLINABLE compare #-}
+
+-- | Given comparison function
+-- yield the result of comparing the value await with the first awaited value.
+compare' :: Monad m => (a -> a -> b) -> P.Pipe a b m r
+compare' f = do
+    i <- P.await
+    P.yield (f i i)
+    go i
+  where
+    go i = forever $ do
+        a <- P.await
+        P.yield (f a i)
+{-# INLINABLE compare' #-}
+
+-- | constantly yields the given value
+always :: Monad m => a -> P.Producer a m r
+always = forever . P.yield
+{-# INLINABLE always #-}
+
+-- | Makes the Producer return/pure the last value yielded, or the input value if nothing
+-- was yielded
+lastOr :: Monad m => a -> P.Producer a m () -> P.Producer a m a
+lastOr = go
+  where
+    go i p =
+        case p of
+            Request a' fa -> Request a' (go i . fa)
+            Respond b fb' -> Respond b (go b . fb')
+            M m -> M (m >>= pure . go i)
+            Pure () -> Pure i
+{-# INLINABLE lastOr #-}
diff --git a/test/Spec.hs b/test/Spec.hs
--- a/test/Spec.hs
+++ b/test/Spec.hs
@@ -1,8 +1,11 @@
 module Main where
 
+import Control.Concurrent.STM
 import Control.Lens
+import Control.Monad.Morph
 import Data.Foldable
 import qualified Pipes as P
+import qualified Pipes.Concurrent as PC
 import qualified Pipes.Misc as PM
 import qualified Pipes.Prelude as PP
 import Test.Hspec
@@ -27,3 +30,6 @@
                         P.>-> PP.map (\a -> (a, a))
                         P.>-> PM.locally (view _2) (set _2) (PP.map (+ 10)))
               xs `shouldBe` zip data1 ((+ 10) <$> data1)
+          it "Filtering STM Producer blocks" $ do
+              sig <- PM.mkProducerSTM (PC.bounded 1) sig1
+              PP.toListM (hoist atomically (sig P.>-> PP.filter even)) `shouldThrow` anyException