Haskell Language
パイプ

備考

プロデューサー

Producerは、川下消費の価値をyieldことができるモナド的な行動です：

type Producer b = Proxy X () () b
yield :: Monad m => a -> Producer a m ()

例えば：

naturals :: Monad m => Producer Int m ()
naturals = each [1..] -- each is a utility function exported by Pipes

もちろん、他の値の関数でもあるProducer持つことができます：

naturalsUntil :: Monad m => Int -> Producer Int m ()
naturalsUntil n = each [1..n]

消費者

Consumerは上流からの値をawaitだけです。

type Consumer a = Proxy () a () X
await :: Monad m => Consumer a m a

例えば：

fancyPrint :: MonadIO m => Consumer String m ()
fancyPrint = forever $ do
  numStr <- await
  liftIO $ putStrLn ("I received: " ++ numStr)

パイプ

パイプはawaitことができ、 yieldすることもできます。

type Pipe a b = Proxy () a () b

このパイプはIntを待ち、それをString変換します：

intToStr :: Monad m => Pipe Int String m ()
intToStr = forever $ await >>= (yield . show)

runEffectでパイプを実行する

私たちは、使用runEffect私たち実行するためにPipe ：

main :: IO ()
main = do
  runEffect $ naturalsUntil 10 >-> intToStr >-> fancyPrint

runEffectは、入力または出力のない自己完結型ProxyであるEffectが必要であることに注意してください。

runEffect :: Monad m => Effect m r -> m r
type Effect = Proxy X () () X

（ Xは空の型で、 Voidとも呼ばれます）。

パイプの接続

>->を使用してProducer 、 Consumer 、およびPipeを接続し、より大きなPipe機能を構成します。

printNaturals :: MonadIO m => Effect m ()
printNaturals = naturalsUntil 10 >-> intToStr >-> fancyPrint

Producer 、 Consumer 、 Pipe 、およびEffectタイプはすべて、一般的なProxyタイプの観点から定義されています。したがって、 >->はさまざまな目的で使用できます。左の引数で定義される型は、右の引数で使用される型と一致する必要があります。

(>->) :: Monad m => Producer b m r -> Consumer b   m r -> Effect       m r
(>->) :: Monad m => Producer b m r -> Pipe     b c m r -> Producer   c m r
(>->) :: Monad m => Pipe   a b m r -> Consumer b   m r -> Consumer a   m r
(>->) :: Monad m => Pipe   a b m r -> Pipe     b c m r -> Pipe     a c m r

プロキシモナドトランス

pipesのコアデータ型は、 Proxyモナドトランスです。 Pipe 、 Producer 、 Consumerなどは、 Proxy観点から定義されています。

以来、 Proxyモナド変圧器である、の定義Pipe Sはモナドスクリプトの形取るawait及びyieldさらにベースモナドの影響を行う、値m 。

パイプとネットワーク通信の結合

パイプは、クライアントとサーバーの間の単純なバイナリ通信をサポートします

この例では：

1. クライアントはFirstMessage接続して送信しFirstMessage
2. サーバーは受信し、 DoSomething 0応答しDoSomething 0
3. クライアントはDoNothingを受信して​​応答しますDoNothing
4. ステップ2および3が無期限に繰り返される

ネットワークを介して交換されるコマンドデータタイプ：

-- Command.hs
{-# LANGUAGE DeriveGeneric #-}
module Command where
import Data.Binary
import GHC.Generics (Generic)

data Command = FirstMessage
           | DoNothing
           | DoSomething Int
           deriving (Show,Generic)

instance Binary Command

ここで、サーバーはクライアントが接続するのを待ちます。

module Server where

import Pipes 
import qualified Pipes.Binary as PipesBinary
import qualified Pipes.Network.TCP as PNT
import qualified Command as C
import qualified Pipes.Parse as PP
import qualified Pipes.Prelude as PipesPrelude

pageSize :: Int
pageSize = 4096

-- pure handler, to be used with PipesPrelude.map
pureHandler :: C.Command -> C.Command 
pureHandler c = c  -- answers the same command that we have receveid

-- impure handler, to be used with PipesPremude.mapM
sideffectHandler :: MonadIO m => C.Command -> m C.Command
sideffectHandler c = do
  liftIO $ putStrLn $ "received message = " ++ (show c)
  return $ C.DoSomething 0    
  -- whatever incoming command `c` from the client, answer DoSomething 0

main :: IO ()
main = PNT.serve (PNT.Host "127.0.0.1") "23456" $
  \(connectionSocket, remoteAddress) -> do
                 putStrLn $ "Remote connection from ip = " ++ (show remoteAddress)
                 _ <- runEffect $ do
                   let bytesReceiver = PNT.fromSocket connectionSocket pageSize
                   let commandDecoder = PP.parsed PipesBinary.decode bytesReceiver
                   commandDecoder >-> PipesPrelude.mapM sideffectHandler >-> for cat PipesBinary.encode >-> PNT.toSocket connectionSocket
                   -- if we want to use the pureHandler
                   --commandDecoder >-> PipesPrelude.map pureHandler >-> for cat PipesBinary.Encode >-> PNT.toSocket connectionSocket
                 return ()

クライアントはこうして接続します：

module Client where

import Pipes
import qualified Pipes.Binary as PipesBinary
import qualified Pipes.Network.TCP as PNT
import qualified Pipes.Prelude as PipesPrelude
import qualified Pipes.Parse as PP
import qualified Command as C

pageSize :: Int
pageSize = 4096

-- pure handler, to be used with PipesPrelude.amp
pureHandler :: C.Command -> C.Command 
pureHandler c = c  -- answer the same command received from the server

-- inpure handler, to be used with PipesPremude.mapM
sideffectHandler :: MonadIO m => C.Command -> m C.Command
sideffectHandler c = do
  liftIO $ putStrLn $ "Received: " ++ (show c)
  return C.DoNothing  -- whatever is received from server, answer DoNothing

main :: IO ()
main = PNT.connect ("127.0.0.1") "23456" $
  \(connectionSocket, remoteAddress) -> do
    putStrLn $ "Connected to distant server ip = " ++ (show remoteAddress)
    sendFirstMessage connectionSocket
    _ <- runEffect $ do
      let bytesReceiver = PNT.fromSocket connectionSocket pageSize
      let commandDecoder = PP.parsed PipesBinary.decode bytesReceiver
      commandDecoder >-> PipesPrelude.mapM sideffectHandler >-> for cat PipesBinary.encode >-> PNT.toSocket connectionSocket
    return ()

sendFirstMessage :: PNT.Socket -> IO ()
sendFirstMessage s = do
  _ <- runEffect $ do
    let encodedProducer = PipesBinary.encode C.FirstMessage 
    encodedProducer >-> PNT.toSocket s  
  return ()