public PipeInstancesAddedEventArgs(IPipe[] pipes)
 {
     if (pipes == null)
         _Pipes = new IPipe[0];
     else
         _Pipes = pipes;
 }

        async void Receiver(IPipe<ConnectionContext> transportPipe, TaskSupervisor supervisor)
        {
            await Repeat.UntilCancelled(supervisor.StopToken, async () =>
            {
                try
                {
                    await _host.ConnectionCache.Send(transportPipe, supervisor.StopToken).ConfigureAwait(false);
                }
                catch (RabbitMqConnectionException ex)
                {
                    if (_log.IsErrorEnabled)
                        _log.ErrorFormat("RabbitMQ connection failed: {0}", ex.Message);

                    var inputAddress = _host.Settings.GetInputAddress(_settings);

                    await _endpointObservers.Faulted(new Faulted(inputAddress, ex)).ConfigureAwait(false);
                }
                catch (TaskCanceledException)
                {
                }
                catch (Exception ex)
                {
                    if (_log.IsErrorEnabled)
                        _log.ErrorFormat("RabbitMQ receive transport failed: {0}", ex.Message);

                    var inputAddress = _host.Settings.GetInputAddress(_settings);

                    await _endpointObservers.Faulted(new Faulted(inputAddress, ex)).ConfigureAwait(false);
                }
            }).ConfigureAwait(false);
        }

        public ReceiveTransportHandle Start(IPipe<ReceiveContext> receivePipe)
        {
            if (_log.IsDebugEnabled)
                _log.DebugFormat("Starting receive transport: {0}", new Uri(_host.Settings.ServiceUri, _settings.QueueDescription.Path));

            var supervisor =
                new TaskSupervisor($"{TypeMetadataCache<ServiceBusReceiveTransport>.ShortName} - {_host.Settings.GetInputAddress(_settings.QueueDescription)}");

            IPipe<ConnectionContext> connectionPipe = Pipe.New<ConnectionContext>(x =>
            {
                x.UseFilter(new PrepareReceiveQueueFilter(_settings, _subscriptionSettings));

                if (_settings.QueueDescription.RequiresSession)
                {
                    x.UseFilter(new MessageSessionReceiverFilter(receivePipe, _receiveObservers, _endpointObservers, supervisor));
                }
                else
                {
                    x.UseFilter(new MessageReceiverFilter(receivePipe, _receiveObservers, _endpointObservers, supervisor));
                }
            });

            Receiver(connectionPipe, supervisor);

            return new Handle(supervisor);
        }

 public void HandleMessage(object message, IPipe pipe)
 {
     lock (this) {
         this.lastMessage = message;
     }
     waiter.Set();
 }

        public Receiver(MessageReceiver messageReceiver, Uri inputAddress, IPipe<ReceiveContext> receivePipe, ReceiveSettings receiveSettings,
            IReceiveObserver receiveObserver, ITaskSupervisor supervisor)
        {
            _messageReceiver = messageReceiver;
            _inputAddress = inputAddress;
            _receivePipe = receivePipe;
            _receiveSettings = receiveSettings;
            _receiveObserver = receiveObserver;
            _supervisor = supervisor;

            _participant = supervisor.CreateParticipant();

            var options = new OnMessageOptions
            {
                AutoComplete = false,
                AutoRenewTimeout = receiveSettings.AutoRenewTimeout,
                MaxConcurrentCalls = receiveSettings.MaxConcurrentCalls
            };

            options.ExceptionReceived += (sender, x) =>
            {
                if (_log.IsErrorEnabled)
                    _log.Error($"Exception received on receiver: {_inputAddress} during {x.Action}", x.Exception);

                _participant.SetComplete();
            };

            messageReceiver.OnMessageAsync(OnMessage, options);

            _participant.SetReady();

            SetupStopTask();
        }

        public void ProcessHandshake(int TimeOut, ref IPipe pipe)
        {
            //usar un waithandle por cada tipo de handshake.
            //Estos waithandles seran liberados cuando la maquina de estados
            //alcance un estado que libere el wait handle.

            try
            {
                LogCurrentState();
                LogSession.LogMessage("Processing Handshake @ WEIGHT Station...");
                this.Workflow.Start();

                //if (!this.Workflow.WaitForResult(TimeOut))
                //{
                //    throw new TimeoutException("Timeout waiting for StableState");
                //}
                LogSession.LogMessage("Weight Workflow Done");
                this.Workflow.Reset();
            }
            catch (Exception exception)
            {
                LogSession.LogException(exception);
                throw exception;
            }
        }

        Task SendUsingNewConnection(IPipe<OwinHostContext> connectionPipe, OwinHostScope scope, CancellationToken stoppingToken)
        {
            try
            {
                if (_cacheTaskScope.StoppingToken.IsCancellationRequested)
                    throw new TaskCanceledException($"The connection is being disconnected: {_settings.ToDebugString()}");

                if (_log.IsDebugEnabled)
                    _log.DebugFormat("Connecting: {0}", _settings.ToDebugString());
                
                if (_log.IsDebugEnabled)
                {
                    _log.DebugFormat("Connected: {0} (address: {1}, local: {2}", _settings.ToDebugString(),
                        _settings.Host, _settings.Port);
                }

                var hostContext = new HttpOwinHostContext(_settings, _cacheTaskScope);

                hostContext.GetOrAddPayload(() => _settings);

                scope.Connected(hostContext);
            }
            catch (Exception ex)
            {
                Interlocked.CompareExchange(ref _scope, null, scope);

                scope.ConnectFaulted(ex);

                throw new HttpConnectionException("Connect failed: " + _settings.ToDebugString(), ex);
            }

            return SendUsingExistingConnection(connectionPipe, scope, stoppingToken);
        }

 public void HandleMessage(object message, IPipe pipe)
 {
     //TODO handle message in another thread using a queue
     foreach (IHandler handler in handlers)
     {
         handler.HandleMessage(message, this);
     }
 }

        Task SendUsingNewConnection(IPipe<ConnectionContext> connectionPipe, ConnectionScope scope, CancellationToken cancellationToken)
        {
            try
            {
                if (_cacheTaskScope.StoppingToken.IsCancellationRequested)
                    throw new TaskCanceledException($"The connection is being disconnected: {_settings.ToDebugString()}");

                if (_log.IsDebugEnabled)
                    _log.DebugFormat("Connecting: {0}", _settings.ToDebugString());

                IConnection connection;
                if (_settings.ClusterMembers?.Any() ?? false)
                {
                    connection = _connectionFactory.CreateConnection(_settings.ClusterMembers, _settings.ClientProvidedName);
                }
                else
                {
                    var hostNames = Enumerable.Repeat(_settings.Host, 1).ToList();

                    connection = _connectionFactory.CreateConnection(hostNames, _settings.ClientProvidedName);
                }


                if (_log.IsDebugEnabled)
                {
                    _log.DebugFormat("Connected: {0} (address: {1}, local: {2}", _settings.ToDebugString(),
                        connection.Endpoint, connection.LocalPort);
                }

                EventHandler<ShutdownEventArgs> connectionShutdown = null;
                connectionShutdown = (obj, reason) =>
                {
                    Interlocked.CompareExchange(ref _scope, null, scope);

                    scope.Shutdown(reason.ReplyText);

                    connection.ConnectionShutdown -= connectionShutdown;
                };

                connection.ConnectionShutdown += connectionShutdown;

                var connectionContext = new RabbitMqConnectionContext(connection, _settings, _cacheTaskScope);

                connectionContext.GetOrAddPayload(() => _settings);

                scope.Connected(connectionContext);
            }
            catch (BrokerUnreachableException ex)
            {
                Interlocked.CompareExchange(ref _scope, null, scope);

                scope.ConnectFaulted(ex);

                throw new RabbitMqConnectionException("Connect failed: " + _settings.ToDebugString(), ex);
            }

            return SendUsingExistingConnection(connectionPipe, scope, cancellationToken);
        }

        public void CopyToArrayCopiesZeroPipesToArrayIfNoneAreSelected()
        {
            Guid[] ids = InitializeTestEntries(3);

            IPipe[] arr = new IPipe[0];

            _selected.CopyTo(arr, 0);
            Assert.Equal(0, arr.Length);
        }

        public static void AddPipe(IPipe pipe)
        {
            if (Pipeline.Any(existingPipe => existingPipe.GetType() == pipe.GetType()))
            {
                return;
            }

            Pipeline.Add(pipe);
        }

		public static void AddPipe(IPipe pipe)
		{
			foreach( var existingPipe in _pipeline )
			{
				if( existingPipe.GetType () == pipe.GetType () )
					return;
			}
			_pipeline.Add (pipe);
		}

        /// <summary>
        /// Rescue exceptions via the alternate pipe
        /// </summary>
        /// <param name="configurator"></param>
        /// <param name="rescuePipe"></param>
        public static void UseDeadLetterQueue(this IPipeConfigurator<ReceiveContext> configurator, IPipe<ReceiveContext> rescuePipe)
        {
            if (configurator == null)
                throw new ArgumentNullException(nameof(configurator));

            var rescueConfigurator = new DeadLetterPipeSpecification(rescuePipe);

            configurator.AddPipeSpecification(rescueConfigurator);
        }

        /// <summary>
        /// Adds a RabbitMQ Basic Consumer to the pipeline
        /// </summary>
        /// <param name="configurator"></param>
        /// <param name="pipe"></param>
        /// <param name="settings"></param>
        /// <param name="receiveObserver"></param>
        /// <param name="endpointObserver"></param>
        /// <param name="exchangeBindings"></param>
        /// <param name="taskSupervisor"></param>
        /// <param name="mediator"></param>
        public static void RabbitMqConsumer(this IPipeConfigurator<ConnectionContext> configurator, IPipe<ReceiveContext> pipe, ReceiveSettings settings, IReceiveObserver receiveObserver, IReceiveEndpointObserver endpointObserver, IEnumerable<ExchangeBindingSettings> exchangeBindings, ITaskSupervisor taskSupervisor, Mediator<ISetPrefetchCount> mediator)
        {
            if (configurator == null)
                throw new ArgumentNullException(nameof(configurator));

            var pipeBuilderConfigurator = new RabbitMqConsumerPipeSpecification(pipe, settings, receiveObserver, endpointObserver, exchangeBindings, taskSupervisor, mediator);

            configurator.AddPipeSpecification(pipeBuilderConfigurator);
        }

        async Task SendUsingNewModel(IPipe<ModelContext> modelPipe, ModelScope scope, CancellationToken cancellationToken)
        {
            IPipe<ConnectionContext> connectionPipe = Pipe.ExecuteAsync<ConnectionContext>(async connectionContext =>
            {
                IModel model = await connectionContext.CreateModel().ConfigureAwait(false);

                EventHandler<ShutdownEventArgs> modelShutdown = null;
                modelShutdown = (obj, reason) =>
                {
                    model.ModelShutdown -= modelShutdown;

                    Interlocked.CompareExchange(ref _scope, null, scope);

                    scope.Close();
                };

                model.ModelShutdown += modelShutdown;

                var modelContext = new RabbitMqModelContext(connectionContext, model, connectionContext.CancellationToken);

                scope.Connected(modelContext);

                try
                {
                    using (SharedModelContext context = await scope.Attach(cancellationToken).ConfigureAwait(false))
                    {
                        await modelPipe.Send(context).ConfigureAwait(false);
                    }
                }
                catch (Exception ex)
                {
                    if (_log.IsDebugEnabled)
                        _log.Debug("The existing model usage threw an exception", ex);

                    Interlocked.CompareExchange(ref _scope, null, scope);

                    scope.Close();

                    throw;
                }
            });

            try
            {
                await _connectionCache.Send(connectionPipe, new CancellationToken()).ConfigureAwait(false);
            }
            catch (Exception exception)
            {
                if (_log.IsDebugEnabled)
                    _log.Debug("The connection threw an exception", exception);

                Interlocked.CompareExchange(ref _scope, null, scope);

                throw;
            }
        }

        /// <summary>
        /// Adds a RabbitMQ Basic Consumer to the pipeline
        /// </summary>
        /// <param name="configurator"></param>
        /// <param name="pipe"></param>
        /// <param name="settings"></param>
        /// <param name="receiveObserver"></param>
        /// <param name="transportObserver"></param>
        /// <param name="exchangeBindings"></param>
        /// <param name="supervisor"></param>
        /// <param name="managementPipe"></param>
        /// <param name="sendEndpointProvider"></param>
        /// <param name="publishEndpointProvider"></param>
        /// <param name="host"></param>
        public static void RabbitMqConsumer(this IPipeConfigurator<ConnectionContext> configurator, IPipe<ReceiveContext> pipe, ReceiveSettings settings, IReceiveObserver receiveObserver, IReceiveTransportObserver transportObserver, IEnumerable<ExchangeBindingSettings> exchangeBindings, ITaskSupervisor supervisor, IManagementPipe managementPipe, ISendEndpointProvider sendEndpointProvider, IPublishEndpointProvider publishEndpointProvider, IRabbitMqHost host)
        {
            if (configurator == null)
                throw new ArgumentNullException(nameof(configurator));

            var pipeBuilderConfigurator = new RabbitMqConsumerPipeSpecification(pipe, settings, receiveObserver, transportObserver, exchangeBindings,
                supervisor, managementPipe, sendEndpointProvider, publishEndpointProvider, host);

            configurator.AddPipeSpecification(pipeBuilderConfigurator);
        }

        /// <summary>
        /// Start the receive transport, returning a Task that can be awaited to signal the transport has 
        /// completely shutdown once the cancellation token is cancelled.
        /// </summary>
        /// <param name="receivePipe"></param>
        /// <returns>A task that is completed once the transport is shut down</returns>
        public ReceiveTransportHandle Start(IPipe<ReceiveContext> receivePipe)
        {
            var stopTokenSource = new CancellationTokenSource();

            IPipe<ConnectionContext> pipe = Pipe.New<ConnectionContext>(x => x.RabbitMqConsumer(receivePipe, _settings, _receiveObservers, _endpointObservers, _exchangeBindings));

            Task receiverTask = Receiver(pipe, stopTokenSource.Token);

            return new Handle(stopTokenSource, receiverTask);
        }

        /// <summary>
        /// Rescue exceptions via the alternate pipe
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="configurator"></param>
        /// <param name="rescuePipe"></param>
        /// <param name="exceptionFilter"></param>
        public static void UseRescue(this IPipeConfigurator<ReceiveContext> configurator, IPipe<ExceptionReceiveContext> rescuePipe,
            IPolicyExceptionFilter exceptionFilter = null)
        {
            if (configurator == null)
                throw new ArgumentNullException(nameof(configurator));

            var rescueConfigurator = new ReceiveContextRescuePipeSpecification(rescuePipe, exceptionFilter);

            configurator.AddPipeSpecification(rescueConfigurator);
        }

        async Task SendUsingNewModel(IPipe<ModelContext> modelPipe, ModelScope scope, CancellationToken cancellationToken)
        {
            IPipe<ConnectionContext> connectionPipe = Pipe.ExecuteAsync<ConnectionContext>(async connectionContext =>
            {
                try
                {
                    if (_log.IsDebugEnabled)
                        _log.DebugFormat("Creating model: {0}", connectionContext.HostSettings.ToDebugString());

                    var model = await connectionContext.CreateModel().ConfigureAwait(false);

                    EventHandler<ShutdownEventArgs> modelShutdown = null;
                    modelShutdown = (obj, reason) =>
                    {
                        model.ModelShutdown -= modelShutdown;

                        Interlocked.CompareExchange(ref _scope, null, scope);

                        scope.Shutdown(reason.ReplyText);
                    };

                    model.ModelShutdown += modelShutdown;

                    var modelContext = new RabbitMqModelContext(connectionContext, model, _cacheTaskScope, _modelSettings);

                    scope.Created(modelContext);
                }
                catch (Exception ex)
                {
                    Interlocked.CompareExchange(ref _scope, null, scope);

                    scope.CreateFaulted(ex);

                    throw;
                }

                await SendUsingExistingModel(modelPipe, scope, cancellationToken).ConfigureAwait(false);
            });

            try
            {
                await _connectionCache.Send(connectionPipe, _cacheTaskScope.StoppedToken).ConfigureAwait(false);
            }
            catch (Exception exception)
            {
                if (_log.IsDebugEnabled)
                    _log.Debug("The connection threw an exception", exception);

                Interlocked.CompareExchange(ref _scope, null, scope);

                scope.CreateFaulted(exception);

                throw;
            }
        }

        /// <summary>
        /// Start the receive transport, returning a Task that can be awaited to signal the transport has 
        /// completely shutdown once the cancellation token is cancelled.
        /// </summary>
        /// <param name="receivePipe"></param>
        /// <returns>A task that is completed once the transport is shut down</returns>
        public ReceiveTransportHandle Start(IPipe<ReceiveContext> receivePipe)
        {
            var supervisor = new TaskSupervisor();

            var pipe = Pipe.New<ConnectionContext>(
                    x => x.RabbitMqConsumer(receivePipe, _settings, _receiveObservers, _endpointObservers, _exchangeBindings, supervisor));

            Receiver(pipe, supervisor);

            return new Handle(supervisor);
        }