netmq / samples Goto Github PK

private void ProcessReceivedMessageAsync(object sender, NetMQSocketEventArgs e)
        {
            try
            {
                var msg = e.Socket.ReceiveMultipartMessage();
                Debug.WriteLine($"Received: {msg}");
                var senderFrame = msg.Pop();               // [e][protocol header][service or command][data]
                var empty = msg.Pop();                     // [protocol header][service or command][data]
                var headerFrame = msg.Pop();               // [service or command][data]
                var header = headerFrame.ConvertToString();

                switch (header)
                {
                    case MDPConstants.MDP_CLIENT_HEADER:
                        {
                            Task.Factory.StartNew(() => ProcessClientMessage(senderFrame, msg), TaskCreationOptions.LongRunning);
                        }
                        break;
                    case MDPConstants.MDP_WORKER_HEADER:
                        ProcessWorkerMessage(senderFrame, msg);
                        break;
                    default:
                        Debug.WriteLine("ERROR - message with invalid protocol header!");
                        break;
                }
            }
            catch (Exception ex)
            {

                Logger.Fatal(ex.Message, ex);
            }
        }

public MDPClientAsync([NotNull] string brokerAddress, string identity)
{
    if (string.IsNullOrWhiteSpace(brokerAddress))
        throw new ArgumentNullException(nameof(brokerAddress), "The broker address must not be null, empty or whitespace!");

    if (!string.IsNullOrWhiteSpace(identity))
        m_identity = Encoding.UTF8.GetBytes(identity);

    m_brokerAddress = brokerAddress;
}

    public object Deserialize(byte[] data, Type targetType)
    {
        try
        {
            var json = Base64Decode(_encoding.GetString(data));
            return JsonConvert.DeserializeObject(json, targetType);
        }
        catch (Exception e)
        {
            throw new FormatException($"Deserialization of :[{_encoding.GetString(data)}] failed", e);
        }
    }

public class PubSubMessageRouter : IPubSubMessageRouter
{
    private Proxy _proxy;
    private NetMQPoller _poller;
    private XSubscriberSocket _xSub;
    private XPublisherSocket _xPub;
    private PushSocket _controlInPush = new PushSocket();
    private PushSocket _controlOutPush = new PushSocket();
    private PullSocket _controlInPull = new PullSocket();
    private PullSocket _controlOutPull = new PullSocket();
    private Task _workerReceived;
    private Task _workerSent;
    private CancellationTokenSource _cancellationTokenSource;
    private CancellationToken _cancellationToken;
    private WeakAction<MessageBase> _subscriberReceived;
    private WeakAction<MessageBase> _subscriberSent;
    private WeakAction<Exception> _onExceptionHandler;
    private ISerializer _serializer;
    private bool _disposed;

    public string FrontendConnectionString { get; }

    public string BackendConnectionString { get; }

    public PubSubMessageRouter(
        string backendConnectionString,
        string frontendConnectionString,
        string controlInConnectionName = "controlIn",
        string controlOutConnectionName = "controlOut")
    {
        BackendConnectionString = backendConnectionString;
        FrontendConnectionString = frontendConnectionString;
        _serializer = new Serialize.Json.JsonObjectSerializer();
        _xSub = new XSubscriberSocket($"@{FrontendConnectionString}");
        _xPub = new XPublisherSocket($"@{BackendConnectionString}");
        _xSub.Options.ReceiveHighWatermark = 10000;
        _xSub.Options.ReceiveBuffer = 44000;
        _xSub.Options.SendBuffer = 44000;
        _xSub.Options.Linger = TimeSpan.FromSeconds(5);
        _xPub.Options.ReceiveHighWatermark = 10000;
        _xPub.Options.ReceiveBuffer = 44000;
        _xPub.Options.SendBuffer = 44000;
        _xPub.Options.Linger = TimeSpan.FromSeconds(5);
        _controlInPush.Bind($"inproc://{controlInConnectionName}");
        _controlInPull.Connect($"inproc://{controlInConnectionName}");
        _controlOutPush.Bind($"inproc://{controlOutConnectionName}");
        _controlOutPull.Connect($"inproc://{controlOutConnectionName}");
        _poller = new NetMQPoller
        {
            _xSub,
            _xPub
        };
        // proxy messages between frontend / backend
        _proxy = new Proxy(
            _xSub,
            _xPub,
            poller: _poller,
            controlIn: _controlInPush,
            controlOut: _controlOutPush
        );
    }

    public PubSubMessageRouter(
        string backendConnectionString,
        string frontendConnectionString,
        Action<MessageBase> handlerReceived,
        Action<MessageBase> handlerSent,
        Action<Exception> onException=null) : this(backendConnectionString, frontendConnectionString)
    {
        _subscriberReceived = new WeakAction<MessageBase>(handlerReceived);
        _subscriberSent = new WeakAction<MessageBase>(handlerSent);

        if (onException != null)
        {
            _onExceptionHandler = new WeakAction<Exception>(onException);
        }
    }


