5
\$\begingroup\$

I'm a zeromq newb and have implemented the REQ/RES Lazy Pirate pattern with C#. It's essentially a port from the C example given in the documentation. It's harnessed by tests, and seems to be work to my expectations. I'm interested to know if I'm understanding the pattern correctly, or if I can improve the API. If anyone knows of a pre-existing solution, I'd love to know too. Any advice would be great.

using System;
using System.Text;
using ZMQ;

namespace zeromq.net {
    public class LazyPirateClient : ReqRes {
        private readonly string _address;
        private readonly long _timeOut;
        private readonly int _retries;
        private bool _responseReceived;
        private int _attempt;

        public LazyPirateClient(string address, long timeOut, int retries) {
            _address = address;
            _timeOut = timeOut;
            _retries = retries;            
        }                                  

        public override void Send(Func<byte[]> byteProvider) {
            _responseReceived = false;
            using (var context = new Context(1)) {
                _attempt = 0;
                do {
                    using (var requester = context.Socket(SocketType.REQ)) {
                        _attempt++;
                        requester.Connect(_address);
                        requester.Send(byteProvider(), SendRecvOpt.NOBLOCK);
                        var item = requester.CreatePollItem(IOMultiPlex.POLLIN);
                        item.PollInHandler += item_PollInHandler;
                        context.Poll(new[] { item }, _timeOut);
                        requester.Linger = 0;
                    }
                } while (_attempt < _retries && !_responseReceived);

                if (!_responseReceived)
                     throw new PermenantFailException(_attempt);
            }
        }

        private void item_PollInHandler(Socket socket, IOMultiPlex revents) {
            _responseReceived = true;
            var bytes = socket.Recv(SendRecvOpt.NOBLOCK);
            OnReceived(new ResponseReceivedEventArgs(bytes, _attempt));
        }                
    }


     public abstract class ReqRes {
        public abstract void Send(Func<byte[]> byteProvider);
        public delegate void ResponseReceived(object sender, ResponseReceivedEventArgs e);
        public event ResponseReceived ReceivedHandler;

        public void OnReceived(ResponseReceivedEventArgs e) {
            var handler = ReceivedHandler;
            if (handler != null) handler(this, e);
        }
     }
\$\endgroup\$
1
  • \$\begingroup\$ Minor improvement, in c# class properties use PascalCase. So you would have Address, Timeout, ResponseReceived, etc. You also have a typo in PermanentFailException. \$\endgroup\$ – ANeves thinks SE is evil Feb 6 '12 at 13:59
2
+50
\$\begingroup\$

It seems zeromq will release sent data itself. With GC supported in C#, this feature is useless. The byteProvider parameter seems inappropriate, so why not just use IEnumerable<byte>?

\$\endgroup\$
3
  • \$\begingroup\$ The motivation for a byteProvider is if you need to regenerate a message on each attempt. For example, if you need to regenerate a timestamp, or id in the message. There is a version (not shown here) that has an overload for IEnumerable<byte> such that: public override void Send(IEnumerable<byte> bytes) { Send(() => bytes); } Using the byteProvider is not the vanilla case, but is there in case you need it. \$\endgroup\$ – stantona Feb 7 '12 at 16:22
  • \$\begingroup\$ The send operation is asynchronous in zeromq, so when send() return, it doesn't mean the data has been sent. I didn't find asynchronous action in your send() method, do you provide another SendAsync()? And if the operation is asynchronous, then, use func<byte[]> as parameter is not proper - the send action will be executed on another thread any other time, if you invoke the func<byte[]> that time, you should have to guarantee the resources referenced in func<byte[]> are still available. \$\endgroup\$ – Simon. Li Feb 8 '12 at 3:43
  • \$\begingroup\$ One thing I have met just like what you done, and in the delegate I referenced an autowaitevent, the program crashed sometime - The autowaitevent might be disposed when I call it in asynchronously delegate. In order to enhance the stability of program, I suggest IEnumerable<byte>. \$\endgroup\$ – Simon. Li Feb 8 '12 at 3:43

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.