public void EventPumpQueueLength(int numberOfProducers, bool producerDelay)
        {
            EventQueue q = new EventQueue();
            EventProducer[] producers = new EventProducer[numberOfProducers];
            for (int i = 0; i < numberOfProducers; i++)
            {
                producers[i] = new EventProducer(q, i, producerDelay);
            }

            using (EventPump pump = new EventPump(NullListener.NULL, q, false))
            {
                pump.Name = "EventPumpQueueLength";
                pump.Start();

                foreach (EventProducer p in producers)
                {
                    p.ProducerThread.Start();
                }
                foreach (EventProducer p in producers)
                {
                    p.ProducerThread.Join();
                }
                pump.Stop();
            }
            Assert.That(q.Count, Is.EqualTo(0));

            foreach (EventProducer p in producers)
            {
                Console.WriteLine(
                    "#Events: {0}, MaxQueueLength: {1}", p.SentEventsCount, p.MaxQueueLength);
                Assert.IsNull(p.Exception, "{0}", p.Exception);
            }
        }

 public EventList(EventQueue<string> events)
 {
     InitializeComponent();
     EventQueue = events;
     EventQueue.Updated += Events_Updated;
     ReloadEvents();
 }

        /**
	     * EventManager constructor comment.
	     *
	     * @param robotProxy robotProxy
	     */

        public EventManager(BasicRobotProxy robotProxy)
        {
            registerNamedEvents();
            this.robotProxy = robotProxy;
            eventQueue = new EventQueue();

            reset();
        }

        /// <summary>
        /// Constructs a new EventManager.
        /// </summary>
        /// <param name="robotProxy">the robot proxy that this event manager applies to.</param>
        public EventManager(BasicRobotProxy robotProxy)
        {
            this.robotProxy = robotProxy;
            eventQueue = new EventQueue();

            RegisterEventNames();
            Reset();
        }

 public Environment(DateTime initialDateTime, int randomSeed, TimeSpan? defaultStep = null) {
   DefaultTimeStepSeconds = (defaultStep ?? TimeSpan.FromSeconds(1)).Duration().TotalSeconds;
   StartDate = initialDateTime;
   Now = initialDateTime;
   Random = new SystemRandom(randomSeed);
   ScheduleQ = new EventQueue(InitialMaxEvents);
   Queue = new Queue<Event>();
   Logger = Console.Out;
 }

 /// <summary>
 /// start a small server that listens to UDP messages through _port 1337(as indicated when initializing the server instance).
 /// </summary>
 /// <param name="_port"></param>
 public Server(int _port)
 {
     this._iPort = _port;
     _lMessageBacklog = new EventQueue<NetworkMessage>();
     _lPlayerRooms = new EventList<Hub>();
     Instance = this;
     Console.WriteLine("Setting up the server...");
     MessageReceived += ParseMessage;
 }

 /// <summary>
 /// Default constructor
 /// </summary>
 public EventOptions()
 {
     HostName = _hostName;
     OnErrorHttpCaptureFlags = HttpCaptureFlags.All;
     SourceLevels = SourceLevels.All;
     ConfigKeys = new Keys();
     ConnectorType = "HashTag.Logging.Client.NLog.NLogEventConnector, HashTag.Logging.Client";
     _queue = new EventQueue(this);
 }

 private static void VerifyQueue(EventQueue q)
 {
     for (int index = 0; index < events.Length; index++)
     {
         Event e = q.Dequeue(false);
         Assert.AreEqual(events[index].GetType(), e.GetType(),
             string.Format("Event {0}", index));
     }
 }

 public void PumpAutoStopsOnRunFinished()
 {
     EventQueue q = new EventQueue();
     EventPump pump = new EventPump( NullListener.NULL, q, true );
     Assert.IsFalse( pump.Pumping, "Should not be pumping initially" );
     StartPump( pump, 1000 );
     Assert.IsTrue( pump.Pumping, "Pump failed to start" );
     q.Enqueue( new RunFinishedEvent( new Exception() ) );
     WaitForPumpToStop( pump, 1000 );
     Assert.IsFalse( pump.Pumping, "Pump failed to stop" );
 }

 public void SetUp()
 {
     theQueue = new EventQueue<FakeTopic>();
     theQueue.Write(new ServerEvent("2", "2"));
     theQueue.Write(new ServerEvent("1", "1"));
     
     for (int i = 3; i < 8; i++)
     {
         theQueue.Write(new ServerEvent(i.ToString(), i.ToString()));
     }    
 }

 public void Attach(EventBus bus)
 {
     var q = new EventQueue(EventQueueId.Data, EventQueueType.Master, EventQueuePriority.Normal, 25600, null)
     {
         IsSynched = true,
         Name = $"attached {bus.framework.Name}"
     };
     q.Enqueue(new OnQueueOpened(q));
     bus.DataPipe.Add(q);
     this.attached[this.attachedCount++] = q;
 }

 public void PumpEvents()
 {
     EventQueue q = new EventQueue();
     EnqueueEvents( q );
     QueuingEventListener el = new QueuingEventListener();
     EventPump pump = new EventPump( el, q, false );
     Assert.IsFalse( pump.Pumping, "Should not be pumping initially" );
     StartPump( pump, 1000 );
     Assert.IsTrue( pump.Pumping, "Pump should still be running" );
     StopPump( pump, 1000 );
     Assert.IsFalse( pump.Pumping, "Pump should have stopped" );
     VerifyQueue( el.Events );
 }

 protected override void OnConnected()
 {
     foreach (var s in Series)
     {
         var q = new EventQueue(EventQueueId.Data, EventQueueType.Master, EventQueuePriority.Normal, 25600, null)
         {
             IsSynched = true,
             Name = s.Name
         };
         q.Enqueue(new OnQueueOpened(q));
         this.framework.EventBus.DataPipe.Add(q);
         this.emitters.Add(new DataSeriesObject(s, DateTime1, DateTime2, q, Processor));
     }
 }

 public void PumpEventsWithAutoStop()
 {
     EventQueue q = new EventQueue();
     EnqueueEvents( q );
     QueuingEventListener el = new QueuingEventListener();
     EventPump pump = new EventPump( el, q, true );
     pump.Start();
     int tries = 10;
     while( --tries > 0 && q.Count > 0 )
     {
         Thread.Sleep(100);
     }
     VerifyQueue( el.Events );
     Assert.IsFalse( pump.Pumping, "Pump failed to stop" );
 }

        public void spin_up_pre_builds()
        {
            var factory = MockRepository.GenerateMock<IEventQueueFactory<FakeTopic>>();
            var queue = new EventQueue<FakeTopic>();
            var topic = new FakeTopic{
                Name = "Top"
            };

            factory.Stub(x => x.BuildFor(topic)).Return(queue);

            var family = new TopicFamily<FakeTopic>(factory);
            family.SpinUpChannel(topic);

            factory.AssertWasCalled(x => x.BuildFor(topic));
        }

        public BaseInput(InputElement input, SelectorElement selector, EventQueue queue)
            : base()
        {
            InputName = input.Name;
            InputType = input.Type;
            SelectorName = selector.Name;
            Processor = selector.Processor;

            switch (selector.Login)
            {
                case "success": Login = LoginStatus.SUCCESS; break;
                case "failure": Login = LoginStatus.FAILURE; break;
                default: Login = LoginStatus.UNKNOWN; break;
            }

            equeue = queue;
        }

        public RuntimeEngine(RuntimeEntityService entityService, ISystemManager systemManager, EventQueue eventQueue, ILogger logger)
        {
            AddEntityAccessGate = new AccessGate();
            EventQueue = eventQueue;
            _logger = logger;
            EntityService = entityService;
            EntityService.AddEntityAccessGate = AddEntityAccessGate;
            SystemManager = systemManager;
            SystemUpdateScheduler = new SystemUpdateScheduler(SystemManager.Systems);

            foreach (var updateBurst in SystemUpdateScheduler.UpdateBursts)
            {
                _logger.Info($"Update Burst: {string.Join(", ", updateBurst.Systems.Select(p => p.System.Name))}");
            }

            // Add any entities that are already loaded into the engine.
            SystemManager.AddEntities(entityService.Entities);
        }

        public void Start(int port, CryptoAlgoId_Values cryptoAlgoId, byte[] content, EventQueue eventQueue)
        {
            this.cryptoAlgoId = cryptoAlgoId;
            this.content = content;
            this.eventQueue = eventQueue;
            this.aes = PccrrUtitlity.CreateAes(cryptoAlgoId);
            this.iv = new byte[16];
            for (int i = 0; i < iv.Length; i++)
            {
                this.iv[i] = (byte)i;
            }

            pccrrServer = new PccrrServer(port);

            pccrrServer.MessageArrived += new MessageArrivedEventArgs(pccrrServer_MessageArrived);

            pccrrServer.StartListening();
        }

        public void Produce(EventEntry item, string processor, EventQueue.Priority priority = EventQueue.Priority.Low)
        {
            if (equeue == null)
            {
                Log.Error("Unable to produce events before queue initialization!?!?");
                return;
            }

            if (string.IsNullOrEmpty(processor))
            {
                Log.Error("Unable to queue events with empty processor name");
                return;
            }

            Log.Info("Service[" + item.Id + "@" + item.Input.Name + "] (re)queued message"
                + " with first processor name " + processor);
            equeue.Produce(item, processor, priority);
        }

        protected bool Authenticate(Session session, Identity identity, Request request)
        {
            var correlationId = new CorrelationID();
            var eventQueue = new EventQueue();
            session.SendAuthorizationRequest(request, identity, eventQueue, correlationId);
            while (true)
            {
                var eventArgs = eventQueue.NextEvent();

                foreach (var message in eventArgs.GetMessages())
                {
                    if (MessageTypeNames.AuthorizationFailure.Equals(message.MessageType))
                        return false;

                    if (MessageTypeNames.AuthorizationSuccess.Equals(message.MessageType))
                        return true;

                    throw new Exception("Unknown message type: " + message);
                }
            }
        }