/// <summary>
        /// Creates a new connection with a custom open frame and a callback to handle remote open frame.
        /// </summary>
        /// <param name="address">The address of remote endpoint to connect to.</param>
        /// <param name="open">If specified, it is sent to open the connection, otherwise an open frame created from the AMQP settings property is sent.</param>
        /// <param name="onOpened">If specified, it is invoked when an open frame is received from the remote peer.</param>
        /// <returns></returns>
        public async Task<Connection> CreateAsync(Address address, Open open, OnOpened onOpened)
        {
            IAsyncTransport transport;
            if (WebSocketTransport.MatchScheme(address.Scheme))
            {
                WebSocketTransport wsTransport = new WebSocketTransport();
                await wsTransport.ConnectAsync(address);
                transport = wsTransport;
            }
            else
            {
                TcpTransport tcpTransport = new TcpTransport();
                await tcpTransport.ConnectAsync(address, this);
                transport = tcpTransport;
            }

            if (address.User != null)
            {
                SaslPlainProfile profile = new SaslPlainProfile(address.User, address.Password);
                transport = await profile.OpenAsync(address.Host, transport);
            }
            else if (this.saslSettings != null && this.saslSettings.Profile != null)
            {
                transport = await this.saslSettings.Profile.OpenAsync(address.Host, transport);
            }

            AsyncPump pump = new AsyncPump(transport);
            Connection connection = new Connection(this.AMQP, address, transport, open, onOpened);
            pump.Start(connection);

            return connection;
        }

        internal async Task<IAsyncTransport> ConnectAsync(Address address, Action<ClientWebSocketOptions> options)
        {
            Uri uri = new UriBuilder()
            {
                Scheme = address.Scheme,
                Port = GetDefaultPort(address.Scheme, address.Port),
                Host = address.Host,
                Path = address.Path
            }.Uri;

            ClientWebSocket cws = new ClientWebSocket();
            cws.Options.AddSubProtocol(WebSocketSubProtocol);
            if (options != null)
            {
                options(cws.Options);
            }

            await cws.ConnectAsync(uri, CancellationToken.None);
            if (cws.SubProtocol != WebSocketSubProtocol)
            {
                cws.Abort();

                throw new NotSupportedException(
                    string.Format(
                        CultureInfo.InvariantCulture,
                        "WebSocket SubProtocol used by the host is not the same that was requested: {0}",
                        cws.SubProtocol ?? "<null>"));
            }

            this.webSocket = cws;

            return this;
        }

        public async Task WebSocketSendReceiveAsync()
        {
            string testName = "WebSocketSendReceiveAsync";

            // assuming it matches the broker's setup and port is not taken
            Address wsAddress = new Address("ws://guest:[email protected]:18080");
            int nMsgs = 50;

            Connection connection = await Connection.Factory.CreateAsync(wsAddress);
            Session session = new Session(connection);
            SenderLink sender = new SenderLink(session, "sender-" + testName, "q1");

            for (int i = 0; i < nMsgs; ++i)
            {
                Message message = new Message();
                message.Properties = new Properties() { MessageId = "msg" + i, GroupId = testName };
                message.ApplicationProperties = new ApplicationProperties();
                message.ApplicationProperties["sn"] = i;
                await sender.SendAsync(message);
            }

            ReceiverLink receiver = new ReceiverLink(session, "receiver-" + testName, "q1");
            for (int i = 0; i < nMsgs; ++i)
            {
                Message message = await receiver.ReceiveAsync();
                Trace.WriteLine(TraceLevel.Information, "receive: {0}", message.ApplicationProperties["sn"]);
                receiver.Accept(message);
            }

            await sender.CloseAsync();
            await receiver.CloseAsync();
            await session.CloseAsync();
            await connection.CloseAsync();
        }

        static void Main(string[] args)
        {
            Amqp.Trace.TraceLevel = Amqp.TraceLevel.Frame | Amqp.TraceLevel.Verbose;
            #if NETMF
            Amqp.Trace.TraceListener = (f, a) => Debug.Print(DateTime.Now.ToString("[hh:ss.fff]") + " " + Fx.Format(f, a));
            #else
            Amqp.Trace.TraceListener = (f, a) => System.Diagnostics.Trace.WriteLine(DateTime.Now.ToString("[hh:ss.fff]") + " " + Fx.Format(f, a));
            #endif
            address = new Address(HOST, PORT, null, null);
            connection = new Connection(address);

            string audience = Fx.Format("{0}/devices/{1}", HOST, DEVICE_ID);
            string resourceUri = Fx.Format("{0}/devices/{1}", HOST, DEVICE_ID);

            string sasToken = GetSharedAccessSignature(null, DEVICE_KEY, resourceUri, new TimeSpan(1, 0, 0));
            bool cbs = PutCbsToken(connection, HOST, sasToken, audience);

            if (cbs)
            {
                session = new Session(connection);

                SendEvent();
                receiverThread = new Thread(ReceiveCommands);
                receiverThread.Start();
            }

            // just as example ...
            // the application ends only after received a command or timeout on receiving
            receiverThread.Join();

            session.Close();
            connection.Close();
        }

        async Task RunSampleAsync()
        {
            ConnectionFactory factory = new ConnectionFactory();
            factory.SASL.Profile = SaslProfile.External;

            Trace.WriteLine(TraceLevel.Information, "Establishing a connection...");
            Address address = new Address(this.Namespace, 5671, null, null, "/", "amqps");
            var connection = await factory.CreateAsync(address);

            // before any operation can be performed, a token must be put to the $cbs node
            Trace.WriteLine(TraceLevel.Information, "Putting a token to the $cbs node...");
            await PutTokenAsync(connection);

            Trace.WriteLine(TraceLevel.Information, "Sending a message...");
            var session = new Session(connection);
            var sender = new SenderLink(session, "ServiceBus.Cbs:sender-link", this.Entity);
            await sender.SendAsync(new Message("test"));
            await sender.CloseAsync();

            Trace.WriteLine(TraceLevel.Information, "Receiving the message back...");
            var receiver = new ReceiverLink(session, "ServiceBus.Cbs:receiver-link", this.Entity);
            var message = await receiver.ReceiveAsync();
            receiver.Accept(message);
            await receiver.CloseAsync();

            Trace.WriteLine(TraceLevel.Information, "Closing the connection...");
            await session.CloseAsync();
            await connection.CloseAsync();
        }

        static void Main(string[] args)
        {
            string brokerUrl = "amqp://localhost:5672";
              string address   = "my_queue";

              Address    brokerAddr = new Address(brokerUrl);
              Connection connection = new Connection(brokerAddr);
              Session    session    = new Session(connection);

              SenderLink   sender   = new   SenderLink(session, "sender",   address);
              ReceiverLink receiver = new ReceiverLink(session, "receiver", address);

              Message helloOut = new Message("Hello World!");
              sender.Send(helloOut);

              Message helloIn = receiver.Receive();
              receiver.Accept(helloIn);

              Console.WriteLine(helloIn.Body.ToString());

              receiver.Close();
              sender.Close();
              session.Close();
              connection.Close();
        }

        /// <summary>
        /// Constructor
        /// </summary>
        /// <param name="baseAddress">Base address to service bus</param>
        /// <param name="settings">AMQP transport settings</param>
        public AmqpMessagingFactory(Uri baseAddress, AmqpTransportSettings settings)
        {
            this.Address = baseAddress;
            this.settings = settings;

            SharedAccessSignatureTokenProvider sasTokenProvider = (SharedAccessSignatureTokenProvider)this.ServiceBusSecuritySettings.TokenProvider;
            this.amqpAddress = new Address(this.Address.Host, this.TransportSettings.Port, sasTokenProvider.KeyName, sasTokenProvider.SharedAccessKey);
        }

 public void Connect(Connection connection, Address address, bool noVerification)
 {
     this.connection = connection;
     this.socket = new StreamSocket();
     this.socket.ConnectAsync(
         new HostName(address.Host),
         address.Port.ToString(),
         address.UseSsl ? SocketProtectionLevel.Ssl : SocketProtectionLevel.PlainSocket).AsTask().GetAwaiter().GetResult();
 }

        public async Task ConnectAsync(Address address, ConnectionFactory factory)
        {
            StreamSocket ss = new StreamSocket();
            await ss.ConnectAsync(
                new HostName(address.Host),
                address.Port.ToString(),
                address.UseSsl ? SocketProtectionLevel.Tls12 : SocketProtectionLevel.PlainSocket);

            this.socket = ss;
        }

            public override Task<IAsyncTransport> CreateAsync(Address address)
            {
                NamedPipeClientStream client = new NamedPipeClientStream(address.Host, address.Path,
                    PipeDirection.InOut, PipeOptions.Asynchronous);
                client.Connect();

                TaskCompletionSource<IAsyncTransport> tcs = new TaskCompletionSource<IAsyncTransport>();
                tcs.SetResult(new NamedPipeTransport(client));
                return tcs.Task;
            }

        //
        // Sample invocation: Interop.Spout.exe --broker localhost:5672 --timeout 30 --address my-queue
        //
        static int Main(string[] args)
        {
            const int ERROR_SUCCESS = 0;
            const int ERROR_OTHER = 2;

            int exitCode = ERROR_SUCCESS;
            Connection connection = null;
            try
            {
                Options options = new Options(args);

                Address address = new Address(options.Url);
                connection = new Connection(address);
                Session session = new Session(connection);
                SenderLink sender = new SenderLink(session, "sender-spout", options.Address);
                // TODO: ReplyTo

                Stopwatch stopwatch = new Stopwatch();
                TimeSpan timespan = new TimeSpan(0, 0, options.Timeout);
                stopwatch.Start();
                for (int nSent = 0;
                    (0 == options.Count || nSent < options.Count) &&
                    (0 == options.Timeout || stopwatch.Elapsed <= timespan);
                    nSent++)
                {
                    string id = options.Id;
                    if (id.Equals(""))
                    {
                        Guid g = Guid.NewGuid();
                        id = g.ToString();
                    }
                    id += ":" + nSent.ToString();

                    Message message = new Message(options.Content);
                    message.Properties = new Properties() { MessageId = id };
                    sender.Send(message);
                    if (options.Print)
                    {
                        Console.WriteLine("Message(Properties={0}, ApplicationProperties={1}, Body={2}",
                                      message.Properties, message.ApplicationProperties, message.Body);
                    }
                }
                sender.Close();
                session.Close();
                connection.Close();
            }
            catch (Exception e)
            {
                Console.WriteLine("Exception {0}.", e);
                if (null != connection)
                    connection.Close();
                exitCode = ERROR_OTHER;
            }
            return exitCode;
        }

        public void Connect(Connection connection, Address address, bool noVerification)
        {
            this.connection = connection;
            var factory = new ConnectionFactory();
            if (noVerification)
            {
                factory.SSL.RemoteCertificateValidationCallback = noneCertValidator;
            }

            this.ConnectAsync(address, factory).GetAwaiter().GetResult();
        }

        public MainPage()
        {
            this.InitializeComponent();

            Amqp.Trace.TraceLevel = Amqp.TraceLevel.Frame | Amqp.TraceLevel.Verbose;
            Amqp.Trace.TraceListener = (f, a) => System.Diagnostics.Debug.WriteLine(DateTime.Now.ToString("[hh:ss.fff]") + " " + Fx.Format(f, a));

            address = new Address(HOST, PORT, null, null);
            connection = new Connection(address);

            session = new Session(connection);
        }

        //
        // Sample invocation: Interop.Drain.exe --broker localhost:5672 --timeout 30 --address my-queue
        //
        static int Main(string[] args)
        {
            const int ERROR_SUCCESS = 0;
            const int ERROR_NO_MESSAGE = 1;
            const int ERROR_OTHER = 2;

            int exitCode = ERROR_SUCCESS;
            Connection connection = null;
            try
            {
                Options options = new Options(args);

                Address address = new Address(options.Url);
                connection = new Connection(address);
                Session session = new Session(connection);
                ReceiverLink receiver = new ReceiverLink(session, "receiver-drain", options.Address);
                int timeout = int.MaxValue;
                if (!options.Forever)
                    timeout = 1000 * options.Timeout;
                Message message = new Message();
                int nReceived = 0;
                receiver.SetCredit(options.InitialCredit);
                while ((message = receiver.Receive(timeout)) != null)
                {
                    nReceived++;
                    if (!options.Quiet)
                    {
                        Console.WriteLine("Message(Properties={0}, ApplicationProperties={1}, Body={2}",
                                      message.Properties, message.ApplicationProperties, message.Body);
                    }
                    receiver.Accept(message);
                    if (options.Count > 0 && nReceived == options.Count)
                    {
                        break;
                    }
                }
                if (message == null)
                {
                    exitCode = ERROR_NO_MESSAGE;
                }
                receiver.Close();
                session.Close();
                connection.Close();
            }
            catch (Exception e)
            {
                Console.WriteLine("Exception {0}.", e);
                if (null != connection)
                    connection.Close();
                exitCode = ERROR_OTHER;
            }
            return exitCode;
        }

        public async Task ConnectAsync(Address address, ConnectionFactory factory)
        {
            Socket socket;
            IPAddress[] ipAddresses;
            IPAddress ipAddress;
            if (IPAddress.TryParse(address.Host, out ipAddress))
            {
                socket = new Socket(ipAddress.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
                ipAddresses = new IPAddress[] { ipAddress };
            }
            else
            {
                socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
                ipAddresses = Dns.GetHostEntry(address.Host).AddressList;
            }

            if (factory.tcpSettings != null)
            {
                factory.tcpSettings.Configure(socket);
            }

            await Task.Factory.FromAsync(
                (c, s) => ((Socket)s).BeginConnect(ipAddresses, address.Port, c, s),
                (r) => ((Socket)r.AsyncState).EndConnect(r),
                socket);

            IAsyncTransport transport;
            if (address.UseSsl)
            {
                SslStream sslStream;
                var ssl = factory.SslInternal;
                if (ssl == null)
                {
                    sslStream = new SslStream(new NetworkStream(socket));
                    await sslStream.AuthenticateAsClientAsync(address.Host);
                }
                else
                {
                    sslStream = new SslStream(new NetworkStream(socket), false, ssl.RemoteCertificateValidationCallback);
                    await sslStream.AuthenticateAsClientAsync(address.Host, ssl.ClientCertificates,
                        ssl.Protocols, ssl.CheckCertificateRevocation);
                }

                transport = new SslSocket(this, sslStream);
            }
            else
            {
                transport = new TcpSocket(this, socket);
            }

            this.socketTransport = transport;
            this.writer = new Writer(this, this.socketTransport);
        }

        public void Connect(Connection connection, Address address, bool noVerification)
        {
            var ipHostEntry = Dns.GetHostEntry(address.Host);
            Exception exception = null;
            TcpSocket socket = null;

            foreach (var ipAddress in ipHostEntry.AddressList)
            {
                if (ipAddress == null)
                {
                    continue;
                }

                try
                {
                    socket = new TcpSocket();
                    socket.Connect(new IPEndPoint(ipAddress, address.Port));
                    exception = null;
                    break;
                }
                catch (SocketException socketException)
                {
                    if (socket != null)
                    {
                        socket.Close();
                        socket = null;
                    }

                    exception = socketException;
                }
            }

            if (exception != null)
            {
                throw exception;
            }

            if (address.UseSsl)
            {
                SslSocket sslSocket = new SslSocket(socket);
                sslSocket.AuthenticateAsClient(
                    address.Host,
                    null,
                    noVerification ? SslVerification.NoVerification : SslVerification.VerifyPeer,
                    SslProtocols.Default);
                this.socketTransport = sslSocket;
            }
            else
            {
                this.socketTransport = socket;
            }
        }

        public SenderLink GetMessageSender(string topic)
        {
            if (_senders.ContainsKey(topic))
                return _senders[topic];

            Address address = new Address("amqp://guest:[email protected]:5672");
            Connection connection = Connection.Factory.CreateAsync(address).Result;
            Session session = new Session(connection);
            SenderLink sender = new SenderLink(session, "sender", topic);

            _senders.TryAdd(topic, sender);
            sender.Closed += Sender_Closed;
            return sender;
        }

        static string[] GetPartitions()
        {
            Trace.WriteLine(TraceLevel.Information, "Retrieving partitions...");
            Trace.WriteLine(TraceLevel.Information, "Establishing a connection...");
            Address address = new Address(sbNamespace, 5671, keyName, keyValue);
            Connection connection = new Connection(address);

            Trace.WriteLine(TraceLevel.Information, "Creating a session...");
            Session session = new Session(connection);

            // create a pair of links for request/response
            Trace.WriteLine(TraceLevel.Information, "Creating a request and a response link...");
            string clientNode = "client-temp-node";
            SenderLink sender = new SenderLink(session, "mgmt-sender", "$management");
            ReceiverLink receiver = new ReceiverLink(
                session,
                "mgmt-receiver",
                new Attach()
                {
                    Source = new Source() { Address = "$management" },
                    Target = new Target() { Address = clientNode }
                },
                null);

            Message request = new Message();
            request.Properties = new Properties() { MessageId = "request1", ReplyTo = clientNode };
            request.ApplicationProperties = new ApplicationProperties();
            request.ApplicationProperties["operation"] = "READ";
            request.ApplicationProperties["name"] = entity;
            request.ApplicationProperties["type"] = "com.microsoft:eventhub";
            sender.Send(request, null, null);

            Message response = receiver.Receive();
            if (response == null)
            {
                throw new Exception("No response was received.");
            }

            receiver.Accept(response);
            receiver.Close();
            sender.Close();
            connection.Close();

            Trace.WriteLine(TraceLevel.Information, "Partition info {0}", response.Body.ToString());
            string[] partitions = (string[])((Map)response.Body)["partition_ids"];
            Trace.WriteLine(TraceLevel.Information, "Partitions {0}", string.Join(",", partitions));
            Trace.WriteLine(TraceLevel.Information, "");

            return partitions;
        }

        public async Task ConnectAsync(Address address, ConnectionFactory factory)
        {
            SocketProtectionLevel spl = !address.UseSsl ?
                SocketProtectionLevel.PlainSocket :
#if NETFX_CORE
                SocketProtectionLevel.Tls12;
#else
                SocketProtectionLevel.Ssl;
#endif
            StreamSocket ss = new StreamSocket();
            await ss.ConnectAsync(new HostName(address.Host), address.Port.ToString(), spl);

            this.socket = ss;
        }

        public TestTarget()
        {
#if !COMPACT_FRAMEWORK && !NETFX_CORE && !NETMF
            this.address = Environment.GetEnvironmentVariable(envVarName);
#endif
            if (this.address == null)
            {
                this.address = defaultAddress;
            }

            // Verify that the URI is well formed.
            Address addr = new Address(this.address);
            // Extract the path without the leading "/".
            path = addr.Path.Substring(1);
        }