/// <summary>
 /// Add a message consumer to monitor for messages received
 /// </summary>
 public void AddMessageConsumer(IMessageConsumer consumer)
 {
     consumer.Listener += (m) =>
     {
         _lastMsgRecd = DateTime.UtcNow;
     };
 }

        public MessageHandler(IMessageConsumer consumer, MessagingClient.EventDelegate callback)
        {
            this.consumer = consumer;
            this.callback = callback;

            AddListener();
        }

 public void Subscribe(IMessageConsumer consumer)
 {
     if (_consumers.Contains(consumer))
     {
         return;
     }
     _consumers.Add(consumer);
 }

        public NMSConsumer(IMessageConsumer consumer, String clientId, String subscriptionId)
        {
            this.consumer = consumer;
            this.clientId = clientId;
            this.subscriptionId = subscriptionId;

            consumer.Listener += onMessage;
        }

 public void Unsubscribe(IMessageConsumer consumer)
 {
     if (!_consumers.Contains(consumer))
     {
         return;
     }
     _consumers.Remove(consumer);
 }

 public Consumer(IMessageConsumer messageConsumer, string clientId, string topicName)
 {
     _consumer = messageConsumer;
     _clientId = clientId;
     _topicName = topicName;
     _consumer.Listener += Update;
     _running = true;
     Delays = new List<TimeSpan>();
 }

        public void RegisterImpl(IMessageConsumer impl, string serviceId)
        {
            var rule = MetaData.GetServiceRoutingRule(serviceId);
            if (rule == null)
            {
                throw new Exception();
            }

            implements[rule.GateRule.GetServiceId()] = impl;
        }

        public void RegisterDelegate(IMessageConsumer consumer, string serviceId)
        {
            var rule = MetaData.GetServiceRoutingRule(serviceId);
            if (rule == null)
            {
                throw new Exception();
            }

            delegates[rule.GateRule.GetServiceId()] = consumer;
        }

 public IMessage FudgeDecodeMessage(ISession session, IMessageConsumer consumer, IMessage message)
 {
     try
     {
         return new ActiveMQObjectMessage { Body = DecodeObject(GetMessage(message)) };
     }
     catch (Exception e)
     {
         return new ActiveMQObjectMessage { Body = e };
     }
 }

 public DistributableCommandBus(ICommandHandlerProvider handlerProvider,
     ILinearCommandManager linearCommandManager,
     IMessageConsumer commandConsumer,
     string receiveEndPoint,
     bool inProc)
     : base(handlerProvider, linearCommandManager, receiveEndPoint, inProc)
 {
     _commandConsumer = commandConsumer as IInProcMessageConsumer;
     _commandDistributor = _commandConsumer;
     _isDistributor = _commandDistributor is IMessageDistributor;
 }

 public MessageSubscriber(IAdvancedBus bus,
     IMessageConsumer messageConsumer, 
     ILogger logger, 
     IEnvironment environment,
     IExchange exchange,
     IQueue queue)
 {
     _messageConsumer = messageConsumer;
     _bus = bus;
     _logger = logger;
     _environment = environment;
     _exchange = exchange;
     _queue = queue;
 }

        static WebApiApplication()
        {
            try
            {
                Configuration.Instance.UseLog4Net()
                                      .RegisterMessageContextType(typeof(MessageContext));

                _Logger = IoCFactory.Resolve<ILoggerFactory>().Create(typeof(WebApiApplication));

                _CommandDistributor = new CommandDistributor("tcp://127.0.0.1:5000",
                                                                new string[] {
                                                                    "tcp://127.0.0.1:5001"
                                                                    , "tcp://127.0.0.1:5002"
                                                                    , "tcp://127.0.0.1:5003"
                                                                }
                                                               );

                Configuration.Instance.RegisterCommandConsumer(_CommandDistributor, "CommandDistributor")
                             .CommandHandlerProviderBuild(null, "CommandHandlers")
                             .RegisterDisposeModule()
                             .RegisterMvc();

                _EventPublisher = IoCFactory.Resolve<IEventPublisher>();
                _EventPublisher.Start();
                _DomainEventConsumer = IoCFactory.Resolve<IMessageConsumer>("DomainEventConsumer");
                _DomainEventConsumer.Start();
                _ApplicationEventConsumer = IoCFactory.Resolve<IMessageConsumer>("ApplicationEventConsumer");
                _ApplicationEventConsumer.Start();

                var commandHandlerProvider = IoCFactory.Resolve<ICommandHandlerProvider>();
                _CommandConsumer1 = new CommandConsumer(commandHandlerProvider,
                                                           "tcp://127.0.0.1:5001");
                _CommandConsumer2 = new CommandConsumer(commandHandlerProvider,
                                                           "tcp://127.0.0.1:5002");
                _CommandConsumer3 = new CommandConsumer(commandHandlerProvider,
                                                           "tcp://127.0.0.1:5003");

                _CommandConsumer1.Start();
                _CommandConsumer2.Start();
                _CommandConsumer3.Start();
                _CommandDistributor.Start();

                _CommandBus = IoCFactory.Resolve<ICommandBus>() as IMessageConsumer;
                _CommandBus.Start();
            }
            catch (Exception ex)
            {
                _Logger.Error(ex.GetBaseException().Message, ex);
            }
        }

 public void AddLocalInstanceSubscription(IMessageConsumer consumer)
 {
     localInstanceSubscriptions.Write(writer =>
     {
         foreach (var type in reflection.GetMessagesConsumed(consumer))
         {
             List<WeakReference> value;
             if (writer.TryGetValue(type.FullName, out value) == false)
             {
                 value = new List<WeakReference>();
                 writer.Add(type.FullName, value);
             }
             value.Add(new WeakReference(consumer));
         }
     });
     RaiseSubscriptionChanged();
 }

        private void btnSubscribe_Click(object sender, EventArgs e)
        {
            try
            {
                if (!GlobalFunction.CheckControlInput(txtTopicName, "Topic Name", 0, false)) return;

                if (m_consumer != null)
                {
                    m_consumer.Close();
                }
                if (txtSelector.Text != "")
                {
                    m_consumer = m_mq.CreateConsumer(rdTopic.Checked, txtTopicName.Text, txtSelector.Text);
                }
                else
                {
                    m_consumer = m_mq.CreateConsumer(rdTopic.Checked, txtTopicName.Text);
                }

                m_consumer.Listener += new MessageListener(consumer_listener);
            }
            catch (System.Exception ex)
            {
                GlobalFunction.MsgBoxException(ex.Message, "btnSubscribe_Click");
            }
        }

        /// <summary>
        /// 消息消费构造器
        /// </summary>
        /// <param name="brokerUri">地址</param>
        /// <param name="username">用户名</param>
        /// <param name="psw">密码</param>
        /// <param name="clientId">客户端标识 兼做队列接收目的地</param>
        /// <param name="isClient">true 客户端；false 服务端</param>
        public OpenWireConsumer(string brokerUri, string username, string psw, string clientId,bool isClient)
        {
            NMSConnectionFactory _factory = new NMSConnectionFactory(brokerUri, clientId);
            _connection = _factory.CreateConnection(username, psw);
            _connection.Start();
            _session = _connection.CreateSession(AcknowledgementMode.AutoAcknowledge);

            if (isClient)
            {
                _qReceiveDest = _session.GetDestination(clientId, DestinationType.TemporaryQueue);
            }
            else
            {
                _qReceiveDest = _session.GetQueue(clientId);
            }

            _messageConsumer = _session.CreateConsumer(_qReceiveDest);
            _messageConsumer.Listener += (message) =>
            {
                if (Listener != null)
                {
                    Listener(message);
                }
            };
        }

        public void Dispose()
        {
            lock (this)
            {
                this.isDisposed = true;
                this.consumer.Dispose();
                this.consumer = null;

                if (this.replyProducer != null)
                {
                    this.replyProducer.Dispose();
                    this.replyProducer = null;
                }

                this.requestReplyCallback = null;

                this.session.Dispose();
                this.session = null;

                this.connection.ConnectionInterrupted -= new EventHandler<NmsConnectionEventArgs>(connection_ConnectionInterrupted);
                this.connection.ConnectionResumed -= new EventHandler<NmsConnectionEventArgs>(connection_ConnectionResumed);

                this.connection = null;
            }
        }

        public Queue(MsgDeliveryMode mode = MsgDeliveryMode.NonPersistent)
        {
            Uri msgQueue = new Uri("activemq:tcp://localhost:61616");

            _factory = new ConnectionFactory(msgQueue);
            try
            {
                _connection = _factory.CreateConnection();
            }
            catch (NMSConnectionException ex)
            {
                Log.FatalException("Error connecting to MQ server", ex);
                throw;
            }
            // TODO check _connection for null
            _connection.RequestTimeout = TimeSpan.FromSeconds(60);
            Session = _connection.CreateSession();

            // TODO need to find out if queue exists.
            // It creates a new queue if it doesn't exist.
            _destination = Session.GetDestination("queue://TwitterSearchStream");
            _consumer = Session.CreateConsumer(_destination);

            _producer = Session.CreateProducer(_destination);
            _producer.RequestTimeout = TimeSpan.FromSeconds(60);
            _producer.DeliveryMode = mode;

            _connection.Start();

            _connection.ExceptionListener += _connection_ExceptionListener;
            _connection.ConnectionInterruptedListener += _connection_ConnectionInterruptedListener;
        }

		override public void SetUp()
        {
            base.SetUp();
			acknowledgementMode = AcknowledgementMode.Transactional;
            Drain();
            consumer = Session.CreateConsumer(Destination);
            producer = Session.CreateProducer(Destination);
        }

 public MessageTransporter()
 {
     _connectionFactory = new Apache.NMS.Stomp.ConnectionFactory("tcp://0.0.0.0:61613");
     _connection = _connectionFactory.CreateConnection();
     _session = _connection.CreateSession();
     _destination = SessionUtil.GetDestination(_session, "queue://testingQueue");
     _messageProducer = _session.CreateProducer(_destination);
     _messageConsumer = _session.CreateConsumer(_destination);
 }

        public void Connect()
        {
            while (!ableToSendEvents) {
                Uri connecturi = null;
                //if (textBoxSIPIPAddress.Text.StartsWith("ssl://"))
                //{
                Console.WriteLine ("Trying to connect to ActiveMQ broker ");
                //	connecturi = new Uri("activemq:" + textBoxSIPIPAddress.Text + ":" + textBoxSIPPort.Text + "?transport.ClientCertSubject=E%[email protected], CN%3DCommunication Tool"); // Connect to the ActiveMQ broker
                //}
                //else
                //{
                //log4.Debug(name + ": Trying to connect to ActiveMQ broker via non-secure connection");
                connecturi = new Uri ("activemq:tcp://localhost:61616"); // Connect to the ActiveMQ broker
                //}
                //Console.WriteLine("activeMQ::About to connect to " + connecturi);

                try {

                    // NOTE: ensure the nmsprovider-activemq.config file exists in the executable folder.
                    IConnectionFactory factory = new ConnectionFactory (connecturi);

                    // Create a new connection and session for publishing events
                    activeMQConnection = factory.CreateConnection ();
                    activeMQSession = activeMQConnection.CreateSession ();

                    IDestination destination = SessionUtil.GetDestination (activeMQSession, "topic://SIFTEO");
                    //Console.WriteLine("activeMQ::Using destination: " + destination);

                    // Create the producer
                    activeMQProducer = activeMQSession.CreateProducer (destination);
                    activeMQProducer.DeliveryMode = MsgDeliveryMode.Persistent;
                    destination = SessionUtil.GetDestination (activeMQSession, "topic://XVR.CCC");
                    activeMQConsumer = activeMQSession.CreateConsumer (destination);
                    //activeMQConsumer.Listener += new MessageListener(OnCCCMessage);

                    // Start the connection so that messages will be processed
                    activeMQConnection.Start ();
                    //activeMQProducer.Persistent = true;

                    // Enable the sending of events
                    //log4.Debug(name + ": ActiveMQ connected on topics XVR.CCC and XVR.SDK");
                    ableToSendEvents = true;

                } catch (Exception exp) {
                    // Report the problem in the output.log (Program Files (x86)\E-Semble\XVR 2012\XVR 2012\XVR_Data\output_log.txt)
                    //Console.WriteLine("*** AN ACTIVE MQ ERROR OCCURED: " + exp.ToString() + " ***");
                    //log4.Error(name + ": Error connecting to ActiveMQ broker: " + exp.Message);
                    //log4.Error((exp.InnerException != null) ? exp.InnerException.StackTrace : "");

                    Console.WriteLine (exp.Message);
                }
                System.Threading.Thread.Sleep (1000);
            }
        }