public async void Init(MessageReceived messageReceivedHandler) {
            this.random = new Random();

            //ServiceBusEnvironment.SystemConnectivity.Mode = ConnectivityMode.AutoDetect;
            
            // Tcp mode does not work when I run in a VM (VirtualBox) and the host 
            // is using a wireless connection. Hard coding to Http.
            ServiceBusEnvironment.SystemConnectivity.Mode = ConnectivityMode.Http;

            string connectionString = CloudConfigurationManager.GetSetting("Microsoft.ServiceBus.ConnectionString");
            this.factory = MessagingFactory.CreateFromConnectionString(connectionString);
            this.namespaceManager = NamespaceManager.CreateFromConnectionString(connectionString);

            if (!namespaceManager.TopicExists(topicName)) {
                namespaceManager.CreateTopic(topicName);
            }

            this.subscriptionName = Guid.NewGuid().ToString();

            // Not needed really, it's a GUID...
            if (!namespaceManager.SubscriptionExists(topicName, subscriptionName)) {
                namespaceManager.CreateSubscription(topicName, subscriptionName);
            }

            this.topicClient = factory.CreateTopicClient(topicName);

            this.subClient = factory.CreateSubscriptionClient(topicName, subscriptionName);

            while (true) {
                await ReceiveMessageTaskAsync(messageReceivedHandler);
            }
        }

        public void InitialiseTopic()
        {
            ServicePointManager.DefaultConnectionLimit = 12;

            string connectionString = CloudConfigurationManager.GetSetting("ServiceBus.TopicConnectionString");
            _topicClient = SubscriptionClient.CreateFromConnectionString(connectionString, "recreateindex", "WebRoles");

            Task.Run(() =>
            {
                _topicClient.OnMessage(async receivedMessage =>
                {
                    var sequenceNumber = receivedMessage.SequenceNumber;
                    try
                    {
                        await _throttling.Execute(async () => ReCreateSearcher());
                    }
                    catch (Exception ex)
                    {
                        //no idea why it does not work but well, log it
                        Trace.TraceWarning("Exception occurred during the read of message '" + sequenceNumber + "': " + ex.Message);
                    }
                }, new OnMessageOptions {
                    AutoComplete = true
                });

                _completedEvent.WaitOne();
            });
        }

        private void InitServiceBusConnection()
        {
            string connectionString = CloudConfigurationManager.GetSetting("Microsoft.ServiceBus.ConnectionString");

            // Initialize clients
            _client = SubscriptionClient.CreateFromConnectionString(connectionString, "channel_sava", "twitter_follower_email");
        }

 public ReliableSubscriptionClient(string sbNamespace, TokenProvider tokenProvider, string path, string name, RetryPolicy<ServiceBusTransientErrorDetectionStrategy> policy)
     : base(sbNamespace, tokenProvider, path, policy)
 {
     //create the queue if it doesn't exist
     bool needsCreation = false;
     try
     {
         needsCreation = !mRetryPolicy.ExecuteAction<bool>(() => mNamespaceManager.SubscriptionExists(path,name));
     }
     catch (MessagingEntityNotFoundException)
     {
         needsCreation = true;
     }
     if (needsCreation)
     {
         try
         {
             mRetryPolicy.ExecuteAction<SubscriptionDescription>(() => mNamespaceManager.CreateSubscription(path,name));
         }
         catch (MessagingEntityAlreadyExistsException)
         {
             //ignore this exception because queue already exists
         }
     }
     mRetryPolicy.ExecuteAction(() => mSubscriptionClient = mMessagingFactory.CreateSubscriptionClient(path,name));
 }

 public static void Initialize()
 {
     if (null == OrdersTopicClient)
     {
         try
         {
             string connectionString = CloudConfigurationManager.GetSetting("Microsoft.ServiceBusTopics.ConnectionString");
             // string connectionString = RoleEnvironment.GetConfigurationSettingValue("Microsoft.ServiceBusTopics.ConnectionString");
             var namespaceManager = NamespaceManager.CreateFromConnectionString(connectionString);
             if (!namespaceManager.TopicExists("OrderTopic"))
             {
                 namespaceManager.CreateTopic("OrderTopic");
             }
             if (!namespaceManager.SubscriptionExists("OrderTopic", "OrderMessages"))
             {
                 namespaceManager.CreateSubscription("OrderTopic", "OrderMessages");
             }
             // Initialize the connection to Service Bus Topics
             OrdersTopicClient = TopicClient.CreateFromConnectionString(connectionString, "OrderTopic");
             SubClient = SubscriptionClient.CreateFromConnectionString(connectionString, "OrderTopic", "OrderMessages");
            
         }
         catch (Exception ex)
         {
             string str = ex.Message;
         }
     }
 }

        public override bool OnStart()
        {
            ServicePointManager.DefaultConnectionLimit = 12;

            var connectionString = CloudConfigurationManager.GetSetting("topicConnectionString");
            var topicName = CloudConfigurationManager.GetSetting("topicName");

            _nsMgr = NamespaceManager.CreateFromConnectionString(connectionString);

            if (!_nsMgr.TopicExists(topicName))
            {
                _nsMgr.CreateTopic(topicName);
            }

            if (!_nsMgr.SubscriptionExists(topicName, "audit"))
            {
                _nsMgr.CreateSubscription(topicName, "audit");
            }

            _client = SubscriptionClient.CreateFromConnectionString(connectionString, topicName, "audit", ReceiveMode.ReceiveAndDelete);

            var result = base.OnStart();

            Trace.TraceInformation("NTM.Auditing has been started");

            return result;
        }

        private async Task<SubscriptionClient> GetSubscriptionClient()
        {
            if (_subscriptionClient != null) return _subscriptionClient;

            _subscriptionClient = await _queueManager.CreateSubscriptionReceiver(_topicPath, _subscriptionName);
            _subscriptionClient.PrefetchCount = ConcurrentHandlerLimit;
            return _subscriptionClient;
        }

        private async Task<SubscriptionClient> GetSubscriptionClient()
        {
            if (_subscriptionClient != null) return _subscriptionClient;

            _subscriptionClient = await _queueManager.CreateSubscriptionReceiver(_topicPath, _subscriptionName);
            _subscriptionClient.PrefetchCount = _prefetchCountSetting;
            return _subscriptionClient;
        }

		public SubscriptionObserver Create(IServiceBus bus, SubscriptionRouter router)
		{
			string path = bus.ControlBus.Endpoint.Address.Uri.AbsolutePath;


			var client = new SubscriptionClient(bus, router, _subscriptionServiceUri, _timeout);

			return client;
		}

        public void CanUnregisterProviderIfRegistered()
        {
            var mockHandler = CreateMockHandler(() => CreateResponseMessageWithStatus(HttpStatusCode.OK));

            ISubscriptionClient client = new SubscriptionClient(subscriptionData, mockHandler);
            bool worked = client.UnregisterResourceType("someResource");

            Assert.IsTrue(worked);
        }

        public async Task Subscribe(Subscriber subscriber, MessageType messageType, ContextBag context)
        {
            //When the subscriber is running V6 and UseLegacyMessageDrivenSubscriptionMode is enabled at the subscriber the 'subcriber.Endpoint' value is null
            var endpoint = subscriber.Endpoint ?? subscriber.TransportAddress.Split('@').First();
            var subscriptionClient = new SubscriptionClient { TransportAddress = subscriber.TransportAddress, Endpoint = endpoint };

            var attempts = 0;

            //note: since we have a design that can run into concurrency exceptions we perform a few retries
            // we should redesign this in the future to use a separate doc per subscriber and message type
            do
            {
                try
                {
                    using (var session = OpenAsyncSession())
                    {
                        var subscriptionDocId = Subscription.FormatId(messageType);

                        var subscription = await session.LoadAsync<Subscription>(subscriptionDocId).ConfigureAwait(false);

                        if (subscription == null)
                        {
                            subscription = new Subscription
                            {
                                Id = subscriptionDocId,
                                MessageType = messageType,
                                Subscribers = new List<SubscriptionClient>()
                            };

                            await session.StoreAsync(subscription).ConfigureAwait(false);
                        }

                        if (!subscription.Subscribers.Contains(subscriptionClient))
                        {
                            subscription.Subscribers.Add(subscriptionClient);
                        }
                        else
                        {
                            var savedSubscription = subscription.Subscribers.Single(s => s.Equals(subscriptionClient));
                            if (savedSubscription.Endpoint != subscriber.Endpoint)
                            {
                                savedSubscription.Endpoint = subscriber.Endpoint;
                            }
                        }

                        await session.SaveChangesAsync().ConfigureAwait(false);
                    }

                    return;
                }
                catch (ConcurrencyException)
                {
                    attempts++;
                }
            } while (attempts < 5);
        }

        private void DiscardSubscriptionClient()
        {
            var subscriptionClient = _subscriptionClient;
            _subscriptionClient = null;

            if (subscriptionClient == null) return;
            if (subscriptionClient.IsClosed) return;

            subscriptionClient.Close();
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="address"></param>
        /// <param name="tryProcessMessage"></param>
        public void Start(Address address, Action<BrokeredMessage> tryProcessMessage)
        {
            _cancelRequested = false;

            _tryProcessMessage = tryProcessMessage;

            _subscriptionClient = SubscriptionClientCreator.Create(address, EventType);

            if (_subscriptionClient != null) _subscriptionClient.BeginReceiveBatch(BatchSize, TimeSpan.FromSeconds(ServerWaitTime), OnMessage, null);
        }

        public AzureServiceBusEventDispatcherReceiver(string connectionString, IEventDispatcher innerEventDispatcher, IEventStore eventStore, string topicName, string subscriptionName)
        {
            _innerEventDispatcher = innerEventDispatcher;
            _eventStore = eventStore;

            AzureHelpers.EnsureTopicExists(connectionString, topicName);
            AzureHelpers.EnsureSubscriptionExists(connectionString, topicName, subscriptionName);

            _subscriptionClient = SubscriptionClient.CreateFromConnectionString(connectionString, topicName, subscriptionName);

            _workerThread = new Thread(DoWork);
        }

        public void CanGetListOfRegisteredProviders()
        {
            string[] knownResourceTypes = {"website", "mobileservice"};

            var mockHandler = CreateMockHandler(() => CreateListResourcesResponseMessage(
                new ProviderResource {Type = "Website", State = "Unregistered"},
                new ProviderResource {Type = "Mobileservice", State = "Registered"}
                ));

            ISubscriptionClient client = new SubscriptionClient(subscriptionData, mockHandler);
            IEnumerable<ProviderResource> actualResourceTypes = client.ListResources(knownResourceTypes);

            CollectionAssert.AreEquivalent(knownResourceTypes, actualResourceTypes.Select(rt => rt.Type.ToLower()).ToList());
        }

        //private SubscriptionClient _subscriptionClient = null;
        //private string _entityPath = null;
        //public QueueCacheDependency(string key, string guid, TokenProvider tokenProvider, Uri uri, string queueName)
        //{
        //    _key = key;
        //    _guid = guid;
        //    //_tokenProvider = tokenProvider;
        //    //_uri = uri;
        //    //_entityPath = entityPath;
        //    var factory = MessagingFactory.Create(uri, tokenProvider);
        //    _receiver = factory.CreateMessageReceiver(queueName);
        //    WaitCallback callback = new WaitCallback(WaitForMessage);
        //    ThreadPool.QueueUserWorkItem(callback);
        //}
        public QueueCacheDependency(string key, SubscriptionClient subscriptionClient, string crc)
        {
            Trace.TraceInformation("QueueCacheDependency created: {0}", key);
            //_subscriptionClient = subscriptionClient;
            //_guid = guid;
            _key = key;
            _crc = crc;

            subscriptionClient.BeginReceive(
                TimeSpan.MaxValue,
                ReceiveDone,
                subscriptionClient
            );
        }

        public EventTopicSubscriber(string connectionString, string topicName, string subscriptionName, 
            OnMessageOptions options, IEventDispatcher eventDispatcher, ILogger logger)
        {
            var namespaceManager = NamespaceManager.CreateFromConnectionString(connectionString);

            if (!namespaceManager.SubscriptionExists(topicName, subscriptionName))
                namespaceManager.CreateSubscription(topicName, subscriptionName);

            _client = SubscriptionClient.CreateFromConnectionString(connectionString, topicName, subscriptionName);
            _options = options;
            _eventDispatcher = eventDispatcher;
            _logger = logger;
            _stopEvent = new ManualResetEvent(false);
        }

        public given_a_topic_sender()
        {
            this.sut = new TestableTopicSender(this.Settings, this.Topic, new Incremental(1, TimeSpan.Zero, TimeSpan.Zero));

            var tokenProvider = TokenProvider.CreateSharedSecretTokenProvider(this.Settings.TokenIssuer, this.Settings.TokenAccessKey);
            var serviceUri = ServiceBusEnvironment.CreateServiceUri(this.Settings.ServiceUriScheme, this.Settings.ServiceNamespace, this.Settings.ServicePath);

            var manager = new NamespaceManager(serviceUri, tokenProvider);
            manager.CreateSubscription(this.Topic, "Test");

            var messagingFactory = MessagingFactory.Create(serviceUri, tokenProvider);
            this.subscriptionClient = messagingFactory.CreateSubscriptionClient(this.Topic, "Test");

        }

        public ServiceBusMessageBus(string connectionString, string topicName, ISerializer serializer = null) {
            _topicName = topicName;
            _serializer = serializer ?? new JsonNetSerializer();
            _subscriptionName = "MessageBus";
            _namespaceManager = NamespaceManager.CreateFromConnectionString(connectionString);
            if (!_namespaceManager.TopicExists(_topicName))
                _namespaceManager.CreateTopic(_topicName);

            _topicClient = TopicClient.CreateFromConnectionString(connectionString, _topicName);
            if (!_namespaceManager.SubscriptionExists(_topicName, _subscriptionName))
                _namespaceManager.CreateSubscription(_topicName, _subscriptionName);

            _subscriptionClient = SubscriptionClient.CreateFromConnectionString(connectionString, _topicName, _subscriptionName, ReceiveMode.ReceiveAndDelete);
            _subscriptionClient.OnMessageAsync(OnMessageAsync, new OnMessageOptions { AutoComplete = true });
        }

        public AzureServiceBusMessageBus(string connectionString, string topicName, ISerializer serializer = null, ILoggerFactory loggerFactory = null) : base(loggerFactory) {
            _serializer = serializer ?? new JsonNetSerializer();

            var namespaceManager = NamespaceManager.CreateFromConnectionString(connectionString);
            if (!namespaceManager.TopicExists(topicName))
                namespaceManager.CreateTopic(topicName);

            _topicClient = TopicClient.CreateFromConnectionString(connectionString, topicName);

            const string subscriptionName = "MessageBus";
            if (!namespaceManager.SubscriptionExists(topicName, subscriptionName))
                namespaceManager.CreateSubscription(topicName, subscriptionName);

            _subscriptionClient = SubscriptionClient.CreateFromConnectionString(connectionString, topicName, subscriptionName, ReceiveMode.ReceiveAndDelete);
            _subscriptionClient.OnMessageAsync(OnMessageAsync, new OnMessageOptions { AutoComplete = true });
        }