    private async void MessagePumpReceived(CancellationToken ct)
    {
        while (true)
        {
            try
            {
                if (_cancellationToken.IsCancellationRequested)
                    break;

                var mqMessage = new NetMQMessage();

                if (_controlInPull.TryReceiveMultipartMessage(ref mqMessage))
                {
                    if (mqMessage.FrameCount == 1) //subscribe notification
                    {
                        continue;
                    }

                    var message = (MessageBase) _serializer.Deserialize(mqMessage.Last.Buffer, typeof(MessageBase));

                    if (_subscriberReceived?.Target != null)
                    {
                        _subscriberReceived.Execute(message);
                    }
                }
                else
                {
                    await Task.Delay(TimeSpan.FromMilliseconds(100), ct);
                }
            }
            catch (Exception ex)
            {
                OnExceptionHandler(ex);
            }
        }
    }

    private async void MessagePumpSent(CancellationToken ct)
    {
        while (true)
        {
            try
            {
                if (_cancellationToken.IsCancellationRequested)
                    break;

                var mqMessage = new NetMQMessage();

                if (_controlOutPull.TryReceiveMultipartMessage(ref mqMessage))
                {
                    if (mqMessage.FrameCount == 1) //subscribe notification
                    {
                        continue;
                    }


                    var message = (MessageBase) _serializer.Deserialize(mqMessage.Last.Buffer, typeof(MessageBase));

                    if (_subscriberSent?.Target != null)
                    {
                        _subscriberSent.Execute(message);
                    }
                }
                else
                {
                    await Task.Delay(TimeSpan.FromMilliseconds(100), ct);
                }
            }
            catch (Exception ex)
            {
                OnExceptionHandler(ex);
            }
        }
    }

    private void OnExceptionHandler(Exception obj)
    {
        if(_onExceptionHandler != null && _onExceptionHandler.IsAlive)
        {
            _onExceptionHandler.Execute(obj);
        }
    }

    public void StartRouting()
    {
        _cancellationTokenSource = new CancellationTokenSource();
        _cancellationToken = _cancellationTokenSource.Token;

        _proxy.Start();
        _poller.RunAsync();

        _workerReceived = Task.Run(
            () => MessagePumpReceived(_cancellationToken),
            _cancellationToken);

        _workerSent = Task.Run(
            () => MessagePumpSent(_cancellationToken),
            _cancellationToken);
    }

    public async Task StopRouting()
    {
        _proxy.Stop();
        _poller.Stop();

        _cancellationTokenSource?.Cancel();
        await Task.WhenAny(Task.WhenAll(_workerSent, _workerReceived),
            Task.Delay(5000)); //wait when all thread finished, but max 5 seconds
    }

