从带有管道的流网络协议获取结果

Question

我正在用管道实现一个简单的网络协议;该协议是一个消息流，每条消息都以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'

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

Answer 1

发布这个问题后，我发现了 https://stackoverflow.com/a/23925496/31490，它为我指明了fuseUpstream的方向：

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

似乎该答案中关于fuseUpstream类型可怕的警告不再适用（因为管道类型被简化了？）;比较：

(=$=) :: 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