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 수 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 할 수 await .

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 들 모나드 스크립트 형태 취할 await 및 yield 별도로베이스 모나드에서 수행 효과, 값 m .

파이프와 네트워크 통신 결합

파이프는 클라이언트와 서버 간의 간단한 이진 통신을 지원합니다.

이 예에서 :

1. 클라이언트가 연결하여 FirstMessage 보냅니다.
2. 서버는 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 ()