如何在 Haskell 中通过管道使用 ZMQ 订阅者答案

【问题标题】：How to use a ZMQ subscriber in Haskell with pipes如何在 Haskell 中通过管道使用 ZMQ 订阅者
【发布时间】：2018-10-26 20:02:11
【问题描述】：

我可以让 ZMQ 订阅者在 Haskell 中工作，但希望获得有关如何将这些数据与 Pipes 一起使用的指导。我编写生产者的尝试在“堆栈构建”时失败，并出现以下错误：

无法将类型“代理 X () c'0 c0 (ZMQ z)”与“ZMQ z”匹配

预期类型：ZMQ z ()

实际类型：Proxy X () c'0 c0 (ZMQ z) ()

{-# LANGUAGE OverloadedStrings #-}
module Main where
import Control.Monad
import Pipes
import qualified Pipes.Prelude as P
import System.ZMQ4.Monadic
import qualified Data.ByteString.Char8 as CS

fromZMQ :: (Receiver r) => Socket z r -> Producer String (ZMQ z) ()
fromZMQ sock = do
    msg <- lift $ receive sock
    yield (CS.unpack msg)
    fromZMQ sock

main :: IO ()
main = --do
  runZMQ $ do
    subSock <- socket Sub  ---subscriptionSocket
    subscribe subSock ""
    connect subSock "tcp://127.0.0.1:4998" 
    forever $ fromZMQ subSock >-> P.take 3 >-> P.print

注意，我想使用 Python 脚本在 ZMQ 上发布的数据。

【问题讨论】：

我建议不要使用 monadic zmq API 并在需要时滚动你自己的 Monad。或者考虑通过 Chan 和工作线程间接到 zmq，该线程只是将数据传输到 zmq 或从 zmq 传输数据。
感谢@Thomas 提出调查陈的建议。因为我认为 zeroMQ 提供了 Chan 提供的缓冲，所以我能够使用 MVar 运行一些东西，这似乎巧妙地解决了我正在努力解决的潜在问题，即。如何在不出现 > 运行时错误的情况下更新状态表示。下面的代码，以防它帮助别人。

标签： haskell zeromq haskell-pipes

【解决方案1】：

根据 Thomas 关于使用 Chan 的建议，我修改了 MVar 示例（下面的链接）来累积接收到的字符串并对其进行计数以演示更新和读取状态

https://www.oreilly.com/library/view/parallel-and-concurrent/9781449335939/ch07.html

{-# LANGUAGE OverloadedStrings #-}

module Main where

import Control.Concurrent  
import Control.Monad
import System.ZMQ4.Monadic
import qualified Data.ByteString.Char8 as CS

newtype State = State (MVar (Int, [CS.ByteString]) ) --(count, list of strings received over zmq)

newState :: IO State
newState = do
  m <- newMVar (0, [])
  return (State m)

updateState :: State -> CS.ByteString -> IO ()
updateState (State m) newString = do
  (count,strList) <- takeMVar m
  putMVar m ( count + 1 , strList ++ [newString] )

showState :: State -> IO String
showState (State m) = do
  count <- takeMVar m
  putMVar m count  --return the lock; no changes
  return (show count)

main = runZMQ $ do
    sub <- socket Sub
    subscribe sub ""
    connect sub "tcp://127.0.0.1:4998"
    s <- liftIO newState 
    forever $ do
      receive sub >>= liftIO . updateState s 
      liftIO $ updateState s "hello"  --'manually' add an additional string on each iteration
      op <- liftIO $ showState s
      liftIO $ print op

【讨论】：

【解决方案2】：

您的代码中唯一的问题是最后一行。

你有一个“管道”：

fromZMQ subSock >-> P.take 3 >-> P.print :: Effect (ZMQ z) ()

当你对它应用forever 时，类型保持不变。但是您需要一个简单的ZMQ z，换句话说，您需要实际执行pipes 计算并为此使用runEffect 函数。附带说明一下，您实际上并不需要 forever，因为流永远不会结束。

所以，您真正要做的就是将最后一行中的 forever 替换为 runEffect。

【讨论】：