SwiftyZeroMQ Publisher 在发布第一条消息后无法识别订阅者

Question

问题：一旦发布者发布了第一条消息，我订阅我的发布者就失败了。

目标：从我的 swift 应用程序通过 ZeroMQ 发布数据流。然后连接和断开几个订阅者并获取消息。

背景：我使用 swift5 和 SwiftyZeroMQ5（我也测试过 SwiftyZeroMQ），我部署在 iPhone 上。我尝试同时订阅 swift 和 python3。只有当我在发布第一条消息之前连接我的订阅者时，它才有效。如果我首先连接我的订阅者然后启动发布者应用程序，然后发布，它也可以工作。 python3上对应的发布和订阅代码不需要以任何特定的顺序启动，代表了我想要的行为。 因为如果我以特定顺序开始，我可以让 sub/pub 工作，我知道 IP 号、端口和主题、格式等是正确的。 请注意，当我同时订阅 python3 和 swift 时，行为是相同的 - 这不是 swift 和 python 之间的兼容性问题。

错误消息：没有错误消息，如果进程没有按照描述的顺序启动，轮询器就不会触发。如果进程按所述顺序启动，轮询器会触发并接收消息。

我尝试了什么：我尝试了关于代码库和设备的发布者和订阅者的不同组合。

• 在同一设备上 Swift pub 到 swift sub [仅按所述顺序工作]
• 在不同设备上 Swift pub 到 swift sub [仅按所述顺序工作]
• Swift pub 到 python3 sub [仅按所述顺序工作]
• Python3 pub 到 swift sub [独立于启动顺序工作]
• Python3 pub 到 python3 sub [独立于启动顺序工作]

我的结论： swift 发布者套接字存在问题：它在发布第一条消息后无法识别新订阅者。

发布者的 Swift 代码，在 viewDidLoad() 中调用 initPublisher。 ZeroMQ 库版本为 4.2.2：

import SwiftyZeroMQ5
var context: SwiftyZeroMQ.Context = try! SwiftyZeroMQ.Context()
var gpsPublisher: SwiftyZeroMQ.Socket?
let gpsPublishEndPoint = "tcp://*:5560"
// Init the publisher socket
func initPublisher()->Bool{
    do{
        self.gpsPublisher = try context.socket(.publish)
        try self.gpsPublisher?.setSendBufferSize(4096)
        try self.gpsPublisher?.setLinger(0)                  // Dont buffer messages
        try self.gpsPublisher?.bind(self.gpsPublishEndPoint)
        return true
        }
    catch{
        print("Publish setup failed!")
        return false
    }
}
// ZMQ publish. Publishes string and serialized json-object
func publish(socket: SwiftyZeroMQ.Socket?, topic: String, json: JSON)->Bool{
    // Create string with topic and json representation
    let publishStr = getJsonStringAndTopic(topic: topic, json: json)
    do{
        try socket?.send(string: publishStr)
        print("publish: Published: " + publishStr)
        return true
    }
    catch{
        print("publish: Error, tried to publish, but failed: " + publishStr)
        return false
    }
}
//The function is repeatedly called in a thread. Only function call shown here below.
_ = self.publish(socket: self.gpsPublisher, topic: "myTopic", json: json)

python3 订阅者代码， zmq.zmq_version() -> '4.3.2':

context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect("tcp://192.168.1.2:5560")
socket.setsockopt_string(zmq.SUBSCRIBE, 'myTopic')
socket.RCVTIMEO = 1000  # in milliseconds
while socket:
  try:
    msg = str(socket.recv(), 'utf-8')
    (topic, msg) = auxiliaries.zmq.demogrify(msg)
    _print((topic, msg))
  except zmq.error.Again as error:
    _print(str(error))
  except KeyboardInterrupt:
    auxiliaries.zmq.close_socket_gracefully(socket)
    socket = None

非常感谢任何帮助、有趣的测试设置等。

Answer 1

我做了更多测试，发现了一些事情和解决方法。

