Pipes `run` with State

I have a producer:

p :: Producer Message IO r.

I can process all messages using:

runEffect $ for p processMessage

where

processMessage :: Message -> Effect IO ().

How can I implement stateful processing using something like:

processMessage :: Message -> Effect (StateT MyState IO) () ?

Short answer:

1. Modify your producer to be agnostic to the monad it's running in
2. Your processMessage is fine
3. The runEffect returns StateT MyState IO (), you need to evaluate it

Longer answer with a dummy example:

Your producer is locked into the IO monad, you need to modify it to be either in MonadIO m or in the explicit state monad.

import Control.Monad.State
import Pipes

type Message = Int

p :: MonadIO m => Producer Message m ()
p = each [1..10]

The signature of your processMessage is already fine. I'm following your signature and adding some simple logic to exercise the IO and State functionality

processMessage :: Message -> Effect (StateT MyState IO) ()
processMessage msg = do
  modify (+ msg)
  liftIO (print msg)

Then the final step. runEffect :: Monad m => Effect m r -> m r, if you substitute the m with a concrete type, this ends up being runEffect :: Effect (StateT MyState IO) () -> StateT MyState IO (), meaning that you'll be left with the state monad which still needs to be executed. There's three variants for executing the state monad, runStateT, evalStateT and execStateT. I chose execStateT :: StateT MyState IO () -> IO MyState variant in this example, but choose whichever you need in your case.

main :: IO ()
main = do
  st <- execStateT (runEffect $ for p processMessage) 0
  putStrLn $ "End state: " <> show st