NetMQ, почему для Req-Rep нужен "SendReady"?
У меня есть проблема, которую мне удалось исправить... Однако я немного обеспокоен, так как не очень понимаю, почему решение сработало;
Я использую NetMQ и, в частности, модуль NetMQ, который имеет несколько сокетов, один из которых - пара REQ-REP.
У меня есть очередь запросов, которые преобразуются в запросы, и сервер обрабатывает каждый тип запроса по мере необходимости и отправляет соответствующий ответ. Это работало без проблем, однако, когда я попытался добавить дополнительный тип запроса, система перестала работать, как ожидалось; что произойдет, если запрос достигнет сервера, сервер отправит ответ... и клиент не получит его. Сообщение не будет получено на клиенте, пока сервер не будет выключен (необычное поведение!).
Я управлял парой REQ-REP с флагом, который я установил перед отправкой запроса, и сбрасывал при получении ответа. Мне удалось решить проблему, вызвав только ответы в событии "SendReady" сокета REQ - это автоматически исправило все мои проблемы, однако я не могу найти ничего в документации, которая говорит мне, почему сокет, возможно, не был в состоянии "sendready", или что это на самом деле делает.
Любая информация, которая могла бы пролиться на то, почему это работает теперь, была бы велика:)
Приветствия.
Редактировать: Источник
Клиент:
"Подписка" запускается как отдельный поток для пользовательского интерфейса
private void Subscribe(string address)
{
using (var req = new RequestSocket(address + ":5555"))
using (var sub = new SubscriberSocket(address + ":5556"))
using (var poller = new NetMQPoller { req, sub })
{
// Send program code when a request for a code update is received
sub.ReceiveReady += (s, a) =>
{
var type = sub.ReceiveFrameString();
var reply = sub.ReceiveFrameString();
switch (type)
{
case "Type1":
manager.ChangeValue(reply);
break;
case "Type2":
string[] args = reply.Split(',');
eventAggregator.PublishOnUIThread(new MyEvent(args[0], (SimObjectActionEventType)Enum.Parse(typeof(MyEventType), args[1])));
break;
}
};
req.ReceiveReady += Req_ReceiveReady;
poller.RunAsync();
sub.Connect(address + ":5556");
sub.SubscribeToAnyTopic();
sub.Options.ReceiveHighWatermark = 10;
reqQueue = new Queue<string[]>();
reqQueue.Enqueue(new string[] { "InitialiseClient", "" });
req_sending = false;
while (programRunning)
{
if (reqQueue.Count > 0 && !req_sending)
{
req_sending = true;
string[] request = reqQueue.Dequeue();
Console.WriteLine("Sending " + request[0] + " " + request[1]);
req.SendMoreFrame(request[0]).SendFrame(request[1]);
}
Thread.Sleep(1);
}
}
}
private void Req_ReceiveReady(object sender, NetMQSocketEventArgs e)
{
var req = e.Socket;
var messageType = req.ReceiveFrameString();
Console.WriteLine("Received {0}", messageType);
switch (messageType)
{
case "Reply1":
// Receive action
break;
case "Reply2":
// Receive action
break;
case "Reply3":
// Receive action
break;
}
req_sending = false;
}
Сервер:
using (var rep = new ResponseSocket("@tcp://*:5555"))
using (var pub = new PublisherSocket("@tcp://*:5556"))
using (var beacon = new NetMQBeacon())
using (var poller = new NetMQPoller { rep, pub, beacon })
{
// Send program code when a request for a code update is received
rep.ReceiveReady += (s, a) =>
{
var messageType = rep.ReceiveFrameString();
var message = rep.ReceiveFrameString();
Console.WriteLine("Received {0} - Content: {1}", messageType, message);
switch (messageType)
{
case "InitialiseClient":
// Send
rep.SendMoreFrame("Reply1").SendFrame(repData);
break;
case "Req2":
// do something
rep.SendMoreFrame("Reply2").SendFrame("RequestOK");
break;
case "Req3":
args = message.Split(',');
if (args.Length == 2)
{
// Do Something
rep.SendMoreFrame("Reply3").SendFrame("RequestOK");
}
else
{
rep.SendMoreFrame("Ack").SendFrame("RequestError - incorrect argument format");
}
break;
case "Req4":
args = message.Split(',');
if (args.Length == 2)
{
requestData = //do something
rep.SendMoreFrame("Reply4").SendFrame(requestData);
}
else
{
rep.SendMoreFrame("Ack").SendFrame("RequestError - incorrect argument format");
}
break;
default:
rep.SendMoreFrame("Ack").SendFrame("Error");
break;
}
};
// setup discovery beacon with 1 second interval
beacon.Configure(5555);
beacon.Publish("server", TimeSpan.FromSeconds(1));
// start the poller
poller.RunAsync();
// run the simulation loop
while (serverRunning)
{
// todo - make this operate for efficiently
// push updated variable values to clients
foreach (string[] message in pubQueue)
{
pub.SendMoreFrame(message[0]).SendFrame(message[1]);
}
pubQueue.Clear();
Thread.Sleep(2);
}
poller.StopAsync();
}
2 ответа
Вы используете сокет запроса из нескольких потоков, что не поддерживается. Вы отправляете в основной поток, а получаете в потоке опросов.
Вместо того, чтобы использовать обычную очередь, попробуйте использовать NetMQQueue, вы можете добавить его в опросник и поставить в очередь из потока пользовательского интерфейса. Затем отправка происходит как в потоке, так и в получателе.
Вы можете прочитать документы здесь: http://netmq.readthedocs.io/en/latest/queue/
Единственное, о чем я могу думать, это то, что сокет REP готов к отправке только после того, как вы действительно получили сообщение полностью (все части). Есть ли шанс, что вы пытаетесь отправить без предварительного запроса?
Можете поделиться источником?