1. 代码在 iPhone 模拟器上运行时按预期工作（模拟器是 x86_64 架构，iPhone8 是 armv7）
2. 我不认为它是相关的，但我确实觉得它很有趣。某些多播和广播协议需要获得 Apple 的批准。您可以在未经批准的情况下在模拟器上运行，但不能在设备上运行。 Apple networking multicast entitlement。由于它部分有效，我确实排除了这一点。
3. 解决方法是在每次发布之前再次绑定套接字。这会引发错误“地址已在使用中”，但似乎并没有造成太大的伤害。
4. 在发布消息之间没有不合理的延迟，如果从 iPhone 运行时未再次绑定发布套接字，则 pub-sub 会在 500-1000 条消息后失败。

我制作了一个最小的工作示例应用程序，如果您愿意或需要更深入地挖掘，您可以使用它。它有“init”、“bind”、“publish”和“bind and publish”按钮。您可以发送一批消息并检查时间等。 我将应用程序运行到 python3 脚本。 我通过 cocoapods 包含了 SwiftyZeroMQ5。

Pod 文件：

platform :ios, '13.0'
target 'ZMQtest' do
  use_frameworks!
  pod 'SwiftyZeroMQ5'
end

SwiftCode（您自己设置按钮..）

import UIKit
import SwiftyZeroMQ5
class ViewController: UIViewController {
    let context = try! SwiftyZeroMQ.Context()
    var publisher: SwiftyZeroMQ.Socket?
    let publishEndPoint = "tcp://*:5560"
    var cnt = 0
    let quota = 1200
    @IBAction func publishButtonPressed(_ sender: Any) {
        while cnt < quota {
            publish()
            //usleep(10000)
        }
        cnt = 1
    }
    @IBAction func bindAndPublishButtonPressed(_ sender: Any) {
        while cnt < quota {
            bindPublisher()
            publish()
        }
        cnt = 1
    }
    @IBAction func initPubButtonPressed(_ sender: Any) {
        initPublisher()
    }
    @IBAction func bindPublisherButtonPressed(_ sender: Any) {
        bindPublisher()
    }
    @IBOutlet weak var statusLabel: UILabel!

    // **************
    // Init publisher
    func initPublisher(){
        do {
            self.publisher = try context.socket(.publish)
            print("Publisher socket created")
        }
        catch {
            print("initPublisher error: ", error)
        }
     }

    // **************
    // Bind publisher
    func bindPublisher(){
        do {
            try self.publisher?.bind(publishEndPoint)
            print("Publisher socket binded to :", publishEndPoint)
        }
        catch {
            print("bindPublisher error: ", error)
        }
    }

    // *****************
    // Publish a message
    func publish(){
        // Publish dummy string
        do{
            cnt += 1
            let str = "topic {\"key\": \"" + String(cnt) + "\"}"
            try self.publisher?.send(string: str)
            statusLabel.text = str
            print("Publish message no: ", String(cnt))
        }
        catch{
            print("publisher error: ", error)
        }
    }

    override func viewDidLoad() {
        super.viewDidLoad()
    }
}

还有python3代码：

#!/usr/bin/env python3
'''Minimal running example of a ZMQ SUB socket.'''
import json
import zmq

def demogrify(msg: str):
  '''inverse of mogrify()'''
  try:
    (topic, message) = msg.split(maxsplit=1)
  except ValueError:
    (topic, message) = (msg, '{}')

  return topic, json.loads(message)

def close_socket_gracefully(socket):
  '''graceful termination'''
  socket.setsockopt(zmq.LINGER, 0) # to avoid hanging infinitely
  socket.close()

if __name__ == "__main__":
  context = zmq.Context()

  socket = context.socket(zmq.SUB)  #pylint: disable=no-member
  socket.connect("tcp://192.168.1.2:5560")
  socket.setsockopt_string(zmq.SUBSCRIBE, '')  #pylint: disable=no-member
  socket.RCVTIMEO = 1000  # in milliseconds

  while socket:
    try:
      msg = str(socket.recv(), 'utf-8')
      (topic, msg) = demogrify(msg)
      print((topic, msg))
    except zmq.error.Again as error:
      print(str(error))
    except KeyboardInterrupt:
      close_socket_gracefully(socket)
      socket = None

我应该将此问题标记为已解决吗？