Interact with an external process.

interact.hs

{-# LANGUAGE OverloadedStrings #-}

import Data.Monoid
import qualified Data.ByteString     as Bytes
import qualified Data.Text           as Text
import qualified Data.Text.Encoding  as Text
import qualified Data.Text.Lazy      as Text.Lazy
import qualified Data.Text.Lazy.IO   as Text.Lazy
import Pipes
import Pipes.Core
import qualified Pipes.Prelude       as Pipes (map)
import qualified                        Pipes.Text
import qualified Pipes.Text.Encoding as Pipes.Text (decodeUtf8)
import qualified Pipes.ByteString    as Pipes.Bytes

import Control.Monad
import Control.Concurrent.MVar
import Control.Concurrent.Async (concurrently)
import qualified Control.Foldl       as Foldl

import System.Process
import System.IO

import Lens.Family

type Emitted = Text.Text

type Received = Text.Lazy.Text

interactor :: Server Received Emitted IO ()
interactor = do
    received <- respond "hi!"
    liftIO (Text.Lazy.putStrLn ("received: " <> received))
    received <- respond "how are you?"
    liftIO (Text.Lazy.putStrLn ("received: " <> received))

emitter :: Server Received Emitted IO () -> MVar Received -> Producer Emitted IO ()
emitter server mvar = server >>~ receiver
    where
    receiver emitted = do
        yield emitted
        line <- liftIO (takeMVar mvar)
        request line >>= receiver

sendLinesToInteractor :: Producer Bytes.ByteString IO () -> MVar Received -> IO ()
sendLinesToInteractor producer mvar = runEffect (lineProducer >-> toMVar mvar)  
    where
    lineProducer = Foldl.purely Pipes.Text.folds lineFold lineFree 
    lineFold = Text.Lazy.fromChunks <$> Foldl.list
    -- errors are ignored with void!    
    lineFree = view Pipes.Text.lines (void (Pipes.Text.decodeUtf8 producer))

    toMVar :: MVar a -> Consumer a IO r
    toMVar mvar = forever (do
        line <- await
        liftIO (putMVar mvar line))    

main :: IO ()
main = do
   mvar <- newEmptyMVar
   (Just sin, Just sout, _, _) <- 
       createProcess (shell "cat") { std_in = CreatePipe, std_out = CreatePipe }
   hSetBuffering sin NoBuffering
   hSetBuffering sout NoBuffering
   concurrently 
        (runEffect (emitter interactor mvar 
                    >->
                    Pipes.map (Text.encodeUtf8 . (`mappend` "\n"))
                    >->
                    Pipes.Bytes.toHandle sin)
         >>
         hClose sin)
        (sendLinesToInteractor (Pipes.Bytes.fromHandle sout) mvar)
   return ()