ESBMessageRequest.Send

gistfile1.cs

public void Send() {
            var transport = _transport ?? Helper.GetTCPTransport();
            var hasAction = _action != null;
            _receivedResponse = false;
            _timeOut = _timeOut == null ? TimeSpan.FromSeconds(10) : _timeOut;
            CorrelationID = CorrelationID == default(Guid) ? Guid.NewGuid() : CorrelationID;
            if (hasAction) {
                var responseTopic = Topic.Select<TResponse>();
                if (responseTopic.State == ObjectState.InValid) Topic.Register<TResponse>();
                var subscriberName = String.Concat("Subscriber", responseTopic.TopicName,
                    Guid.NewGuid());
                _subscriber =
                    Subscriber.New(subscriberName)
                        .SubscribeTo(Topic.Select<TResponse>().Equal("CorrelationID", CorrelationID.ToString()))
                        .AddTransport(responseTopic.TopicName, transport, responseTopic.TopicName);
                var ex = MethodHelper.Try(() => _subscriber.Save());
                if (ex != null) {
                    MethodHelper.Try(() => _subscriber.Delete());
                    throw new ESBException(String.Format("Could not make request because of the following exception: {0}", ex.ToString()));
                } 
                _handler = _subscriber.OnMessageReceived<TResponse>(responseTopic.TopicName,
                    msg =>
                    {
                        _action(msg);
                        CleanUp();
                        _receivedResponse = true;
                    }, error =>
                    {
                        error.Continue = true;
                    }, _responseInterval ?? TimeSpan.FromMilliseconds(100), 1);
            }
            if (Topic.Select<TRequest>().State == ObjectState.InValid) Topic.Register<TRequest>();
            var requestTopic = Topic.Select<TRequest>();
            requestTopic.Publish(this as TRequest);
            if (hasAction) {
                var action = new Action(() =>
                {
                    _startTime = DateTime.Now;
                    while (!_receivedResponse) {
                        if (DateTime.Now.Subtract(_startTime).TotalMilliseconds >= _timeOut.Value.TotalMilliseconds) break;
                        Thread.Sleep(0);
                    }
                    CleanUp();
                });
                if (_waitForResponse) action();
                else Task.Factory.StartNew(action);
            }
        }