private static IDbConnectionProvider InitDb(DirectoryInfo directory, QueueName queueName)
        {
            var dbPath = Path.Combine(directory.FullName, queueName + ".db");
            var connectionStringSettings = new ConnectionStringSettings
            {
                Name = dbPath,
                ConnectionString = "Data Source=" + dbPath + "; Version=3",
                ProviderName = "System.Data.SQLite"
            };

            var connectionProvider = new SingletonConnectionProvider(connectionStringSettings);
            var connection = connectionProvider.GetConnection();
            try
            {
                using (var command = connection.CreateCommand())
                {
                    command.CommandType = CommandType.Text;
                    command.CommandText = new SQLiteDialect().CreateMessageQueueingServiceObjectsCommand;
                    command.ExecuteNonQuery();
                }
            }
            finally
            {
                connectionProvider.ReleaseConnection(connection);
            }
            return connectionProvider;
        }

        public Task<PollerResult<Event>> NextAsync(QueueName name)
        {
            var connection = _connectionProvider.GetConnection();
            var result = new PollerResult<Event>(false , Event.Empty);
            using (var channel = connection.CreateModel())
            {
                var consumer = new QueueingBasicConsumer(channel);
                channel.BasicConsume(name.SubscriptionName, true, consumer);
                BasicDeliverEventArgs eventArgs = null;
                if (consumer.Queue.Dequeue((int)_longPollingTimeout.TotalMilliseconds, 
                    out eventArgs))
                {
                    var @event = new Event()
                    {
                        Body = Encoding.UTF8.GetString(eventArgs.Body),
                        QueueName = name.SubscriptionName,
                        UnderlyingMessage = eventArgs,
                        ContentType = "text/plain", // should it be JSON?
                        EventType = name.SubscriptionName
                    };

                    result = new PollerResult<Event>(true, @event);
                }
                return Task.FromResult(result);
            }

        }

        public async Task PushBatchAsync(IEnumerable<Event> messages)
        {

            const int BatchSize = 50;
            
            var msgs = messages.ToArray();
            if (!msgs.Any())
                return;

            var message = msgs.First();
            var queueName = new QueueName(message.QueueName);
            int i = 0;

            if (queueName.IsSimpleQueue)
            {
                var client = _clientProvider.GetQueueClient(queueName);
                while (i < msgs.Length)
                {
                    await client.SendBatchAsync(msgs.Skip(i).Take(BatchSize).Select(x => x.ToMessage()));
                    i += BatchSize;
                }
            }
            else
            {
                var client = _clientProvider.GetTopicClient(queueName);
                while (i < msgs.Length)
                {
                    await client.SendBatchAsync(msgs.Skip(i).Take(BatchSize).Select(x => x.ToMessage()));
                    i += BatchSize;
                }
            }
        }

 public RouteRegistration(string messageFilter, QueueName destination)
 {
     Condition.Requires(messageFilter, "messageFilter").IsNotNullOrEmpty();
       Condition.Requires(destination, "destination").IsNotNull();
       MessageFilter = messageFilter;
       Destination = destination;
 }

 public SubscribeCommand(Type subscribedMessageType, QueueName subscriberQueue)
 {
     Condition.Requires(subscribedMessageType, "subscribedMessageType").IsNotNull();
       Condition.Requires(subscriberQueue, "subscriberQueue").IsNotNull();
       SubscribedMessagesTypeName = subscribedMessageType.AssemblyQualifiedName;
       SubscriberQueueName = subscriberQueue.Name;
 }

 /// <summary>
 /// Initializes a new instance of the <see cref="T:System.Object"/> class.
 /// </summary>
 public MsmqQueueObserver(string serviceBusName, QueueName inputQueue, QueueName errorQueue, MsmqQueueTransport queueTransport)
 {
     _serviceBusName = serviceBusName;
     _inputQueue = inputQueue;
     _errorQueue = errorQueue;
     _queueTransport = queueTransport;
 }

        public SQLMessageQueue(IDbConnectionProvider connectionProvider, ISQLDialect dialect, QueueName queueName,
            IQueueListener listener, QueueOptions options = default(QueueOptions))
        {
            if (connectionProvider == null) throw new ArgumentNullException("connectionProvider");
            if (dialect == null) throw new ArgumentNullException("dialect");
            if (queueName == null) throw new ArgumentNullException("queueName");
            if (listener == null) throw new ArgumentNullException("listener");

            _connectionProvider = connectionProvider;
            _dialect = dialect;
            _queueName = queueName;

            _listener = listener;
            _autoAcknowledge = options.AutoAcknowledge;
            _maxAttempts = options.MaxAttempts <= 0 ? 10 : options.MaxAttempts;
            _retryDelay = options.RetryDelay < TimeSpan.Zero ? TimeSpan.Zero : options.RetryDelay;

            var concurrencyLimit = options.ConcurrencyLimit <= 0
                ? QueueOptions.DefaultConcurrencyLimit
                : options.ConcurrencyLimit;
            _concurrentMessageProcessingSlot = new SemaphoreSlim(concurrencyLimit);

            _cancellationTokenSource = new CancellationTokenSource();
            _queuedMessages = new BufferBlock<SQLQueuedMessage>(new DataflowBlockOptions
            {
                CancellationToken = _cancellationTokenSource.Token
            });
        }

        /// <summary>
        /// Stages indexes. Should run only using one process.
        /// </summary>
        /// <param name="scope"></param>
        /// <param name="documentType"></param>
        /// <param name="rebuild"></param>
        /// <exception cref="System.ArgumentNullException">scope</exception>
        public void Prepare(string scope, string documentType = "", bool rebuild = false)
        {
            if (String.IsNullOrEmpty(scope))
                throw new ArgumentNullException("scope");

            foreach (var builder in _indexBuilders)
            {
                // skip not requested indexers or index using all if index is not specified
                if (!String.IsNullOrEmpty(documentType) && !documentType.Equals(builder.DocumentType))
                    continue;

                // Execute builder, which will create partitions and put them in the queue
                var queueName = new QueueName("index-{0}-{1}-in", scope, builder.DocumentType);

                var config = GetBuildConfig(_repository, queueName.Scope, queueName.DocumentType);

                var lastBuild = DateTime.UtcNow;
                var newBuildDate = lastBuild;
                if (config.Status == BuildStatus.NeverStarted.GetHashCode() || rebuild) // build was never started, so set min date
                {
                    rebuild = true;
                    lastBuild = DateTime.MinValue;
                    config.LastBuildDate = DateTime.UtcNow.AddYears(-30); // have to set the date to something repository won't complain
                }
                else
                {
                    lastBuild = config.LastBuildDate.AddSeconds(-30); // make sure we get all the changes 
                }

                // Delete all the records
                if (rebuild)
                {
                    _searchProvider.RemoveAll(queueName.Scope, queueName.DocumentType);
                }

                var partitions = builder.CreatePartitions(queueName.Scope, lastBuild);

                var newPartitionsExist = false; // tells if there are any partitions that has been processed
                foreach (var partition in partitions)
                {
                    newPartitionsExist = true;
                    //_observer.Notify(new ConsumeBegin(msg, consumer, envelope.QueueName));
                   _messageSender.Send(queueName.ToString(), partition);
                }

                var newBuildStatus = BuildStatus.Started;
                if (newPartitionsExist)
                {
                    _messageSender.Send(queueName.ToString(), new SearchIndexStatusMessage(queueName.Scope, queueName.DocumentType, BuildStatus.Completed));
                }
                else
                {
                    newBuildStatus = BuildStatus.Completed;
                }

                config.LastBuildDate = newBuildDate;
                config.Status = newBuildStatus.GetHashCode();
                _repository.UnitOfWork.Commit();
            }
        }

        public RabbitMQQueue(QueueName queueName, IQueueListener listener, IConnection connection,
            Encoding encoding = null, QueueOptions options = default(QueueOptions))
        {
            if (queueName == null) throw new ArgumentNullException("queueName");
            if (listener == null) throw new ArgumentNullException("listener");
            if (connection == null) throw new ArgumentNullException("connection");

            _queueName = queueName;
            _queueExchange = _queueName.GetExchangeName();
            _retryQueueName = queueName.GetRetryQueueName();
            _retryExchange = _queueName.GetRetryExchangeName();
            _deadLetterExchange = _queueName.GetDeadLetterExchangeName();

            _listener = listener;
            _connection = connection;
            _encoding = encoding ?? Encoding.UTF8;
            _ttl = options.TTL;
            _maxAttempts = Math.Max(options.MaxAttempts, 1);
            _retryDelay = options.RetryDelay < TimeSpan.Zero ? TimeSpan.Zero : options.RetryDelay;
            _cancellationTokenSource = new CancellationTokenSource();

            var autoAcknowledge = options.AutoAcknowledge;
            var concurrencyLimit = Math.Max(options.ConcurrencyLimit, 1);
            _consumers = new DurableConsumer[concurrencyLimit];
            for (var i = 0; i < _consumers.Length; i++)
            {
                var consumerTag = _queueName + "_" + i;
                _consumers[i] = new DurableConsumer(_connection, queueName, HandleDelivery, consumerTag,
                    autoAcknowledge);
            }
        }

        public async Task Given_ClaimsPrincipal_When_Reading_Principal_Should_Be_Read()
        {
            var tempDir = GetTempDirectory();
            var queueName = new QueueName(Guid.NewGuid().ToString());
            var queuePath = Path.Combine(tempDir.FullName, queueName);
            var queueDir = new DirectoryInfo(queuePath);
            if (!queueDir.Exists)
            {
                queueDir.Create();
            }

            var message = new Message(new MessageHeaders
            {
                {HeaderName.ContentType, "text/plain"},
                {HeaderName.MessageId, Guid.NewGuid().ToString()}
            }, "Hello, world!");

            var senderPrincipal = new ClaimsPrincipal(new ClaimsIdentity(new[]
            {
                new Claim("username", "testuser"),
                new Claim("role", "testrole")
            }));

            var file = (await MessageFile.Create(queueDir, message, senderPrincipal)).File;
            var messageFile = new MessageFile(file);
            var readSenderPrincipal = await messageFile.ReadSenderPrincipal();
            var readMessage = await messageFile.ReadMessage();

            Assert.That(readSenderPrincipal, Is.EqualTo(senderPrincipal).Using(new ClaimsPrincipalEqualityComparer()));
            Assert.That(readMessage, Is.EqualTo(message).Using(new MessageEqualityComparer()));
        }

 public async Task DeleteQueueAsyncAwait()
 {
     #region DeleteQueueAsync (await)
     IQueueingService queueingService = new CloudQueuesProvider(identity, region, clientId, internalUrl, identityProvider);
     QueueName queueName = new QueueName("ExampleQueue");
     await queueingService.DeleteQueueAsync(queueName, CancellationToken.None);
     #endregion
 }

 public void CreateQueue()
 {
     #region CreateQueueAsync (TPL)
     IQueueingService queueingService = new CloudQueuesProvider(identity, region, clientId, internalUrl, identityProvider);
     QueueName queueName = new QueueName("ExampleQueue");
     Task<bool> task = queueingService.CreateQueueAsync(queueName, CancellationToken.None);
     #endregion
 }

 public static async Task PublishMessage(Message message, IPrincipal principal, IConnection connection,
     QueueName queueName, string exchange = "", Encoding encoding = null, int attempts = 0)
 {
     using (var channel = connection.CreateModel())
     {
         await PublishMessage(message, principal, channel, queueName, exchange, encoding, attempts);
     }
 }

 public Task CreateQueue(QueueName queueName, IQueueListener listener, QueueOptions options = default(QueueOptions), CancellationToken cancellationToken = default(CancellationToken))
 {
     if (!_queues.TryAdd(queueName, new InMemoryQueue(listener, options)))
     {
         throw new QueueAlreadyExistsException(queueName);
     }
     return Task.FromResult(true);
 }

 /// <summary>
 /// Initializes a new <see cref="HandlingRule"/> with the supplied message
 /// <paramref name="specification"/>, <see cref="MessageHandler"/>, and
 /// <paramref name="queueName"/>.
 /// </summary>
 /// <param name="specification">The message specification that selects messages 
 /// to which the handling rule applies</param>
 /// <param name="messageHandler">The handler to which messages matching the
 /// specification will be routed</param>
 /// <param name="queueName">(Optional) The name of the queue in which matching
 /// messages will be placed while they await handling</param>
 /// <remarks>
 /// If the <paramref name="queueName"/> is ommitted, a default queue name will
 /// be generated based on the MD5 hash of the full type name of the supplied
 /// <paramref name="messageHandler"/>
 /// </remarks>
 /// <seealso cref="GenerateQueueName"/>
 public HandlingRule(IMessageSpecification specification, IMessageHandler messageHandler,
     QueueName queueName = null)
 {
     if (specification == null) throw new ArgumentNullException("specification");
     if (messageHandler == null) throw new ArgumentNullException("messageHandler");
     _specification = specification;
     _messageHandler = messageHandler;
     _queueName = queueName ?? GenerateQueueName(messageHandler);
 }

        public async Task EnqueueMessage(QueueName queueName, Message message, IPrincipal senderPrincipal, CancellationToken cancellationToken = default(CancellationToken))
        {
            FilesystemMessageQueue queue;
            if (!_queues.TryGetValue(queueName, out queue)) throw new QueueNotFoundException(queueName);

            Log.DebugFormat("Enqueueing message ID {0} in filesystem queue \"{1}\"...", message.Headers.MessageId, queueName);
            await queue.Enqueue(message, senderPrincipal, cancellationToken);
            Log.DebugFormat("Message ID {0} enqueued successfully in filesystem queue \"{1}\"", message.Headers.MessageId, queueName);
        }

        public void AddRoute(string messageFilter, QueueName destination)
        {
            Condition.Requires(messageFilter, "messageFilter").IsNotNull();
              Condition.Requires(destination, "destination").IsNotNull();

              if (Routes.Any(r => r.MessageFilter == messageFilter && r.Destination == destination))
            throw new InvalidOperationException(string.Format("The subscription for {0} at {1} is already added.", messageFilter, destination));
              Routes.Add(new RouteRegistration(messageFilter, destination));
        }

 public Task EnqueueMessage(QueueName queueName, Message message, IPrincipal senderPrincipal, CancellationToken cancellationToken = default(CancellationToken))
 {
     InMemoryQueue queue;
     if (!_queues.TryGetValue(queueName, out queue))
     {
         throw new QueueNotFoundException(queueName);
     }
     return queue.Enqueue(message, senderPrincipal);
 }

 public string Name(QueueName name)
 {
     var val = _ht[name];
     if (val == null)
     {
         val = name.ToString("G");
         _ht[name] = val;
     }
     return val.ToString();
 }

        public async Task Given_Queued_Message_When_Acknowledged_Then_Message_Should_Be_Deleted()
        {
            var listenerCalledEvent = new ManualResetEvent(false);
            var tempDir = GetTempDirectory();
            var queueName = new QueueName(Guid.NewGuid().ToString());
            var queuePath = Path.Combine(tempDir.FullName, queueName);
            var queueDir = new DirectoryInfo(queuePath);
            if (!queueDir.Exists)
            {
                queueDir.Create();
            }

            var fsQueueingService = new FilesystemMessageQueueingService(tempDir);
            fsQueueingService.Init();

            var mockListener = new Mock<IQueueListener>();
            mockListener.Setup(
                x =>
                    x.MessageReceived(It.IsAny<Message>(), It.IsAny<IQueuedMessageContext>(),
                        It.IsAny<CancellationToken>()))
                .Callback<Message, IQueuedMessageContext, CancellationToken>((msg, ctx, ct) =>
                {
                    ctx.Acknowledge();
                    listenerCalledEvent.Set();
                })
                .Returns(Task.FromResult(true));

            await fsQueueingService.CreateQueue(queueName, mockListener.Object, new QueueOptions {MaxAttempts = 1});

            var message = new Message(new MessageHeaders
            {
                {HeaderName.ContentType, "text/plain"},
                {HeaderName.MessageId, Guid.NewGuid().ToString()}
            }, "Hello, world!");

            await fsQueueingService.EnqueueMessage(queueName, message, Thread.CurrentPrincipal);
            await listenerCalledEvent.WaitOneAsync(TimeSpan.FromSeconds(1));

            // The listener is called before the file is deleted, so there is a possible
            // race condition here.  Wait for a second to allow the delete to take place
            // before enumerating the files to see that they were actually deleted.
            await Task.Delay(TimeSpan.FromSeconds(1));

            var messageEqualityComparer = new MessageEqualityComparer();
            mockListener.Verify(
                x =>
                    x.MessageReceived(It.Is<Message>(m => messageEqualityComparer.Equals(m, message)),
                        It.IsAny<IQueuedMessageContext>(), It.IsAny<CancellationToken>()), Times.Once());

            var queuedMessages = queueDir.EnumerateFiles()
                .Select(f => new MessageFile(f))
                .ToList();

            Assert.That(queuedMessages, Is.Empty);
        }