NetMQ.FiniteStateMachineException: Rep.XRecv - cannot receive another request

Question

I constantly get NetMQ.FiniteStateMachineException

sure, my code works... the exception does not occur right away... but over the course of a few hours, probably it will happen.

Does anyone know what is going on here to cause this exception?

I'm not really sure why, even though I did read the explanation here: c# - ZeroMQ FiniteStateMachineException in REQ/REP pattern - Stack Overflow

I get a bunch of these:

the exception

NetMQ.FiniteStateMachineException: Rep.XRecv - cannot receive another request
   at NetMQ.Core.Patterns.Rep.XRecv(Msg& msg)
   at NetMQ.Core.SocketBase.TryRecv(Msg& msg, TimeSpan timeout)
   at NetMQ.NetMQSocket.TryReceive(Msg& msg, TimeSpan timeout)
   at NetMQ.ReceivingSocketExtensions.ReceiveFrameString(IReceivingSocket socket, Encoding encoding, Boolean& more)
   at NinjaTrader.NinjaScript.AddOns.anAddOn.ZeroMQ_Server()
NetMQ.FiniteStateMachineException: Rep.XRecv - cannot receive another request
   at NetMQ.Core.Patterns.Rep.XRecv(Msg& msg)
   at NetMQ.Core.SocketBase.TryRecv(Msg& msg, TimeSpan timeout)
   at NetMQ.NetMQSocket.TryReceive(Msg& msg, TimeSpan timeout)
   at NetMQ.ReceivingSocketExtensions.ReceiveFrameString(IReceivingSocket socket, Encoding encoding, Boolean& more)
   at NinjaTrader.NinjaScript.AddOns.anAddOn.ZeroMQ_Server()

my code

        // thread start code

                if (thread == null) {
                    print2("Addon {0}: is starting, listening on port: {1}...", GetType().Name, ZeroPort);
                    thread = new Thread(ZeroMQ_Server);
                    thread.Start();
                }

        // zeroMQ code

        #region TaskCallBack - NetMQ
        // This thread procedure performs the task.
        private void ZeroMQ_Server()
        { 
            bool quit = false;
            string bindAddress = "tcp://*:"+ZeroPort;
            try {
                while (!quit) {
                    try {
                        using (var repSocket = new ResponseSocket()) 
                        {
                            curRepSocket = repSocket;
                            print2("*** BINDING on {0} ***", bindAddress);
                            repSocket.Bind(bindAddress); 

                            while (!quit) {
                                try {
                                    Running = true;

                                    var msgStr = repSocket.ReceiveFrameString();
                                    print2("[►] {2} [REP:{0}] {1}", bindAddress, msgStr, DateTime.Now.ToString("HH:mm:ss.fff"));
                                    if (processMsg(msgStr)) {
                                        StringBuilder csv = new StringBuilder();
                                        // string building stuff here

                                        string cs = csv.ToString();
                                        print2("[◄] csv: {0}", cs);
                                        repSocket.SendFrame(cs);
                                    } else {
                                        repSocket.SendFrame("Unrecognized Command: " + msgStr);
                                        break;
                                    }

                                } catch (Exception e) {
                                    quit = isThreadAborted(e);
                                }
                            }
                        }

                    } catch (Exception e) {
                        if (e is AddressAlreadyInUseException) {
                            //print2("");
                        } else quit = isThreadAborted(e);
                    } finally {
                        curRepSocket = null;
                        Running = false;
                    }
                }
            } finally {
                //NetMQConfig.Cleanup(); 
            }
        }

        private bool isThreadAborted(Exception e) {
            if (e is ThreadAbortException) {
                print2("\n*** thread aborting... ***");
                return true;
            } else {
                print2(e);
                return false;
            }
        }

Answer 1

Response socket is a state machine, you must reply to each request. From the code, it seems that if processMsg throws you don't send anything back, therefore you cannot receive again and get the exception.

It can also if Send failed if the client is gone.

Try to use router instead, like so:

while (true)
{
    bool more;
    var msg = routerSocket.ReceiveFrameBytes(out more);

    // Forwarding the routing id.
    routerSocket.SendMoreFrame(msg);

    // Bottom, next frame is the message
    if (msg.Length == 0)
        break;
}

// Write your handling here