    public async void Dispose(bool disposing)
    {
        if (_disposed)
            return;

        if (disposing)
        {
            await StopRouting();
            _poller?.Dispose();
            _poller = null;
            _proxy = null;
            _xSub?.Dispose();
            _xSub = null;
            _xPub?.Dispose();
            _xPub = null;
            _controlInPush?.Dispose();
            _controlInPush = null;
            _controlInPull?.Dispose();
            _controlInPull = null;
            _controlOutPush?.Dispose();
            _controlOutPush = null;
            _controlOutPull?.Dispose();
            _controlOutPull = null;
            _workerSent = null;
            _workerReceived = null;
            _cancellationTokenSource?.Dispose();
            _cancellationTokenSource = null;
            _subscriberReceived.MarkForDeletion();
            _subscriberReceived = null;
            _subscriberSent.MarkForDeletion();
            _subscriberSent = null;
            _onExceptionHandler.MarkForDeletion();
            _onExceptionHandler = null;
            _serializer = null;
        }

        _disposed = true;
    }

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
        GC.Collect();
    }
}

            var response = new RequestSuiteMetaDataResponse();

            try
            {
                var request = new JsonObjectSerializer().Deserialize<RequestSuiteMetaData>(message.Payload);
               //some code skipped
               
                response.Succeed = true;
                response.SuiteName = queryResponse.SuiteName;
                response.TestPhases = queryResponse.TestPhases;// long content around 8 kB
            catch (Exception exp)
            {
                await logger.LogMessage(LogLevel.Error, exp.GetAllMessages());
                response.Error = exp.GetAllMessages();
                response.Succeed = false;
            }

            var payloadString = new JsonObjectSerializer().SerializeToString(response);

            var messageToSend = new MessageBase
            {
                CorrelationId = message.CorrelationId,
                PayloadType = typeof(IRequestSuiteMetaDataResponse).FullName,
                SentTime = DateTime.Now,
                Payload = payloadString
            };

            // problem source:
            await logger.LogMessage(LogLevel.Debug,
                $"Sending message: {Base64Decode(messageToSend.Payload)} of type: {messageToSend.PayloadType}");
            Thread.Sleep(50);
            await context.ResponsePublisher.SendMessage(context.ResponseTopic, messageToSend);  
        }
    }

    private readonly string _source;
    private readonly string _logTopic;
    private ISerializer _serializer;
    private IPublisher _publisher;
    private bool _disposed;

    public CdtMessageLogger(string source, 
        string frontendConnectionString, 
        string logTopic = TopicsList.AllLogs, 
        ISerializer serializer = null, 
        SingleThreadTaskScheduler scheduler = null)
    {
        _source = source;
        _logTopic = logTopic;
        _serializer = serializer ?? new Serialize.Json.JsonObjectSerializer();
        _publisher = new Publisher($">{frontendConnectionString}", scheduler);
    }

    public Task LogMessage(LogLevel level, string message)
    {
        var messageToSend = new MessageBase
        {
            CorrelationId = Guid.NewGuid(),
            PayloadType = typeof(ILogMessage).FullName,
            SentTime = DateTime.Now,
            Payload = _serializer.SerializeToString(new LogMessage
            {
                Level = level,
                Message = RemoveSpecialChars(message),
                Source = _source
            })
        };

        return _publisher.SendMessage(_logTopic, messageToSend);
    }

    private string RemoveSpecialChars(string input)
    {
        
        var output = new StringBuilder(input.Length);
        foreach (var c in input)
        {
            switch (c)
            {
                case TestSlash:
                    output.AppendFormat("{0}{1}", TestBackSlash, TestSlash);
                    break;

                case TestBackSlash:
                    output.AppendFormat("{0}{0}", TestBackSlash);
                    break;

                case TestDblQuote:
                    output.AppendFormat("{0}{1}", TestBackSlash, TestDblQuote);
                    break;

                case '\b':
                    output.Append("\\b");
                    break;

                case '\f':
                    output.Append("\\f");
                    break;

                case '\n':
                    output.Append("\\n");
                    break;

                case '\r':
                    output.Append("\\r");
                    break;

                case '\t':
                    output.Append("\\t");
                    break;

                default:
                    output.Append(c);
                    break;
            }
        }

        return output.ToString();
    }


    public void Dispose(bool disposing)
    {
        if (_disposed)
            return;

        if (disposing)
        {
            _publisher?.Dispose();
            _publisher = null;
            _serializer = null;
        }

        _disposed = true;
    }

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
        GC.Collect();
    }
}

    public Publisher(string connectionString , SingleThreadTaskScheduler scheduler=null)
    {
        _pub = new PublisherSocket(connectionString);
        _pub.Options.SendHighWatermark = 1000;
        _pub.Options.ReceiveBuffer = 44000;
        _pub.Options.SendBuffer = 44000;
        _pub.Options.ReconnectInterval = TimeSpan.FromMilliseconds(50);
        _pub.Options.Linger = TimeSpan.FromSeconds(5);
        _serializer = new Serialize.Json.JsonObjectSerializer();
        _taskScheduler = scheduler ?? new SingleThreadTaskScheduler();
    }

    public void Dispose(bool disposing)
    {
        if (_disposed)
            return;

        if (disposing)
        {
            _pub?.Close();
            _pub?.Dispose();
            _pub = null;
            _serializer = null;
            _taskScheduler.StopSchedulerThread();
            _taskScheduler.Dispose();
            _taskScheduler = null;
        }

        _disposed = true;
    }

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
        GC.Collect();
    }

    public Task SendMessage(string topicName, MessageBase message)
    {
        return Task.Factory.StartNew(() =>
        {
            var data = _serializer.Serialize(message);
            var messToSend = new NetMQMessage();
            messToSend.Append(topicName);
            messToSend.Append(data);
            _pub.SendMultipartMessage(messToSend);
        }, CancellationToken.None, TaskCreationOptions.None, _taskScheduler);
    }
}

System.ArgumentOutOfRangeException: Index was out of range. Must be non-negative and less than the size of the collection.
Parameter name: index
at MajordomoProtocol.MDPBroker.ProcessReceivedMessage(Object sender, NetMQSocketEventArgs e) in MDPBroker.cs:line 317
at NetMQ.NetMQSocket.InvokeEvents(Object sender, PollEvents events)
at NetMQ.NetMQPoller.RunPoller()
at NetMQ.NetMQPoller.Run(SynchronizationContext syncContext)
at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state)
--- End of stack trace from previous location where exception was thrown ---
at System.Threading.Tasks.Task.ExecuteWithThreadLocal(Task& currentTaskSlot)