static void Main(string[] args)
        {
            //Create host and register custom transport listener
            var uri = new Uri(address);
            var host = new ContainerHost(new List<Uri>() { uri }, null, uri.UserInfo);
            host.CustomTransports.Add("pipe", NamedPipeTransport.Listener);
            host.RegisterMessageProcessor(nodeName, new MessageProcessor());
            host.Open();
            Console.WriteLine("Listener: running");

            //Create factory with custom transport factory
            var factory = new ConnectionFactory(new TransportProvider[] { NamedPipeTransport.Factory });
            var connection = factory.CreateAsync(new Address(address)).GetAwaiter().GetResult();
            var session = new Session(connection);
            var sender = new SenderLink(session, "message-client", nodeName);
            Console.WriteLine("Client: sending a message");
            sender.Send(new Message("Hello Pipe!"));
            sender.Close();
            session.Close();
            connection.Close();
            Console.WriteLine("Client: closed");

            host.Close();
            Console.WriteLine("Listener: closed");
        }

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

 /// <summary>
 /// Initializes a sender link.
 /// </summary>
 /// <param name="session">The session within which to create the link.</param>
 /// <param name="name">The link name.</param>
 /// <param name="attach">The attach frame to send for this link.</param>
 /// <param name="onAttached">The callback to invoke when an attach is received from peer.</param>
 public SenderLink(Session session, string name, Attach attach, OnAttached onAttached)
     : base(session, name, onAttached)
 {
     this.settleMode = attach.SndSettleMode;
     this.outgoingList = new LinkedList();
     this.SendAttach(false, this.deliveryCount, attach);
 }

        public void CloseConnectionWithDetachTest()
        {
            this.testListener.RegisterTarget(TestPoint.Close, (stream, channel, fields) =>
            {
                // send a detach
                TestListener.FRM(stream, 0x16UL, 0, channel, 0u, true);
                return TestOutcome.Continue;
            });

            string testName = "CloseConnectionWithDetachTest";

            Trace.WriteLine(TraceLevel.Information, "sync test");
            {
                Connection connection = new Connection(this.address);
                Session session = new Session(connection);
                SenderLink sender = new SenderLink(session, "sender-" + testName, "any");
                sender.Send(new Message("test") { Properties = new Properties() { MessageId = testName } });
                connection.Close();
                Assert.IsTrue(connection.Error == null, "connection has error!" + connection.Error);
            }

            Trace.WriteLine(TraceLevel.Information, "async test");
            Task.Factory.StartNew(async () =>
            {
                Connection connection = await Connection.Factory.CreateAsync(this.address);
                Session session = new Session(connection);
                SenderLink sender = new SenderLink(session, "sender-" + testName, "any");
                await sender.SendAsync(new Message("test") { Properties = new Properties() { MessageId = testName } });
                await connection.CloseAsync();
                Assert.IsTrue(connection.Error == null, "connection has error!" + connection.Error);

            }).Unwrap().GetAwaiter().GetResult();
        }

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

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

            Connection connection = await Connection.Factory.CreateAsync(this.testTarget.Address);
            Session session = new Session(connection);
            SenderLink sender = new SenderLink(session, "sender-" + testName, testTarget.Path);

            Student student = new Student("Tom");
            student.Age = 16;
            student.Address = new StreetAddress() { FullAddress = "100 Main St. Small Town" };
            student.DateOfBirth = new System.DateTime(1988, 5, 1, 1, 2, 3, 100, System.DateTimeKind.Utc);

            Message message = new Message(student);
            message.Properties = new Properties() { MessageId = "student" };
            await sender.SendAsync(message);

            ReceiverLink receiver = new ReceiverLink(session, "receiver-" + testName, testTarget.Path);
            Message message2 = await receiver.ReceiveAsync();
            Trace.WriteLine(TraceLevel.Information, "receive: {0}", message2.Properties);
            receiver.Accept(message);

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

            Student student2 = message2.GetBody<Student>();
            Assert.AreEqual(student.Age, student2.Age - 1); // incremented in OnDeserialized
            Assert.AreEqual(student.DateOfBirth, student2.DateOfBirth);
            Assert.AreEqual(student.Address.FullAddress, student2.Address.FullAddress);
        }

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

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

        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 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 static void Main()
        {
            // write your code here

            // wait for DHCP-allocated IP address
            // while (IPAddress.GetDefaultLocalAddress() == IPAddress.Any) ;

            // wait for network connectivity
            // while (!System.Net.NetworkInformation.NetworkInterface.GetIsNetworkAvailable()) ;

            Microsoft.SPOT.Net.NetworkInformation.NetworkChange.NetworkAvailabilityChanged += NetworkChange_NetworkAvailabilityChanged;
            Microsoft.SPOT.Net.NetworkInformation.NetworkChange.NetworkAddressChanged += NetworkChange_NetworkAddressChanged;

            networkAvailableEvent.WaitOne();
            Debug.Print("link is up!");
            networkAddressChangedEvent.WaitOne();
            Debug.Print("address acquired: " + Microsoft.SPOT.Net.NetworkInformation.NetworkInterface.GetAllNetworkInterfaces()[0].IPAddress);

            Debug.Print("\r\n*** GET NETWORK INTERFACE SETTINGS ***");
            Microsoft.SPOT.Net.NetworkInformation.NetworkInterface[] networkInterfaces = Microsoft.SPOT.Net.NetworkInformation.NetworkInterface.GetAllNetworkInterfaces();
            Debug.Print("Found " + networkInterfaces.Length + " network interfaces.");

            // get date/time via NTP
            DateTime dateTime = MFToolkit.Net.Ntp.NtpClient.GetNetworkTime();
            Utility.SetLocalTime(dateTime);

            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("/devices/{0}/events", 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();

            Thread.Sleep(Timeout.Infinite);
        }

 /// <summary>
 /// Initializes a receiver link.
 /// </summary>
 /// <param name="session">The session within which to create the link.</param>
 /// <param name="name">The link name.</param>
 /// <param name="attach">The attach frame to send for this link.</param>
 /// <param name="onAttached">The callback to invoke when an attach is received from peer.</param>
 public ReceiverLink(Session session, string name, Attach attach, OnAttached onAttached)
     : base(session, name, onAttached)
 {
     this.totalCredit = -1;
     this.receivedMessages = new LinkedList();
     this.waiterList = new LinkedList();
     this.SendAttach(true, 0, attach);
 }

 /// <summary>
 /// Initializes the link.
 /// </summary>
 /// <param name="session">The session.</param>
 /// <param name="name">The link name.</param>
 /// <param name="onAttached">The callback to handle received attach.</param>
 protected Link(Session session, string name, OnAttached onAttached)
 {
     this.session = session;
     this.name = name;
     this.onAttached = onAttached;
     this.handle = session.AddLink(this);
     this.state = LinkState.Start;
 }

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

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

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

        protected void GetPartitions(Session session)
        {
            ReceiverLink receiverLink = null;
            SenderLink senderLink = null;

            try
            {
                // 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 = new SenderLink(session, "mgmt-sender", "$management");
                receiverLink = new ReceiverLink(
                    session,
                    "mgmt-receiver",
                    new Attach()
                    {
                        Source = new Source() { Address = "$management" },
                        Target = new Target() { Address = clientNode }
                    },
                    null);

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

                var response = receiverLink.Receive(15000); // time out after 15 seconds
                if (response == null)
                {
                    throw new Exception("No get partitions response was received.");
                }

                receiverLink.Accept(response);

                Trace.WriteLine(TraceLevel.Information, "Partition info {0}", response.Body.ToString());
                var partitionStrings = (string[])((Map)response.Body)["partition_ids"];
                Trace.WriteLine(TraceLevel.Information, "Partitions {0}", string.Join(",", partitionStrings));
                this.partitions = new List<string>(partitionStrings);
            }
            catch (Exception x)
            {
                Trace.WriteLine(TraceLevel.Error, "Error retrieving partitions:\r\n{0}", x.ToString());
                throw x;
            }
            finally
            {
                if (receiverLink != null) receiverLink.Close();
                if (senderLink != null) senderLink.Close();
            }
        }

        public override void Send(BrokeredMessage brokeredMessage)
        {
            if (this.factory.OpenConnection())
            {
                if (this.session == null)
                {
                    this.session = new Session(this.factory.Connection);
                    this.link = new SenderLink(this.session, "amqp-send-link " + this.entity, entity);
                }

                Message message = brokeredMessage.ToAmqpMessage();
                this.link.Send(message);
            }
        }

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