NetMQ訂戶與發佈的消息

Question

我有其產生在結構的PUB插座消息發佈服務器類，用下面的代碼阻斷：

this.context = NetMQContext.Create(); 
this.pubSocket = this.context.CreatePublisherSocket(); 
var portNumber = this.installerSettings.PublisherPort; 
this.pubSocket.Bind("tcp://127.0.0.1:" + portNumber); 

發送使用messagePublishingServer.Publish(message)消息執行：

this.pubSocket.SendMoreFrame(string.Empty).SendFrame(message); 

以下xBehave測試...

[Scenario] 
    public void PublishMessageScenario() 
    { 
     MessagePublishingServer messagePublishingServer = null; 
     NetMQContext context; 
     NetMQ.Sockets.SubscriberSocket subSocket = null; 
     string receivedMessage = null; 

     "Given a running message publishing server"._(() => 
     { 
      var installerSettingsManager = A.Fake<IInstallerSettingsManager>(); 
      var settings = new InstallerSettings { PublisherPort = "5348" }; 
      A.CallTo(() => installerSettingsManager.Settings).Returns(settings); 
      messagePublishingServer = new MessagePublishingServer(installerSettingsManager); 
     }); 

     "And a subscriber connected to the publishing server"._(() => 
     { 
      context = NetMQContext.Create(); 
      subSocket = context.CreateSubscriberSocket(); 
      subSocket.Options.ReceiveHighWatermark = 1000; 
      subSocket.Connect("tcp://127.0.0.1:5348"); 
      subSocket.Subscribe(string.Empty); 
     }); 

     "When publishing a message"._(() => 
     { 
      messagePublishingServer.Publish("test message"); 

      // Receive the topic 
      subSocket.ReceiveFrameString(); 

      // and the message 
      receivedMessage = subSocket.ReceiveFrameString(); 
     }); 

     "Then the subscriber must have received it"._(() => 
     { 
      receivedMessage.Should().NotBeNullOrEmpty(); 
      receivedMessage.Should().Be("test message"); 
     }); 
    } 

... bl在我發現意想不到的第一個subSocket.ReceiveFrameString()。訂閱者套接字是否應該在發佈接收消息之前排隊發佈的消息？

Answer 1

出版商就像收音機，如果沒有連接，當出版商出版你錯過的消息訂閱。我的建議是在用戶連接後進行100ms睡眠（僅用於測試）。

Answer 2

從源（ReceivingSocketExtensions.cs ）：

/// Receive a single frame from socket, blocking until one arrives, and decode as a string using ... 
    public static string ReceiveFrameString([NotNull] this IReceivingSocket socket) 

和

/// If no message is immediately available, return <c>false</c>. 
    public static bool TryReceiveFrameString([NotNull] this IReceivingSocket socket, out string frameString)