1

我正在使用管道实现一个简单的网络协议；该协议是一个消息流，每条消息都以 uint32 为前缀，描述消息的长度。（然后消息数据具有进一步的内部结构，但这在这里并不重要，因为我可以在解析之前将整个消息读入内存，因为预期的消息大小很小）。该协议在两个方向上都是相同的，客户端向服务器发送包含请求的消息，而服务器返回包含响应的消息（没有操作并发）。

我的想法是在两个简单的管道之上构建代码Message（我自己的类型描述各种可能的消息），ByteString反之亦然：

import qualified Data.ByteString as B
import qualified Data.ByteString.Lazy as LB

data Message = ...
parseMessage :: LB.ByteString -> Message
serializeMessage :: Message -> LB.ByteString

messageReceiver :: Conduit B.ByteString IO Message
messageReceiver = loop
  where
    loop = do
      lenBytes <- takeCE 4 =$= sinkLazy
      message <- takeCE (runGet getWord32be' lenBytes) =$= sinkLazy
      yield $ parseMessage message
      loop

messageSender :: Conduit Message IO B.ByteString
messageSender = concatMapC $ \message ->
  let messageBytes = serializeMessage message
      lenBytes = runPut $ putWord32be' (LB.length messageBytes)
  in map LB.toStrict [lenBytes, messageBytes]

到目前为止，一切都很好; 或者至少，代码类型检查，尽管我确信有一种更优雅的方式来编写它（尤其是 中的循环messageReceiver）。现在我想写一些东西来连接服务器，发送请求，得到响应，然后断开连接。我写了这个：

runOneCommand request = do
  yield request
  response <- await
  return response

但是，我不确定如何实际将其连接到网络客户端源和接收器，以使我得到“响应”值。我试过这个：

appSource agent $$ messageReceiver =$= runOneCommand =$= messageSender =$= appSink agent

无法编译：

Couldn't match type `Data.Maybe.Maybe SSH.Agent.Message' with `()'
Expected type: conduit-1.2.3.1:Data.Conduit.Internal.Conduit.Conduit
                 SSH.Agent.Message ghc-prim:GHC.Types.IO SSH.Agent.Message
  Actual type: conduit-1.2.3.1:Data.Conduit.Internal.Conduit.ConduitM
                 SSH.Agent.Message
                 SSH.Agent.Message
                 ghc-prim:GHC.Types.IO
                 (Data.Maybe.Maybe SSH.Agent.Message)
In the return type of a call of `Main.runOneCommand'
In the first argument of `(conduit-1.2.3.1:Data.Conduit.Internal.Conduit.=$=)', namely
  `Main.runOneCommand SSH.Agent.RequestIdentities'
In the second argument of `(conduit-1.2.3.1:Data.Conduit.Internal.Conduit.=$=)', namely
  `Main.runOneCommand SSH.Agent.RequestIdentities
   conduit-1.2.3.1:Data.Conduit.Internal.Conduit.=$=
     Main.messageSender
     conduit-1.2.3.1:Data.Conduit.Internal.Conduit.=$=
       Data.Conduit.Network.appSink agent'
Couldn't match type `Data.Maybe.Maybe SSH.Agent.Message' with `()'
Expected type: conduit-1.2.3.1:Data.Conduit.Internal.Conduit.Conduit
                 SSH.Agent.Message ghc-prim:GHC.Types.IO SSH.Agent.Message
  Actual type: conduit-1.2.3.1:Data.Conduit.Internal.Conduit.ConduitM
                 SSH.Agent.Message
                 SSH.Agent.Message
                 ghc-prim:GHC.Types.IO
                 (Data.Maybe.Maybe SSH.Agent.Message)
In the return type of a call of `Main.runOneCommand'
In the first argument of `(conduit-1.2.3.1:Data.Conduit.Internal.Conduit.=$=)', namely
  `Main.runOneCommand SSH.Agent.RequestIdentities'
In the second argument of `(conduit-1.2.3.1:Data.Conduit.Internal.Conduit.=$=)', namely
  `Main.runOneCommand SSH.Agent.RequestIdentities
   conduit-1.2.3.1:Data.Conduit.Internal.Conduit.=$=
     Main.messageSender
     conduit-1.2.3.1:Data.Conduit.Internal.Conduit.=$=
       Data.Conduit.Network.appSink agent'

假设我在这里正确地遵循了类型，这将失败，因为网络客户端的接收器期望返回类型为(), not Message，所以我想我在这里需要一些其他形式的管道组合，但我不知道是什么。

4

1 回答 1

1

发布此问题后，我发现https://stackoverflow.com/a/23925496/31490指出了我的方向fuseUpstream：

  response <- appSource agent $$ messageReceiver =$= runOneCommand RequestIdentities `fuseUpstream` messageSender `fuseUpstream` appSink agent

似乎关于该fuseUpstream答案类型的可怕性的警告不再适用（因为 Conduit 类型被简化了？）；相比：

(=$=) :: Monad m => Conduit a m b -> ConduitM b c m r -> ConduitM a c m r
fuseBoth :: Monad m => ConduitM a b m r1 -> ConduitM b c m r2 -> ConduitM a c m (r1, r2)
fuseUpstream :: Monad m => ConduitM a b m r -> Conduit b m c -> ConduitM a c m r
于 2014-12-28T11:59:49.700 回答