public static void Test(ITransport t, IQueueStrategy strategy,
            Endpoint queueEndpoint, Action<Message> send, Func<MessageEnumerator>
            enumer)
        {
            Guid id = Guid.NewGuid();
            var serializer = new XmlMessageSerializer(new
                                                          DefaultReflection(), new DefaultKernel());

            var subscriptionStorage = new MsmqSubscriptionStorage(new
                                                                      DefaultReflection(),
                                                                  serializer,
                                                                  queueEndpoint.Uri,
                                                                  new
                                                                      EndpointRouter(),
                                                                  strategy);
            subscriptionStorage.Initialize();

            var wait = new ManualResetEvent(false);

            subscriptionStorage.SubscriptionChanged += () => wait.Set();

            t.AdministrativeMessageArrived +=
                subscriptionStorage.HandleAdministrativeMessage;

            Message msg = new MessageBuilder
                (serializer).GenerateMsmqMessageFromMessageBatch(new
                                                                     AddInstanceSubscription
                {
                    Endpoint = queueEndpoint.Uri.ToString(),
                    InstanceSubscriptionKey = id,
                    Type = typeof (TestMessage2).FullName,
                });
            send(msg);

            wait.WaitOne();

            msg = new MessageBuilder
                (serializer).GenerateMsmqMessageFromMessageBatch(new
                                                                     RemoveInstanceSubscription
                {
                    Endpoint = queueEndpoint.Uri.ToString(),
                    InstanceSubscriptionKey = id,
                    Type = typeof (TestMessage2).FullName,
                });

            wait.Reset();

            send(msg);

            wait.WaitOne();

            IEnumerable<Uri> uris = subscriptionStorage
                .GetSubscriptionsFor(typeof (TestMessage2));
            Assert.Equal(0, uris.Count());

            int count = 0;
            MessageEnumerator copy = enumer();
            while (copy.MoveNext()) count++;
            Assert.Equal(0, count);
        }

 /// <summary>
 /// 初始化
 /// </summary>
 /// <param name="endpoint"></param>
 public RemoteServiceInterceptor(Endpoint endpoint)
     : this(endpoint
     , endpoint.Resolve<ILoggerFactory>()
     , endpoint.Resolve<ISerializer>()
     , endpoint.Resolve<ILoadBalancingHelper>())
 {
 }

        private async Task SendMessageAsync(Mocks.CloudBlobStorageProviderMock cloudStorage, Mocks.InboxHttpHandlerMock inboxMock, CryptoSettings senderCrypto, OwnEndpoint senderEndpoint, Endpoint receiverEndpoint)
        {
            Requires.NotNull(cloudStorage, "cloudStorage");
            Requires.NotNull(senderCrypto, "senderCrypto");
            Requires.NotNull(senderEndpoint, "senderEndpoint");
            Requires.NotNull(receiverEndpoint, "receiverEndpoint");

            var httpHandler = new Mocks.HttpMessageHandlerMock();

            cloudStorage.AddHttpHandler(httpHandler);

            inboxMock.Register(httpHandler);

            var sentMessage = Valid.Message;

            var channel = new Channel()
            {
                HttpClient = new HttpClient(httpHandler),
                CloudBlobStorage = cloudStorage,
                CryptoServices = senderCrypto,
                Endpoint = senderEndpoint,
                Logger = this.logger,
            };

            await channel.PostAsync(sentMessage, new[] { receiverEndpoint }, Valid.ExpirationUtc);
        }

        public static QueueInfo GetQueuePath(Endpoint endpoint)
        {
            var uri = endpoint.Uri;
            if (uri.AbsolutePath.IndexOf("/", 1) >= 0)
            {
                throw new InvalidOperationException(
                    "Invalid enpoint url : " + uri + Environment.NewLine +
                    "Queue Endpoints can't have a child folders (including 'public')" + Environment.NewLine +
                    "Good: 'msmq://machinename/queue_name'." + Environment.NewLine +
                    " Bad: msmq://machinename/round_file/queue_name"
                    );
            }

            string hostName = uri.Host;
            string queuePathWithFlatSubQueue =
                uri.AbsolutePath.Substring(1);
            if (string.IsNullOrEmpty(uri.Fragment) == false && uri.Fragment.Length > 1)
                queuePathWithFlatSubQueue += uri.Fragment;
            if (string.Compare(hostName, ".") == 0 ||
                string.Compare(hostName, Environment.MachineName, true) == 0 ||
                string.Compare(hostName, "localhost", true) == 0)
            {
                return new QueueInfo
                {
                    IsLocal = true,
                    QueuePath = Environment.MachineName + @"\private$\" + queuePathWithFlatSubQueue,
                    QueueUri = uri,
                    Transactional = endpoint.Transactional
                };
            }

            IPAddress address;
            if (IPAddress.TryParse(hostName, out address))
            {
                return new QueueInfo
                {
                    IsLocal = false,
                    QueuePath = "FormatName:DIRECT=TCP:" + hostName + @"\private$\" + queuePathWithFlatSubQueue,
					QueueUri = uri,
					Transactional = endpoint.Transactional
                };
            }
            if (guidRegEx.IsMatch(hostName))
            {
                return new QueueInfo
                {
                    IsLocal = false,
                    QueuePath = "FormatName:PRIVATE=" + hostName + @"\" + queuePathWithFlatSubQueue,
					QueueUri = uri,
					Transactional = endpoint.Transactional
                };  
            }
            return new QueueInfo
            {
                IsLocal = false,
                QueuePath = "FormatName:DIRECT=OS:" + hostName + @"\private$\" + queuePathWithFlatSubQueue,
				QueueUri = uri,
				Transactional = endpoint.Transactional
            };
        }

        public void EndpointSendAndReceive()
        {
            using (var management = new RabbitMqEndpointManagement(_queue))
            {
                management.BindQueue(_queue.Name, _exchange.Name, ExchangeType.Fanout, "", null);
            }

            IMessageSerializer serializer = new XmlMessageSerializer();

            var message = new BugsBunny {Food = "Carrot"};

            IDuplexTransport transport = _factory.BuildLoopback(new TransportSettings(_exchange));
            IOutboundTransport error = _factory.BuildError(new TransportSettings(_error));

            var sendEndpoint = new Endpoint(_exchange, serializer, transport, error,
                new InMemoryInboundMessageTracker(5));
            sendEndpoint.Send(message);


            var receiveEndpoint = new Endpoint(_queue, serializer, transport, error,
                new InMemoryInboundMessageTracker(5));
            receiveEndpoint.Receive(o =>
                {
                    return b =>
                        {
                            var bb = (BugsBunny)b;
                            Console.WriteLine(bb.Food);
                        };
                }, TimeSpan.Zero);
        }

        public HttpResponseMessage Put(Endpoint endpoint)
        {
            try
            {
                if (endpoint == null)
                    throw WebException(Constants.ErrorParameterEndpointCannotBeNull, HttpStatusCode.BadRequest);

                // Set the username under which the app will store the channel
                endpoint.UserId = this.mapUsernameDelegate(this.Request);

                if (endpoint.ExpirationTime == null || endpoint.ExpirationTime == DateTime.MinValue)
                    endpoint.ExpirationTime = DateTime.UtcNow.Add(defaultEndpointExpirationTime);

                this.Repository.InsertOrUpdate(endpoint);
                return new HttpResponseMessage(HttpStatusCode.Accepted);
            }
            catch (HttpResponseException)
            {
                throw;
            }
            catch (Exception exception)
            {
                throw WebException(exception.Message, HttpStatusCode.InternalServerError);
            }
        }

        public void Run()
        {
            while (true)
            {
                var url = Console.ReadLine();
                if (url == null)
                {
                    break;
                }

                url = url.Trim();
                if (!string.IsNullOrEmpty(url))
                {
                    try
                    {
                        var endPoint = new Endpoint(url);
                        var output = this.dispatcher.DispatchAction(endPoint);
                        Console.WriteLine(output);
                    }
                    catch (Exception ex)
                    {
                        Console.WriteLine(ex.Message);
                    }
                }
            }
        }

        public void Configure(AbstractRhinoServiceBusConfiguration config, IBusContainerBuilder builder, IServiceLocator locator)
        {
            var busConfig = config as RhinoServiceBusConfiguration;
            if (busConfig == null)
                return;

            var busElement = config.ConfigurationSection.Bus;

            if (busElement == null)
                return;

            var loadBalancerEndpointAsString = busElement.LoadBalancerEndpoint;

            if (string.IsNullOrEmpty(loadBalancerEndpointAsString))
                return;

            Uri loadBalancerEndpoint;
            if (!Uri.TryCreate(loadBalancerEndpointAsString, UriKind.Absolute, out loadBalancerEndpoint))
                throw new ConfigurationErrorsException(
                    "Attribute 'loadBalancerEndpoint' on 'bus' has an invalid value '" + loadBalancerEndpointAsString + "'");

            var endpoint = new Endpoint { Uri = loadBalancerEndpoint };
            builder.RegisterLoadBalancerEndpoint(endpoint.Uri);
            config.AddMessageModule<LoadBalancerMessageModule>();
        }

        public MessageQueue[] InitializeQueue(Endpoint queueEndpoint, QueueType queueType)
        {
        	var queue = MsmqUtil.GetQueuePath(queueEndpoint).Create();
        	switch (queueType)
            {
                case QueueType.Standard:
                    return new[]
	                {
	                    queue,
	                    MsmqUtil.OpenOrCreateQueue(GetErrorsQueuePath(), QueueAccessMode.SendAndReceive, queue),
	                    MsmqUtil.OpenOrCreateQueue(GetSubscriptionQueuePath(), QueueAccessMode.SendAndReceive, queue),
	                    MsmqUtil.OpenOrCreateQueue(GetDiscardedQueuePath(), QueueAccessMode.SendAndReceive, queue),
	                    MsmqUtil.OpenOrCreateQueue(GetTimeoutQueuePath(), QueueAccessMode.SendAndReceive, queue),
	                };
                case QueueType.LoadBalancer:
                    return new[]
	                {
	                    queue,
	                    MsmqUtil.OpenOrCreateQueue(GetKnownWorkersQueuePath(), QueueAccessMode.SendAndReceive, queue),
	                    MsmqUtil.OpenOrCreateQueue(GetKnownEndpointsQueuePath(), QueueAccessMode.SendAndReceive, queue),
	                };
                case QueueType.Raw:
                    return new[]
	                {
	                    queue,
	                };
                default:
                    throw new ArgumentOutOfRangeException("queueType", "Can't handle queue type: " + queueType);
            }
        }

 public Binding Create(Endpoint serviceInterface)
 {
     System.Net.ServicePointManager.ServerCertificateValidationCallback =
         ((sender, certificate, chain, sslPolicyErrors) => true);
     var secureBinding = new WS2007FederationHttpBinding(WSFederationHttpSecurityMode.TransportWithMessageCredential)
                             {
                                 Name = "secure",
                                 TransactionFlow = false,
                                 HostNameComparisonMode = serviceInterface.HostNameComparisonMode.ParseAsEnum(HostNameComparisonMode.StrongWildcard),
                                 MaxBufferPoolSize = serviceInterface.MaxBufferPoolSize,
                                 MaxReceivedMessageSize = serviceInterface.MaxReceivedSize,
                                 MessageEncoding = serviceInterface.MessageFormat.ParseAsEnum(WSMessageEncoding.Text),
                                 TextEncoding = Encoding.UTF8,
                                 ReaderQuotas = XmlDictionaryReaderQuotas.Max,
                                 BypassProxyOnLocal = true,
                                 UseDefaultWebProxy = false
                             };
     if (ConfigurationManagerHelper.GetValueOnKey("stardust.UseDefaultProxy")=="true")
     {
         secureBinding.BypassProxyOnLocal = false;
         secureBinding.UseDefaultWebProxy = true; 
     }
     SetSecuritySettings(serviceInterface, secureBinding);
     return secureBinding;
 }

 private EndpointDetails(Endpoint endpoint)
 {
     Id = endpoint.Identity.Id;
     Address = endpoint.Identity.Address;
     MonitorType = endpoint.Identity.MonitorType;
     Name = endpoint.Metadata.Name;
     Group = endpoint.Metadata.Group;
     var health = endpoint.Health;
     LastModifiedTime = endpoint.LastModifiedTimeUtc;
     if (health != null)
     {
         Status = health.Status;
         LastCheckUtc = health.CheckTimeUtc;
         LastResponseTime = health.ResponseTime;
         Details = health.Details.ToDictionary(p => p.Key, p => p.Value);
     }
     else
     {
         Details = new Dictionary<string, string>();
         Status = EndpointStatus.NotRun;
     }
     Tags = endpoint.Metadata.Tags;
     RegisteredOnUtc = endpoint.Metadata.RegisteredOnUtc;
     RegistrationUpdatedOnUtc = endpoint.Metadata.RegistrationUpdatedOnUtc;
     MonitorTag = endpoint.Metadata.MonitorTag;
 }

		public void DefaultContactCtor() {
			var contact = new Endpoint();
			Assert.That(contact.MessageReceivingEndpoint, Is.Null);
			Assert.That(contact.EncryptionKeyPublicMaterial, Is.Null);
			Assert.That(contact.SigningKeyPublicMaterial, Is.Null);
			Assert.That(contact.CreatedOnUtc, Is.EqualTo(DateTime.UtcNow).Within(TimeSpan.FromMinutes(1)));
		}

        public MsmqFlatQueueTestBase()
        {
            defaultTestBase = new MsmqTestBase();

            testQueueEndPoint = new Endpoint
            {
                Uri = new Uri("msmq://localhost/test_queue")
            };
            testQueuePath = MsmqUtil.GetQueuePath(testQueueEndPoint).QueuePath;

            transactionalTestQueueEndpoint = new Endpoint
            {
                Uri = new Uri("msmq://localhost/transactional_test_queue")
            };
            transactionalTestQueuePath = MsmqUtil.GetQueuePath(transactionalTestQueueEndpoint).QueuePath;

            subscriptionsEndpoint = new Endpoint
            {
                Uri = new Uri(testQueueEndPoint.Uri + "#" + subscriptions)
            };
            subscriptionQueuePath = MsmqUtil.GetQueuePath(subscriptionsEndpoint).QueuePath;

            SetupQueues();

            queue = MsmqUtil.GetQueuePath(testQueueEndPoint).Open();
            transactionalQueue = MsmqUtil.GetQueuePath(transactionalTestQueueEndpoint).Open();
            subscriptions = MsmqUtil.GetQueuePath(subscriptionsEndpoint).Open(QueueAccessMode.SendAndReceive,
                new XmlMessageFormatter(new[]
                {
                    typeof (string)
                }));
        }

 public Binding Create(Endpoint serviceInterface)
 {
     var binding = new BasicHttpBinding
                       {
                           Name = serviceInterface.EndpointName,
                           AllowCookies = false,
                           HostNameComparisonMode = serviceInterface.HostNameComparisonMode.ParseAsEnum(HostNameComparisonMode.StrongWildcard),
                           MaxBufferPoolSize = serviceInterface.MaxBufferPoolSize,
                           MaxReceivedMessageSize = serviceInterface.MaxReceivedSize,
                           MessageEncoding = serviceInterface.MessageFormat.ParseAsEnum(WSMessageEncoding.Mtom),
                           MaxBufferSize = serviceInterface.MaxBufferSize,
                           TransferMode = TransferMode.Streamed,
                           TextEncoding = Encoding.UTF8,
                           ReaderQuotas = XmlDictionaryReaderQuotas.Max,
                           BypassProxyOnLocal = true,
                           UseDefaultWebProxy = false,
                             
                       };
     if (ConfigurationManagerHelper.GetValueOnKey("stardust.UseDefaultProxy") == "true")
     {
         binding.BypassProxyOnLocal = false;
         binding.UseDefaultWebProxy = true;
     }
     return binding;
 }

        static void Main(string[] args)
        {
            LogFactory.BuildLogger = type => new ConsoleWindowLogger(type);
            ConsoleWindowLogger.MinimumLogLevel = LogLevel.Info;

            var endpoint = new Endpoint();
            endpoint.Start();

            using (var scope = Settings.RootContainer.BeginLifetimeScope())
            {
                var dtoCompanyQueryService = scope.GetInstance<DtoCompanyQueryService>();

                var dtoEmployeeQueryService = scope.GetInstance<DtoEmployeeQueryService>();

                var googlePage = dtoCompanyQueryService
                    .Query
                    .Where(c => c.Employees.Any(employee => employee.Name.Contains("Smith")))
                    .OrderBy(company => company.Name).FetchPage(1, 10);

                var firstCompanyDto = dtoCompanyQueryService.Query.First();
                var companiesWithMoreThanTwoEmployees = dtoCompanyQueryService.Query.Where(company => company.Employees.Count > 2).ToArray();

                var ordered = dtoCompanyQueryService.Query
                    .OrderByDescending(company => company.Employees.Count)
                    .ThenByDescending(company => company.Name)
                    .ToArray();

                dtoEmployeeQueryService.Query.FirstOrDefault(employee => employee.Name.Length > 2);
            }
        }

 public Binding Create(Endpoint serviceInterface)
 {
     if (Utilities.Utilities.GetDisableCertificateValidation().EqualsCaseInsensitive("true"))
         System.Net.ServicePointManager.ServerCertificateValidationCallback =
             ((sender, certificate, chain, sslPolicyErrors) => true);
     var binding = new BasicHttpBinding(BasicHttpSecurityMode.Transport)
     {
         Name = serviceInterface.EndpointName,
         AllowCookies = false,
         HostNameComparisonMode = serviceInterface.HostNameComparisonMode.ParseAsEnum(HostNameComparisonMode.StrongWildcard),
         MaxBufferPoolSize = serviceInterface.MaxBufferPoolSize,
         MaxReceivedMessageSize = serviceInterface.MaxReceivedSize,
         MessageEncoding = serviceInterface.MessageFormat.ParseAsEnum(WSMessageEncoding.Text),
         TextEncoding = Encoding.UTF8,
         ReaderQuotas = XmlDictionaryReaderQuotas.Max,
         BypassProxyOnLocal = true,
         UseDefaultWebProxy = false
     };
     if (ConfigurationManagerHelper.GetValueOnKey("stardust.UseDefaultProxy") == "true")
     {
         binding.BypassProxyOnLocal = false;
         binding.UseDefaultWebProxy = true;
     }
     binding.Security.Transport.ClientCredentialType = HttpClientCredentialType.Windows;
     return binding;
 }

        public void SaveEndpoint(Endpoint endpoint)
        {
            using (var conn = _db.OpenConnection())
            using (var tx = conn.BeginTransaction())
            {
                var tags = endpoint.Metadata.Tags.ToDbString();

                conn.Execute(
                    $"replace into EndpointConfig(MonitorType, Address, GroupName, Name, Id, Password, RegisteredOnUtc, RegistrationUpdatedOnUtc, MonitorTag{(tags != null ? ", Tags" : "")}) values(@MonitorType,@Address,@Group,@Name,@Id,@Password,@RegisteredOnUtc,@RegistrationUpdatedOnUtc,@MonitorTag{(tags != null ? ",@Tags" : "")})",
                    new
                    {
                        endpoint.Identity.MonitorType,
                        endpoint.Identity.Address,
                        endpoint.Metadata.Group,
                        endpoint.Metadata.Name,
                        endpoint.Identity.Id,
                        endpoint.Password,
                        endpoint.Metadata.RegisteredOnUtc,
                        endpoint.Metadata.RegistrationUpdatedOnUtc,
                        endpoint.Metadata.MonitorTag,
                        tags
                    }, tx);

                tx.Commit();
            }
        }

 public MessageQueue[] InitializeQueue(Endpoint queueEndpoint, QueueType queueType)
 {
     var path = MsmqUtil.GetQueuePath(queueEndpoint);
     return new[]
     {
         path.Create()
     };
 }

 public void Report( Endpoint endpoint, EndpointStatus status )
 {
     var message = BuildMessage( endpoint, status );
     using ( var client = BuildSmtpClient() )
     {
         client.Send( message );
     }
 }

 protected void CreateQueues(QueueType mainQueueType, Endpoint mainQueueEndpoint, string user)
 {
     // will create the queues if they are not already there
     var queues = queueStrategy.InitializeQueue(mainQueueEndpoint, mainQueueType);
     foreach (var queue in queues)
     {
         GrantPermissions(queue, user);
     }
 }