haskell - Pipes-2.1.0 包中的最终确定

Question

我正在使用 Pipes-2.1.0 包和 Zeromq3-haskell 包来构建一个小型消息管道。一切似乎都很顺利，除了我无法理解框架的最终确定。

在下面的框架中我获取了两个资源；一个 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 中使用框架的终结有什么误解？

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

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。