static void Main(string[] args)
        {
            IPublisher publisher = new Publisher(countOfChannels, countOfSubscribers);
            IDataMediator dataMediator = (IDataMediator) publisher;

            ISubscriber[] subscribers = new ISubscriber[countOfSubscribers];
            for (int i = countOfSubscribers - 1; i >= 0; i--)
            {
                subscribers[i] = new Subscriber(publisher);
            }

            runSubscribe(subscribers, publisher);
            Thread.CurrentThread.Priority = ThreadPriority.Highest;

            for (int i = 0; i < 1000; i++)
            {
                Stopwatch latency = Stopwatch.StartNew();

                int onLoadSubscribersCount = dataMediator.onLoad(randomBytes(10000), rnd.Next(countOfChannels));

                latency.Stop();
                Console.Write("latency is\t" + latency.Elapsed.Milliseconds + "\tfor\t" + onLoadSubscribersCount + "\tsubscribers");
                Console.WriteLine();

            }

            Console.WriteLine("thats all");
            Console.ReadLine();
        }

 public void SetupMessaging(ISubscriber mainBus)
 {
     mainBus.Subscribe<SystemMessage.StateChangeMessage>(_projectionManager);
     if (_runProjections >= RunProjections.System)
     {
         mainBus.Subscribe<ProjectionManagementMessage.Post>(_projectionManager);
         mainBus.Subscribe<ProjectionManagementMessage.UpdateQuery>(_projectionManager);
         mainBus.Subscribe<ProjectionManagementMessage.GetQuery>(_projectionManager);
         mainBus.Subscribe<ProjectionManagementMessage.Delete>(_projectionManager);
         mainBus.Subscribe<ProjectionManagementMessage.GetStatistics>(_projectionManager);
         mainBus.Subscribe<ProjectionManagementMessage.GetState>(_projectionManager);
         mainBus.Subscribe<ProjectionManagementMessage.GetResult>(_projectionManager);
         mainBus.Subscribe<ProjectionManagementMessage.Disable>(_projectionManager);
         mainBus.Subscribe<ProjectionManagementMessage.Enable>(_projectionManager);
         mainBus.Subscribe<ProjectionManagementMessage.Abort>(_projectionManager);
         mainBus.Subscribe<ProjectionManagementMessage.SetRunAs>(_projectionManager);
         mainBus.Subscribe<ProjectionManagementMessage.Reset>(_projectionManager);
         mainBus.Subscribe<ProjectionManagementMessage.StartSlaveProjections>(_projectionManager);
         mainBus.Subscribe<ProjectionManagementMessage.RegisterSystemProjection>(_projectionManager);
         mainBus.Subscribe<ProjectionManagementMessage.Internal.CleanupExpired>(_projectionManager);
         mainBus.Subscribe<ProjectionManagementMessage.Internal.RegularTimeout>(_projectionManager);
         mainBus.Subscribe<ProjectionManagementMessage.Internal.Deleted>(_projectionManager);
         mainBus.Subscribe<ProjectionManagementMessage.RegisterSystemProjection>(_projectionManager);
         mainBus.Subscribe<CoreProjectionManagementMessage.Started>(_projectionManager);
         mainBus.Subscribe<CoreProjectionManagementMessage.Stopped>(_projectionManager);
         mainBus.Subscribe<CoreProjectionManagementMessage.Faulted>(_projectionManager);
         mainBus.Subscribe<CoreProjectionManagementMessage.Prepared>(_projectionManager);
         mainBus.Subscribe<CoreProjectionManagementMessage.StateReport>(_projectionManager);
         mainBus.Subscribe<CoreProjectionManagementMessage.ResultReport>(_projectionManager);
         mainBus.Subscribe<CoreProjectionManagementMessage.StatisticsReport>(_projectionManager);
         mainBus.Subscribe<CoreProjectionManagementMessage.SlaveProjectionReaderAssigned>(_projectionManager);
     }
     mainBus.Subscribe<ClientMessage.WriteEventsCompleted>(_projectionManager);
     mainBus.Subscribe<ClientMessage.ReadStreamEventsBackwardCompleted>(_projectionManager);
 }

 public void Remove(ISubscriber subscriber)
 {
     if (subscribers.ContainsKey(subscriber.GetType()))
     {
         subscribers.Remove(subscriber.GetType());
     }
 }

 public void Subscribe(ISubscriber subscriber)
 {
     if (!_subscribers.Contains(subscriber))
     {
         _subscribers.Add(subscriber);
     }
 }

 void IPublisher.subscribe(ISubscriber subscriber, IList<int> channelsToSubscribe)
 {
     int[] subscribedChannels = subscriberChannels.GetOrAdd(subscriber, getInitialArray(this.channelsCapacity));
     foreach (var channel in channelsToSubscribe)
     {
         if (0 <= channel && channel < channelSubscribers.Length - 1 && subscribedChannels[channel] == -1)
         {
             for (int i = 0; i < this.subscribersCapacity; i++)
             {
                 if (channelSubscribers[channel][i] == null)
                 {
                     locks[channel][i].EnterWriteLock();
                     try
                     {
                         subscribedChannels[channel] = channel;
                         if (channelSubscribers[channel][i] == null)
                         {
                             channelSubscribers[channel][i] = subscriber;
                             break;
                         }
                     }
                     finally
                     {
                         locks[channel][i].ExitWriteLock();
                     }
                 }
             }
         }
     }
 }

 public void Add(ISubscriber subscriber)
 {
     if (!subscribers.ContainsKey(subscriber.GetType()))
     {
         subscribers.Add(subscriber.GetType(), subscriber);
     }
 }

 public void UnSubscribe(ISubscriber subscriber)
 {
     if (_subscribers.Contains(subscriber))
     {
         _subscribers.Remove(subscriber);
     }
 }

 public LunchFilter(IFiber fiber,
     ISubscriber<SimpleYouTubeClip> input,
     IPublisher<SimpleYouTubeClip> output)
 {
     this.output = output;
     this.subscription  = input.Subscribe(fiber, NewClip);
 }

        public EventStoreEmbeddedNodeConnection(ConnectionSettings settings, string connectionName, IPublisher publisher, ISubscriber bus, IAuthenticationProvider authenticationProvider)
        {
            Ensure.NotNull(publisher, "publisher");
            Ensure.NotNull(settings, "settings");

            Guid connectionId = Guid.NewGuid();

            _settings = settings;
            _connectionName = connectionName;
            _publisher = publisher;
            _authenticationProvider = authenticationProvider;
            _subscriptionBus = new InMemoryBus("Embedded Client Subscriptions");
            _subscriptions = new EmbeddedSubscriber(_subscriptionBus, _authenticationProvider, _settings.Log, connectionId);
            
            _subscriptionBus.Subscribe<ClientMessage.SubscriptionConfirmation>(_subscriptions);
            _subscriptionBus.Subscribe<ClientMessage.SubscriptionDropped>(_subscriptions);
            _subscriptionBus.Subscribe<ClientMessage.StreamEventAppeared>(_subscriptions);
            _subscriptionBus.Subscribe<ClientMessage.PersistentSubscriptionConfirmation>(_subscriptions);
            _subscriptionBus.Subscribe<ClientMessage.PersistentSubscriptionStreamEventAppeared>(_subscriptions);
            _subscriptionBus.Subscribe(new AdHocHandler<ClientMessage.SubscribeToStream>(_publisher.Publish));
            _subscriptionBus.Subscribe(new AdHocHandler<ClientMessage.UnsubscribeFromStream>(_publisher.Publish));
            _subscriptionBus.Subscribe(new AdHocHandler<ClientMessage.ConnectToPersistentSubscription>(_publisher.Publish));

            bus.Subscribe(new AdHocHandler<SystemMessage.BecomeShutdown>(_ => Disconnected(this, new ClientConnectionEventArgs(this, new IPEndPoint(IPAddress.None, 0)))));
        }

 public void Subscribe(ISubscriber subscriber)
 {
     if(_subscribers.Contains(subscriber) == false)
     {
         _subscribers.Add(subscriber);
     }
 }

        private void TestMassivePublish(ISubscriber conn, string caption)
        {
            const int loop = 100000;

            GC.Collect(GC.MaxGeneration, GCCollectionMode.Forced);
            GC.WaitForPendingFinalizers();

            var tasks = new Task[loop];

            var withFAF = Stopwatch.StartNew();
            for (int i = 0; i < loop; i++)
                conn.Publish("foo", "bar", CommandFlags.FireAndForget);
            withFAF.Stop();

            GC.Collect(GC.MaxGeneration, GCCollectionMode.Forced);
            GC.WaitForPendingFinalizers();

            var withAsync = Stopwatch.StartNew();
            for (int i = 0; i < loop; i++)
                tasks[i] = conn.PublishAsync("foo", "bar");
            conn.WaitAll(tasks);
            withAsync.Stop();

            Assert.Less(1, 2, "sanity check");
            Assert.Less(withFAF.ElapsedMilliseconds, withAsync.ElapsedMilliseconds, caption);
            Console.WriteLine("{2}: {0}ms (F+F) vs {1}ms (async)",
                withFAF.ElapsedMilliseconds, withAsync.ElapsedMilliseconds, caption);
        }

        public MainWindow(ISubscriber<PlayerCommand> commandChannel, IPublisher<RichYouTubeClip> finishedChannel)
        {
            InitializeComponent();
            this.finishedChannel = finishedChannel;

            finishedTimer = new Timer(TimerExpired);

            progressTimer = new System.Timers.Timer(1000);
            progressTimer.AutoReset = true;
            progressTimer.Elapsed += new System.Timers.ElapsedEventHandler(progressTimer_Elapsed);

            var fiber     = new DispatcherFiber(Dispatcher);
            commandChannel.Subscribe(fiber, OnPlayerCommand);
            fiber.Start();

            WindowStyle = WindowStyle.None;
            WindowState = WindowState.Maximized;
            ShowInTaskbar = false;

            Cursor = Cursors.None;
            try
            {
                tv.Initialize();
            }
            catch (Exception ex)
            {
                MessageBox.Show(ex.Message, "Error initialize remote control", MessageBoxButton.OK, MessageBoxImage.Error);
            }
        }

 public void Register(
     TFChunkDb db, QueuedHandler mainQueue, ISubscriber mainBus, TimerService timerService,
     ITimeProvider timeProvider, IHttpForwarder httpForwarder, HttpService[] httpServices, IPublisher networkSendService)
 {
     _projections = new EventStore.Projections.Core.Projections(
         db, mainQueue, mainBus, timerService, timeProvider, httpForwarder, httpServices, networkSendService,
         projectionWorkerThreadCount: _projectionWorkerThreadCount, runProjections: _runProjections);
 }

 public PlaylistController(IFiber fiber,
     ISubscriber<RichYouTubeClip> newClips,
     ISubscriber<RichYouTubeClip> finishedPlaying,
     IPublisher<PlayerCommand> playerCommands)
 {
     this.playerCommands = playerCommands;
     this.newClipSubscription  = newClips.Subscribe(fiber, NewClip);
     this.finishedSubscription = finishedPlaying.Subscribe(fiber, Finished);
 }

        public RedisSubscriberTests()
        {
            _subscriber = A.Fake<ISubscriber>();
            _remoteCache = A.Fake<ICacheAside>();
            _itemSerializer = A.Fake<IItemSerializer>();

            _connection = A.Fake<IConnectionMultiplexer>();
            A.CallTo(() => _connection.GetSubscriber(A<object>._)).Returns(_subscriber);
        }

        public RedisConnection(IDatabase redis, ISubscriber subscriber, string jobStorageIdentity)
        {
            _subscriber = subscriber;
            _jobStorageIdentity = jobStorageIdentity;
            Redis = redis;
            mre = new ManualResetEvent(false);

            _subscriber.Subscribe(string.Format("{0}JobFetchChannel", RedisStorage.Prefix), (channel, val) => { mre.Set();});
        }

        public Projections(
            TFChunkDb db, QueuedHandler mainQueue, ISubscriber mainBus, TimerService timerService, ITimeProvider timeProvider,
            IHttpForwarder httpForwarder, HttpService[] httpServices, IPublisher networkSendQueue,
            int projectionWorkerThreadCount, RunProjections runProjections)
        {
            _projectionWorkerThreadCount = projectionWorkerThreadCount;
            SetupMessaging(
                db, mainQueue, mainBus, timerService, timeProvider, httpForwarder, httpServices, networkSendQueue,
                runProjections);

        }

        public RedisSubscribe(ISubscriber Subscriber, string Prefix)
        {
            Channel = Prefix + "announce";
            _Subscriber = Subscriber;
            _Event = new ManualResetEvent(false);

            _Subscriber.Subscribe(Channel, delegate
            {
                _Event.Set();
            });
        }

 public void RemoveAll(ISubscriber subscriber)
 {
     lock (_registering)
     {
         RegistryEntry entry;
         if (_bySubscriber.TryGetValue(subscriber, out entry))
         {
             _byGuid.Remove(entry.Guid);
             removeByType(entry);
         }
     }
 }

        public PersistentSubscriptionConsumerStrategyRegistry(IPublisher mainQueue, ISubscriber mainBus, IPersistentSubscriptionConsumerStrategyFactory[] additionalConsumerStrategies)
        {
            _mainQueue = mainQueue;
            _mainBus = mainBus;
            Register(new DelegatePersistentSubscriptionConsumerStrategyFactory(SystemConsumerStrategies.RoundRobin, (subId, queue, bus) => new RoundRobinPersistentSubscriptionConsumerStrategy()));
            Register(new DelegatePersistentSubscriptionConsumerStrategyFactory(SystemConsumerStrategies.DispatchToSingle, (subId, queue, bus) => new DispatchToSinglePersistentSubscriptionConsumerStrategy()));
            Register(new DelegatePersistentSubscriptionConsumerStrategyFactory(SystemConsumerStrategies.Pinned, (subId, queue, bus) => new PinnedPersistentSubscriptionConsumerStrategy(new XXHashUnsafe())));

            foreach (var consumerStrategyFactory in additionalConsumerStrategies)
            {
                Register(consumerStrategyFactory);
            }
        }