pcon-appendix.hs

-- game.hs

import           Control.Concurrent hiding (yield)
import           Control.Concurrent.Async
import           Control.Monad
import           Control.Monad.IO.Class
import           Control.Monad.Trans.Maybe
import           Control.Monad.Trans.State.Strict

import           Pipes
import           Pipes.Concurrent
import qualified Pipes.Prelude as P

-- The game events
data Event = Harm Integer | Heal Integer | Quit

-- The game state
type Health = Integer

handler :: Consumer Event (StateT Health (MaybeT IO)) r
handler = forever $ do
    event  <- await
    health <- lift $ do
        case event of
            Harm n -> modify (subtract n)
            Heal n -> modify (+        n)
            Quit   -> mzero
        get
    liftIO $ putStrLn $ "Health = " ++ show health

user :: Producer Event IO r
user = forever $ do
    command <- lift getLine
    case command of
        "potion" -> yield (Heal 10)
        "quit"   -> yield  Quit
        _        -> lift $ putStrLn "Invalid command"

acidRain :: Producer Event IO r
acidRain = forever $ do
    yield (Harm 1)
    lift $ threadDelay 2000000

main = do
    (input, output) <- spawn Unbounded
    forkIO $ do run $ (acidRain >-> toInput input)
                performGC
    forkIO $ do run $ (user     >-> toInput input)
                performGC
    runMaybeT $ (`evalStateT` 100) $ run $
       (hoist (lift . lift) . fromOutput output >-> handler)

-- work.hs


worker :: (Show a) => Int -> Consumer a IO r
worker i = forever $ do
    a <- await
    lift $ threadDelay 1000000
    lift $ putStrLn $ "Worker #" ++ show i ++ ": Processed " ++ show a

user' :: Producer String IO ()
user' = P.stdin >-> P.takeWhile (/= "quit")

main' = do
    let buffer1 = Unbounded
        buffer2 = Single
        buffer3 = Bounded 100
    (input, output) <- spawn buffer1
    let consumer1 i =              worker i
        consumer2 i = P.take 2 >-> worker i
    as <- forM [1..3] $ \i -> async $ do
        run $ (fromOutput output >-> consumer1 i)
        performGC
    let producer1 = for (each [1..10])
        producer2 = user
        producer3 = (for (each [1..])) >-> (P.tee $ lift $ putStrLn)
    a  <- async $ do run $ (producer1 >-> toInput input)
                     performGC
    mapM_ wait (a:as)

-- peek.hs

-- Fast input updates
inputDevice :: (Monad m) => Producer Integer m ()
inputDevice = for (each [1..])

-- Slow output updates
outputDevice :: Consumer Integer IO r
outputDevice () = forever $ do
    n <- await
    lift $ do
        print n
        threadDelay 1000000

main'' = do
    (input, output) <- spawn (Latest 0)
    a1 <- async $ do
        run $ (inputDevice >-> toInput input)
        performGC
    a2 <- async $ do
        run $ (fromOutput output >-> P.take 5 >-> outputDevice)
        performGC
    mapM_ wait [a1, a2]

-- callback.hs


onLines :: (String -> IO a) -> IO b
onLines callback = forever $ do
    str <- getLine
    callback str

onLines' :: Producer String IO ()
onLines' = do
    (input, output) <- lift $ spawn Single
    lift $ forkIO $ onLines (\str -> atomically $ send input str)
    fromOutput output ()

main''' = run $ (onLines' >-> P.takeWhile (/= "quit") >-> lift $ putStrLn) ()