CODEWITHSUNDEEP
.netzeromqnetmq
iiwaasnet
iiwaasnet

Reputation: 51

Messages are not delivered after reconnect over ROUTER-ROUTER in NetMQ

I have a problem with ROUTER-ROUTER communication with NetMQ.

Two applications, Client and Server. Both using ROUTER sockets. Client ROUTER socket is not setting it's identity explicitly, only Server.

After the connection established first time, messages are routed from client (ROUTER) to server (ROUTER). But as soon as client calls Disconnect() and then Connect() again (after a very short time, i.e. 10 msec), the server is not receiving messages any more.

What is interesting, that reconnection works fine if ZeroMq C# lib is used.

Is it a known issue or I'm doing something wrong here?

Minimal code to reproduce the issue:

  public class PureNetMQ
  {
    private static readonly byte[] ServerIdentity = {1, 2, 3};
    private static readonly byte[] ClientIdentity = {1, 2, 4};
    private static readonly string Address = "tcp://127.0.0.1:5000";

    private static void Main(string[] args)
    {
      var cancellationTokenSource = new CancellationTokenSource();
      Task.Factory.StartNew(_ => RunServer(cancellationTokenSource.Token), cancellationTokenSource.Token, TaskCreationOptions.LongRunning);
      Thread.Sleep(TimeSpan.FromSeconds(2));

      RunClient();

      Console.WriteLine("Done");
      Console.ReadLine();
      cancellationTokenSource.Cancel(true);
    }

    private static void RunClient()
    {
      using (var context = NetMQContext.Create())
      {
        using (NetMQSocket socket = context.CreateRouterSocket())
        {
          //socket.Options.Identity = Guid.NewGuid().ToByteArray();
          //socket.Options.RouterMandatory = true;
          socket.Options.Linger = TimeSpan.Zero;

          socket.Connect(Address);

          Thread.Sleep(TimeSpan.FromSeconds(2));

          socket.SendMore(ServerIdentity);
          socket.SendMore(Encoding.UTF8.GetBytes(""));
          socket.Send(Encoding.UTF8.GetBytes(Guid.NewGuid().ToString()));
          Console.WriteLine("Message sent...");

          socket.Disconnect(Address);
          Thread.Sleep(TimeSpan.FromSeconds(5));
          Console.WriteLine("Disconnected");

          //socket.Options.Identity = Guid.NewGuid().ToByteArray();
          socket.Connect(Address);
          Thread.Sleep(TimeSpan.FromSeconds(5));
          Console.WriteLine("Reconnected");


          socket.SendMore(ServerIdentity);
          socket.SendMore(Encoding.UTF8.GetBytes(""));
          socket.Send(Encoding.UTF8.GetBytes(Guid.NewGuid().ToString()));
          Console.WriteLine("Message sent...");

          Thread.Sleep(TimeSpan.FromSeconds(5));
        }
      }
    }

    private static void RunServer(CancellationToken token)
    {
      using (var context = NetMQContext.Create())
      {
        using (var socket = context.CreateRouterSocket())
        {
          socket.Options.Identity = ServerIdentity;
          //socket.Options.RouterMandatory = true;
          socket.Options.Linger = TimeSpan.Zero;

          socket.Bind(Address);

          while (!token.IsCancellationRequested)
          {
            var message = socket.ReceiveMessage();

            for (var i = 2; i < message.FrameCount; i++)
            {
              Console.WriteLine("Message: {0}", Encoding.UTF8.GetString(message[i].Buffer));
            }
          }
        }
      }
    }
  }

Upvotes: 2

Views: 1194

Answers (1)

somdoron
somdoron

Reputation: 4842

Zeromq had a fix for this that was not ported to NetMQ yet. I ported it now, you can use it from:

https://github.com/somdoron/netmq

It soon should be merged to master soon: https://github.com/zeromq/netmq/pull/373

Also another issue that is relevant and was solved in ZeroMQ is the handling of duplicate identities in router. by default the second socket with the identity is ignored. ZeroMQ have the handover option, if enabled the new socket take over the old one. You should consider using this or port it to netmq as well.

https://github.com/zeromq/libzmq/pull/729/files https://github.com/zeromq/libzmq/pull/731

Upvotes: 4

Related Questions