haskell - 在 Pipes-2.1.0 包中完成

Question

我正在使用 Pipes-2.1.0 包和 zeromq3-haskell 包来构建一个小消息管道。一切似乎都很顺利，只是我无法理解 Frames 的最终确定。

在下面的 Frame 中，我获取了两个资源；一个 zeromq 上下文和一个 zeromq 套接字。然后我不断等待消息(以 ByteStrings 的形式)在 zeromq 套接字上发布。

{-# LANGUAGE RebindableSyntax    #-}
{-# LANGUAGE ScopedTypeVariables #-}

module PipesZeroMQ where

import           Control.Frame
import           Control.IMonad.Do
import           Control.IMonad.Trans
import qualified Control.Monad          as M
import           Data.ByteString        (ByteString)
import           Data.String
import           Prelude                hiding (Monad(..))
import qualified System.ZMQ3            as ZMQ

type Address = String

fromList :: (M.Monad m) => [b] -> Frame b m (M a) (M a) ()
fromList xs = mapMR_ yield xs

publisher :: Address -> Frame Void IO (M ByteString) C ()
publisher addr = do
  c  <- liftU $ ZMQ.init 1
  s  <-liftU $ ZMQ.socket c ZMQ.Pub
  liftU $ ZMQ.bind s addr   
  liftU $ print "Socket open for business!!!"

  foreverR $ do
    bs <- await
    finallyF (ZMQ.close s M.>> ZMQ.term c M.>> print "ZMQ socket closed") $ do
         (liftU $ ZMQ.send s [] bs)
         (liftU (print "Sending message"))

现在，如果我尝试这个:

λ> runFrame $ (publisher localAddress) <-< (fromList ["This", "that", "that"] >> close)

我明白了:

"Socket open for business"
"Sending message"
"ZMQ socket closed"
*** Exception: ZMQError { errno = 88, source = "send", message = "Socket operation on non-socket" }

publisher收到后敲定，但一个 BytesString .

为什么会这样？

我对在 Pipes-2.1.0 中使用 Frames 进行定稿有什么误解？

如果我开始攻击它，外面的树有机会吗？

Answer 1

您在写 publisher 时犯了一个错误功能:

foreverR $ do
    bs <- await
    finallyF (ZMQ.close s M.>> ZMQ.term c M.>> print "ZMQ socket closed") $ do
         (liftU $ ZMQ.send s [] bs)
         (liftU (print "Sending message"))

您可能想要放置 finallyF外 foreverR循环:

finallyF (...) $ foreverR $ do
    bs <- await
    liftU $ ZMQ.send s [] bs)
    liftU (print "Sending message")

你写它的方式，它在每​​次发送后完成，所以它完全按照你的要求去做:在每次发送后完成。 finallyF一旦它包装的操作完成，则调用终结器，无论它是否成功终止。您也可以使用 catchF在这种情况下，因为循环永远不会终止:

 catchF (...) $ foreverR $ do
    bs <- await
    liftU $ ZMQ.send s [] bs)
    liftU (print "Sending message")

或者，您可以将其保留在循环内，但切换到 catchF这样终结器就不会在每次发送后运行:

foreverR $ do
    bs <- await
    catchF (ZMQ.close s M.>> ZMQ.term c M.>> print "ZMQ socket closed") $ do
         (liftU $ ZMQ.send s [] bs)
         (liftU (print "Sending message"))

另外，如果你打算写一个基于管道的 zeroMQ 库，请与我保持联系，因为我计划在下一个版本中将帧返回到一个普通的 monad，还有很多新的功能增强，例如关闭和重新初始化资源的能力。要联系我，请使用我的 gmail.com 地址和用户名 Gabriel439。