static void RunRequestClient(string address)
        {
            Connection connection = new Connection(new Address(address));
            Session session = new Session(connection);

            string replyTo = "client-reply-to";
            Attach recvAttach = new Attach()
            {
                Source = new Source() { Address = "request_processor" },
                Target = new Target() { Address = replyTo }
            };

            ReceiverLink receiver = new ReceiverLink(session, "request-client-receiver", recvAttach, null);
            SenderLink sender = new SenderLink(session, "request-client-sender", "request_processor");

            Message request = new Message("hello");
            request.Properties = new Properties() { MessageId = "request1", ReplyTo = replyTo };
            sender.Send(request, null, null);
            Console.WriteLine("Sent request {0} body {1}", request.Properties, request.Body);

            Message response = receiver.Receive();
            Console.WriteLine("Received response: {0} body {1}", response.Properties, response.Body);
            receiver.Accept(response);

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

        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();
        }

        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();
        }

        public void TestMethod_BasicSendReceive()
        {
            string testName = "BasicSendReceive";
            const int nMsgs = 200;
            Connection connection = new Connection(testTarget.Address);
            Session session = new Session(connection);
            SenderLink sender = new SenderLink(session, "sender-" + testName, testTarget.Path);

            for (int i = 0; i < nMsgs; ++i)
            {
                Message message = new Message("msg" + i);
                message.Properties = new Properties() { GroupId = "abcdefg" };
                message.ApplicationProperties = new ApplicationProperties();
                message.ApplicationProperties["sn"] = i;
                sender.Send(message, null, null);
            }

            ReceiverLink receiver = new ReceiverLink(session, "receiver-" + testName, testTarget.Path);
            for (int i = 0; i < nMsgs; ++i)
            {
                Message message = receiver.Receive();
                Trace.WriteLine(TraceLevel.Verbose, "receive: {0}", message.ApplicationProperties["sn"]);
                receiver.Accept(message);
            }
            
            sender.Close();
            receiver.Close();
            session.Close();
            connection.Close();
        }

        public void PerfAtLeastOnceSend()
        {
            string testName = "PerfAtLeastOnceSend";
            Connection connection = new Connection(address);
            Session session = new Session(connection);
            this.sender = new SenderLink(session, "sender-" + testName, "q1");

            this.onOutcome = OnSendComplete;
            this.done = new ManualResetEvent(false);
            this.totalCount = 1000000;
            this.completedCount = 0;
            this.initialCount = 300;
            this.batchCount = 100;
            Trace.TraceLevel = TraceLevel.Information;

            var watch = new System.Diagnostics.Stopwatch();
            watch.Start();

            this.SendMessages(initialCount);

            this.done.WaitOne();
            watch.Stop();
            Trace.WriteLine(TraceLevel.Information, "total: {0}, time: {1}ms", this.totalCount, watch.ElapsedMilliseconds);

            connection.Close();
        }

        public void TransactedPosting()
        {
            string testName = "TransactedPosting";
            int nMsgs = 5;

            Connection connection = new Connection(this.address);
            Session session = new Session(connection);
            SenderLink sender = new SenderLink(session, "sender-" + testName, "q1");

            // commit
            using (var ts = new TransactionScope())
            {
                for (int i = 0; i < nMsgs; i++)
                {
                    Message message = new Message("test");
                    message.Properties = new Properties() { MessageId = "commit" + i, GroupId = testName };
                    sender.Send(message);
                }

                ts.Complete();
            }

            // rollback
            using (var ts = new TransactionScope())
            {
                for (int i = nMsgs; i < nMsgs * 2; i++)
                {
                    Message message = new Message("test");
                    message.Properties = new Properties() { MessageId = "rollback" + i, GroupId = testName };
                    sender.Send(message);
                }
            }

            // commit
            using (var ts = new TransactionScope())
            {
                for (int i = 0; i < nMsgs; i++)
                {
                    Message message = new Message("test");
                    message.Properties = new Properties() { MessageId = "commit" + i, GroupId = testName };
                    sender.Send(message);
                }

                ts.Complete();
            }

            ReceiverLink receiver = new ReceiverLink(session, "receiver-" + testName, "q1");
            for (int i = 0; i < nMsgs * 2; i++)
            {
                Message message = receiver.Receive();
                Trace.WriteLine(TraceLevel.Information, "receive: {0}", message.Properties.MessageId);
                receiver.Accept(message);
                Assert.IsTrue(message.Properties.MessageId.StartsWith("commit"));
            }

            connection.Close();
        }

 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();
 }

        //
        // 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 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;
        }

        async Task StartAsync(Connection connection)
        {
            try
            {
                await this.PumpAsync(connection.OnHeader, connection.OnFrame);
            }
            catch (Exception exception)
            {
                connection.OnIoException(exception);
            }

        }

        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;
        }

            void Setup()
            {
                this.connection = new Connection(new Address(address));
                this.session = new Session(connection);

                Attach recvAttach = new Attach()
                {
                    Source = new Source() { Address = "request_processor" },
                    Target = new Target() { Address = this.replyTo }
                };

                this.receiver = new ReceiverLink(session, "request-client-receiver", recvAttach, null);
                this.receiver.Start(300);
                this.sender = new SenderLink(session, "request-client-sender", "request_processor");
            }

        public void ContainerHostCloseTest()
        {
            string name = MethodInfo.GetCurrentMethod().Name;
            this.host.RegisterMessageProcessor(name, new TestMessageProcessor());

            //Create a client to send data to the host message processor
            var closedEvent = new ManualResetEvent(false);
            var connection = new Connection(Address);
            connection.Closed += (AmqpObject obj, Error error) =>
            {
                closedEvent.Set();
            };

            var session = new Session(connection);
            var sender = new SenderLink(session, "sender-link", name);

            //Send one message while the host is open
            sender.Send(new Message("Hello"), SendTimeout);

            //Close the host. this should close existing connections
            this.host.Close();

            Assert.IsTrue(closedEvent.WaitOne(10000), "connection is not closed after host is closed.");

            try
            {
                sender.Send(new Message("test"));
                Assert.IsTrue(false, "exception not thrown");
            }
            catch (AmqpException exception)
            {
                Assert.IsTrue(exception.Error != null, "Error is null");
                Assert.AreEqual((Symbol)ErrorCode.ConnectionForced, exception.Error.Condition, "Wrong error code");
            }

            connection.Close();

            // Reopen the host and send again
            this.host = new ContainerHost(new List<Uri>() { Uri }, null, Uri.UserInfo);
            this.host.RegisterMessageProcessor(name, new TestMessageProcessor());
            this.host.Open();

            connection = new Connection(Address);
            session = new Session(connection);
            sender = new SenderLink(session, "sender-link", name);
            sender.Send(new Message("Hello"), SendTimeout);
            connection.Close();
        }

        public void Connect(Connection connection, Address address, bool noVerification)
        {
            TcpSocket socket = new TcpSocket();
            socket.Connect(address.Host, address.Port);

            if (address.UseSsl)
            {
                SslSocket sslSocket = new SslSocket(socket, noVerification);
                sslSocket.AuthenticateAsClient(address.Host);
                this.socketTransport = sslSocket;
            }
            else
            {
                this.socketTransport = socket;
            }
        }

        internal Session(Connection connection, Begin begin)
        {
            this.connection = connection;
            this.channel = connection.AddSession(this);
            this.handleMax = begin.HandleMax;
            this.localLinks = new Link[1];
            this.remoteLinks = new Link[1];
            this.incomingList = new LinkedList();
            this.outgoingList = new LinkedList();
            this.nextOutgoingId = uint.MaxValue - 2u;
            this.outgoingWindow = begin.IncomingWindow;
            this.incomingDeliveryId = uint.MaxValue;

            begin.NextOutgoingId = this.nextOutgoingId;
            this.state = State.BeginSent;
            this.SendBegin(begin);
        }

        public void Send(OutgoingTransportMessage transportMessage, IEnumerable<string> addresses)
        {
            var messageBuffer = messageEncoder.Encode(transportMessage.Message);

            var message = new Message(messageBuffer);
            message.Header = new Header();
            message.Header.Durable = true;
            message.Properties = new Properties();
            message.Properties.CreationTime = DateTime.UtcNow;
            message.Properties.MessageId = Guid.NewGuid().ToString();
            message.Properties.ReplyTo = "TODO";
            message.ApplicationProperties = new ApplicationProperties();
            message.ApplicationProperties["LightRail.ContentType"] = messageEncoder.ContentType;
            message.ApplicationProperties["LightRail.EnclosedMessageTypes"] = string.Join(",", messageMapper.GetEnclosedMessageTypes(transportMessage.Message.GetType()).Distinct());
            foreach (var pair in transportMessage.Headers)
            {
                message.ApplicationProperties[pair.Key] = pair.Value;
            }

            var connection = new Connection(amqpAddress);
            var session = new Session(connection);
            // Azure does not support Amqp transactions "The server was unable to process the request; please retry the operation. If the problem persists, please contact your Service Bus administrator and provide the tracking id..TrackingId:583da4f8d58d4fa59dc9521c6f799cb8_GWIN-AN5B307EEHM,TimeStamp:11.7.2014. 7:44:17"
            try
            {
                foreach (var address in addresses)
                {
                    logger.Info("Sending Message {0} to {1}", message.Properties.MessageId, address);
                    var senderLink = new SenderLink(session, Guid.NewGuid().ToString(), address);
                    try
                    {
                        senderLink.Send(message);
                    }
                    finally
                    {
                        senderLink.Close();
                    }
                }
            }
            finally
            {
                session.Close();
                connection.Close();
            }
        }

        static void ReceiveMessages(string node, int count, string sessionId)
        {
            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);

            Trace.WriteLine(TraceLevel.Information, "Accepting a message session '{0}'...", sessionId ?? "<any>");
            Map filters = new Map();
            filters.Add(new Symbol("com.microsoft:session-filter"), sessionId);

            ReceiverLink receiver = new ReceiverLink(
                session,
                "sessionful-receiver-link",
                new Source() { Address = node, FilterSet = filters },
                null);

            for (int i = 0; i < count; i++)
            {
                Message message = receiver.Receive(30000);
                if (message == null)
                {
                    break;
                }

                if (i == 0)
                {
                    Trace.WriteLine(TraceLevel.Information, "Received message from session '{0}'", message.Properties.GroupId);
                }

                receiver.Accept(message);
            }

            Trace.WriteLine(TraceLevel.Information, "Finished receiving. Shutting down...");
            Trace.WriteLine(TraceLevel.Information, "");

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

		async static Task<bool> PutCbsToken(Connection connection, string host, string shareAccessSignature, string audience)
		{
			bool result = true;
			Session session = new Session(connection);

			string cbsReplyToAddress = "cbs-reply-to";
			var cbsSender = new SenderLink(session, "cbs-sender", "$cbs");
			var cbsReceiver = new ReceiverLink(session, cbsReplyToAddress, "$cbs");

			// construct the put-token message
			var request = new Message(shareAccessSignature);
			request.Properties = new Properties();
			request.Properties.MessageId = Guid.NewGuid().ToString();
			request.Properties.ReplyTo = cbsReplyToAddress;
			request.ApplicationProperties = new ApplicationProperties();
			request.ApplicationProperties["operation"] = "put-token";
			request.ApplicationProperties["type"] = "azure-devices.net:sastoken";
			request.ApplicationProperties["name"] = audience;
			await cbsSender.SendAsync(request);

			// receive the response
			var response = await cbsReceiver.ReceiveAsync();
			if (response == null || response.Properties == null || response.ApplicationProperties == null)
			{
				result = false;
			}
			else {
				int statusCode = (int)response.ApplicationProperties["status-code"];
				string statusCodeDescription = (string)response.ApplicationProperties["status-description"];
				if (statusCode != (int)202 && statusCode != (int)200) // !Accepted && !OK
				{
					result = false;
				}
			}

			// the sender/receiver may be kept open for refreshing tokens
			await cbsSender.CloseAsync();
			await cbsReceiver.CloseAsync();
			await session.CloseAsync();

			return result;
		}

        static void RunMessageClient(string address)
        {
            const int nMsgs = 10;
            Connection connection = new Connection(new Address(address));
            Session session = new Session(connection);
            SenderLink sender = new SenderLink(session, "message-client", "message_processor");

            for (int i = 0; i < nMsgs; ++i)
            {
                Message message = new Message("hello");
                message.Properties = new Properties() { MessageId = "msg" + i };
                message.ApplicationProperties = new ApplicationProperties();
                message.ApplicationProperties["sn"] = i;
                sender.Send(message);
                Console.WriteLine("Sent message {0} body {1}", message.Properties, message.Body);
            }

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