From a650ff5c78c56ccf6f8cb0cf202c121eb40c82db Mon Sep 17 00:00:00 2001
From: Martin Regen <mregen@microsoft.com>
Date: Fri, 4 Nov 2016 16:04:52 +0100
Subject: [PATCH 1/6] improve autostart of boilers by using the state machine
 start call

---
 .../Opc.Ua.Sample/Boiler/BoilerNodeManager.cs  | 18 +++++++++++++-----
 .../Opc.Ua.Sample/Boiler/BoilerState.cs        |  2 +-
 2 files changed, 14 insertions(+), 6 deletions(-)

diff --git a/SampleApplications/Samples/Opc.Ua.Sample/Boiler/BoilerNodeManager.cs b/SampleApplications/Samples/Opc.Ua.Sample/Boiler/BoilerNodeManager.cs
index 7adeefe3e8..c2e3c53a2b 100644
--- a/SampleApplications/Samples/Opc.Ua.Sample/Boiler/BoilerNodeManager.cs
+++ b/SampleApplications/Samples/Opc.Ua.Sample/Boiler/BoilerNodeManager.cs
@@ -132,8 +132,12 @@ namespace Boiler
 
             AddPredefinedNode(context, boiler);
 
-            // Autostart boiler simulation
-            boiler.OnControlSimulation(context, null, 0, Opc.Ua.Methods.ProgramStateMachineType_Start, null, null);
+            // Autostart boiler simulation state machine
+            MethodState start = boiler.Simulation.Start;
+            IList<Variant> inputArguments = new List<Variant>();
+            IList<Variant> outputArguments = new List<Variant>();
+            List<ServiceResult> errors = new List<ServiceResult>();
+            start.Call(context, boiler.NodeId, inputArguments, errors, outputArguments);
 
         }
 
@@ -203,15 +207,19 @@ namespace Boiler
                     BoilerState activeNode = new BoilerState(passiveNode.Parent);
                     activeNode.Create(context, passiveNode);
 
-                    // Autostart boiler simulation
-                    activeNode.OnControlSimulation(context, null, 0, Opc.Ua.Methods.ProgramStateMachineType_Start, null, null);
-
                     // replace the node in the parent.
                     if (passiveNode.Parent != null)
                     {
                         passiveNode.Parent.ReplaceChild(context, activeNode);
                     }
 
+                    // Autostart boiler simulation state machine
+                    MethodState start = activeNode.Simulation.Start;
+                    IList<Variant> inputArguments = new List<Variant>();
+                    IList<Variant> outputArguments = new List<Variant>();
+                    List<ServiceResult> errors = new List<ServiceResult>();
+                    start.Call(context, activeNode.NodeId, inputArguments, errors, outputArguments);
+
                     return activeNode;
                 }
             }
diff --git a/SampleApplications/Samples/Opc.Ua.Sample/Boiler/BoilerState.cs b/SampleApplications/Samples/Opc.Ua.Sample/Boiler/BoilerState.cs
index 1c7c37fb7e..1afd4dcbf3 100644
--- a/SampleApplications/Samples/Opc.Ua.Sample/Boiler/BoilerState.cs
+++ b/SampleApplications/Samples/Opc.Ua.Sample/Boiler/BoilerState.cs
@@ -71,7 +71,7 @@ namespace Boiler
         /// <summary>
         /// Changes the state of the simulation.
         /// </summary>
-        public ServiceResult OnControlSimulation(
+        private ServiceResult OnControlSimulation(
             ISystemContext context,
             StateMachineState machine,
             uint transitionId,

From d49eca0ae3687f84fe20aed2c95d1b157145bbbe Mon Sep 17 00:00:00 2001
From: Martin Regen <mregen@microsoft.com>
Date: Mon, 7 Nov 2016 11:28:27 +0100
Subject: [PATCH 2/6] fix threading issues in console publisher

---
 .../NetCoreConsolePublisher/AmqpConnection.cs |  61 +++++-
 .../Opc.Ua.Publisher.Endpoints.xml            |  18 +-
 .../Opc.Ua.Publisher.config.xml               |   8 +-
 .../NetCoreConsolePublisher/Program.cs        | 188 ++++++++++--------
 4 files changed, 162 insertions(+), 113 deletions(-)

diff --git a/SampleApplications/Samples/NetCoreConsolePublisher/AmqpConnection.cs b/SampleApplications/Samples/NetCoreConsolePublisher/AmqpConnection.cs
index a3edd4e6f1..f7923b88ba 100644
--- a/SampleApplications/Samples/NetCoreConsolePublisher/AmqpConnection.cs
+++ b/SampleApplications/Samples/NetCoreConsolePublisher/AmqpConnection.cs
@@ -75,12 +75,14 @@ namespace Opc.Ua.Publisher
         private Connection m_connection;
         private Session m_session;
         private SenderLink m_link;
-        private Queue<ArraySegment<byte>> messages;
+        private LinkedList<ArraySegment<byte>> messages;
 
         private DateTime m_currentExpiryTime;
         private Timer m_tokenRenewalTimer;
         private bool m_closed;
         private int m_counter;
+        private object m_sending;
+        private int m_sendallthreads;
 
         #endregion
 
@@ -108,7 +110,13 @@ namespace Opc.Ua.Publisher
         /// </summary>
         private void Initialize()
         {
-            messages = new Queue<ArraySegment<byte>>();
+            m_connection = null;
+            m_session = null;
+            m_link = null;
+            m_closed = true;
+            m_sending = new object();
+            m_sendallthreads = 0;
+            messages = new LinkedList<ArraySegment<byte>>();
         }
 
         #endregion
@@ -156,7 +164,7 @@ namespace Opc.Ua.Publisher
         {
             lock (messages)
             {
-                messages.Enqueue(body);
+                messages.AddLast(body);
             }
 
             if (IsClosed())
@@ -175,16 +183,40 @@ namespace Opc.Ua.Publisher
         /// </summary>
         protected void SendAll()
         {
+            // make sure only one send all task is active
+            if (Interlocked.Increment(ref m_sendallthreads) != 1)
+            {
+                Interlocked.Decrement(ref m_sendallthreads);
+                return;
+            }
+
             try
             {
-                lock (messages)
+
+                while (!IsClosed())
                 {
-                    while (!IsClosed() && messages.Count != 0)
+                    ArraySegment<byte> onemessage;
+                    lock (messages)
                     {
-                        bool sent = SendOne(messages.Peek());
-                        if (sent)
+                        if (messages.Count == 0)
                         {
-                            messages.Dequeue();
+                            break;
+                        }
+                        onemessage = messages.First.Value;
+                        messages.RemoveFirst();
+                    }
+
+                    bool sent;
+                    lock (m_sending)
+                    {
+                        sent = SendOne(onemessage);
+                    }
+
+                    if (!sent)
+                    {
+                        lock (messages)
+                        {
+                            messages.AddFirst(onemessage);
                         }
                     }
                 }
@@ -193,6 +225,10 @@ namespace Opc.Ua.Publisher
             {
                 Close();
             }
+            finally
+            {
+                Interlocked.Decrement(ref m_sendallthreads);
+            }
         }
 
         /// <summary>
@@ -360,7 +396,10 @@ namespace Opc.Ua.Publisher
             }
 
             // Ensure we have a token
-            await RenewTokenAsync(GenerateSharedAccessToken());
+            lock (m_sending)
+            {
+                RenewTokenAsync(GenerateSharedAccessToken()).Wait();
+            }
 
             // then start the periodic renewal
             int interval = (int)(TokenLifetime * 0.8);
@@ -442,12 +481,12 @@ namespace Opc.Ua.Publisher
         {
             try
             {
-                lock (messages)
+                lock (m_sending)
                 {
                     bool result = RenewTokenAsync(GenerateSharedAccessToken()).Wait(60000);
                     if (!result)
                     {
-                        Utils.Trace( "Unexpected timeout error renewing token.");
+                        Utils.Trace("Unexpected timeout error renewing token.");
                     }
                 }
             }
diff --git a/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.Endpoints.xml b/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.Endpoints.xml
index a828504cf2..a73d2d3093 100644
--- a/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.Endpoints.xml
+++ b/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.Endpoints.xml
@@ -7,23 +7,7 @@
   <ua:Endpoints>
     <ua:ConfiguredEndpoint>
       <ua:Endpoint>
-        <EndpointUrl>opc.tcp://localhost:61210/UA/SampleClient</EndpointUrl>
-        <SecurityMode>SignAndEncrypt_3</SecurityMode>
-        <SecurityPolicyUri>http://opcfoundation.org/UA/SecurityPolicy#Basic128Rsa15</SecurityPolicyUri>
-        <UserIdentityTokens>
-          <UserTokenPolicy>
-            <TokenType>Anonymous_0</TokenType>
-          </UserTokenPolicy>
-        </UserIdentityTokens>
-        <TransportProfileUri>http://opcfoundation.org/UA/profiles/transport/wsxmlorbinary</TransportProfileUri>
-        <SecurityLevel>0</SecurityLevel>
-      </ua:Endpoint>
-      <ua:UpdateBeforeConnect>true</ua:UpdateBeforeConnect>
-      <ua:SelectedUserTokenPolicy>0</ua:SelectedUserTokenPolicy>
-    </ua:ConfiguredEndpoint>
-    <ua:ConfiguredEndpoint>
-      <ua:Endpoint>
-        <EndpointUrl>opc.tcp://192.168.2.101:51210/UA/SampleServer</EndpointUrl>
+        <EndpointUrl>opc.tcp://localhost:61410/UA/SampleClient</EndpointUrl>
         <SecurityMode>SignAndEncrypt_3</SecurityMode>
         <SecurityPolicyUri>http://opcfoundation.org/UA/SecurityPolicy#Basic128Rsa15</SecurityPolicyUri>
         <UserIdentityTokens>
diff --git a/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.config.xml b/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.config.xml
index 586cead167..06724f0c7e 100644
--- a/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.config.xml
+++ b/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.config.xml
@@ -91,6 +91,10 @@
       <StorePath>OPC Foundation/CertificateStores/RejectedCertificates</StorePath>
     </RejectedCertificateStore>
 
+    <!-- WARNING: The following setting (to automatically accept untrusted certificates) should be used
+    for easy debugging purposes ONLY and turned off for production deployments! -->
+    <AutoAcceptUntrustedCertificates>true</AutoAcceptUntrustedCertificates>
+
   </SecurityConfiguration>
 
   <TransportConfigurations></TransportConfigurations>
@@ -143,8 +147,8 @@
          For example, a URL used for an endpoint which uses the Basic256 security policy would look like this:
          http://localhost:61211/UA/SampleClient/Basic256 -->
     <BaseAddresses>
-      <ua:String>opc.tcp://localhost:61210/UA/SampleClient</ua:String>
-      <ua:String>https://localhost:61212/UA/SampleClient/</ua:String>
+      <ua:String>opc.tcp://localhost:61410/UA/SampleClient</ua:String>
+      <ua:String>https://localhost:61412/UA/SampleClient/</ua:String>
     </BaseAddresses>
 
     <!-- 
diff --git a/SampleApplications/Samples/NetCoreConsolePublisher/Program.cs b/SampleApplications/Samples/NetCoreConsolePublisher/Program.cs
index 32f48456ec..1eab107bf6 100644
--- a/SampleApplications/Samples/NetCoreConsolePublisher/Program.cs
+++ b/SampleApplications/Samples/NetCoreConsolePublisher/Program.cs
@@ -19,6 +19,7 @@ using System;
 using System.Collections.Generic;
 using System.Runtime.Serialization;
 using System.Text;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace NetCoreConsolePublisher
@@ -58,16 +59,22 @@ namespace NetCoreConsolePublisher
             {
                 Console.WriteLine(message);
             }
+
             if (ask)
             {
-                ConsoleKeyInfo result = Console.ReadKey();
-                Console.WriteLine();
-                return await Task.FromResult((result.KeyChar == 'y') || (result.KeyChar == 'Y') || (result.KeyChar == '\r'));
-            }
-            else
-            {
-                return await Task.FromResult(true);
+                try
+                {
+                    ConsoleKeyInfo result = Console.ReadKey();
+                    Console.WriteLine();
+                    return await Task.FromResult((result.KeyChar == 'y') || (result.KeyChar == 'Y') || (result.KeyChar == '\r'));
+                }
+                catch
+                {
+                    // intentionally fall through
+                }
             }
+
+            return await Task.FromResult(true);
         }
     }
 
@@ -80,6 +87,38 @@ namespace NetCoreConsolePublisher
         private static Session m_session = null;
 
         public static void Main(string[] args)
+        {
+            bool started = false;
+            try
+            {
+                Task t = ConsoleSamplePublisher();
+                t.Wait();
+                Console.WriteLine("Publisher started. Press any key to exit...");
+                started = true;
+            }
+            catch (Exception ex)
+            {
+                Utils.Trace("Exception:" + ex.Message);
+                Console.WriteLine("Exception: {0}", ex.Message);
+                Console.WriteLine("Press any key to exit...");
+            }
+
+            try {
+                Console.ReadKey(true);
+            }
+            catch
+            {
+                if (started)
+                {
+                    // wait forever if there is no console
+                    Thread.Sleep(Timeout.Infinite);
+                }
+            }
+
+            CleanupPublisher();
+        }
+
+        private static async Task ConsoleSamplePublisher()
         {
             ApplicationInstance.MessageDlg = new ApplicationMessageDlg();
             ApplicationInstance application = new ApplicationInstance();
@@ -87,90 +126,73 @@ namespace NetCoreConsolePublisher
             application.ApplicationType = ApplicationType.ClientAndServer;
             application.ConfigSectionName = "Opc.Ua.Publisher";
 
-            try
+            // load the application configuration.
+            m_configuration = await application.LoadApplicationConfiguration(false);
+
+            // check the application certificate.
+            bool certOK = await application.CheckApplicationInstanceCertificate(false, 0);
+            if (!certOK)
             {
-                // load the application configuration.
-                Task<ApplicationConfiguration> task = application.LoadApplicationConfiguration(false);
-                task.Wait();
-                m_configuration = task.Result;
-
-                // check the application certificate.
-                Task<bool> task2 = application.CheckApplicationInstanceCertificate(false, 0);
-                task2.Wait();
-                bool certOK = task2.Result;
-                if (!certOK)
-                {
-                    throw new Exception("Application instance certificate invalid!");
-                }
-
-                // start the server.
-                Task task3 = application.Start(new SampleServer());
-                task3.Wait();
-
-                // get list of cached endpoints.
-                m_endpoints = m_configuration.LoadCachedEndpoints(true);
-                m_endpoints.DiscoveryUrls = m_configuration.ClientConfiguration.WellKnownDiscoveryUrls;
-
-                // start publishers.
-                m_publishers = AmqpConnectionCollection.Load(m_configuration);
-                foreach (AmqpConnection publisher in m_publishers)
-                {
-                    Task t = publisher.OpenAsync();
-                }
-
-                m_MonitoredItem_Notification = new MonitoredItemNotificationEventHandler(MonitoredItem_Notification);
-
-                // connect to a server.
-                EndpointConnect();
-
-                // publish preconfigured nodes
-                PublishedNodesCollection nodes = PublishedNodesCollection.Load(m_configuration);
-                foreach (NodeId node in nodes)
-                {
-                    CreateMonitoredItem(node);
-                }
-
-                Console.WriteLine("Publisher started. Press any key to exit...");
-                Console.ReadKey(true);
-            
-                if (m_publishers != null)
-                {
-                    foreach (var publisher in m_publishers)
-                    {
-                        publisher.Close();
-                    }
-                }
-
-                if (m_session != null)
-                {
-                    m_session.Close();
-                }
+                throw new Exception("Application instance certificate invalid!");
             }
-            catch (ServiceResultException ex)
+
+            // start the server.
+            await application.Start(new SampleServer());
+
+            // get list of cached endpoints.
+            m_endpoints = m_configuration.LoadCachedEndpoints(true);
+            m_endpoints.DiscoveryUrls = m_configuration.ClientConfiguration.WellKnownDiscoveryUrls;
+
+            // start publishers.
+            m_publishers = AmqpConnectionCollection.Load(m_configuration);
+            foreach (AmqpConnection publisher in m_publishers)
             {
-                Utils.Trace("ServiceResultException:" + ex.Message);
-                Console.WriteLine("Exception: {0}", ex.Message);
-                Console.WriteLine("Press any key to exit...");
-                Console.ReadKey(true);
+                await publisher.OpenAsync();
+            }
+
+            m_MonitoredItem_Notification = new MonitoredItemNotificationEventHandler(MonitoredItem_Notification);
+
+            // connect to first server
+            await EndpointConnect(0);
+
+            // publish preconfigured nodes
+            PublishedNodesCollection nodes = PublishedNodesCollection.Load(m_configuration);
+            foreach (NodeId node in nodes)
+            {
+                CreateMonitoredItem(node);
             }
         }
 
-        public static void EndpointConnect()
+        private static void CleanupPublisher()
         {
-            // Connect to the first cached endpoint in our list
-            if (m_endpoints.Count > 0)
+            if (m_publishers != null)
             {
-                Task<Session> task = Session.Create(
-                m_configuration,
-                m_endpoints[0],
-                true,
-                false,
-                m_configuration.ApplicationName,
-                60000,
-                null,
-                null);
-                task.Wait();
-                m_session = task.Result;
+                foreach (var publisher in m_publishers)
+                {
+                    publisher.Close();
+                }
+            }
+
+            if (m_session != null)
+            {
+                m_session.Close();
+            }
+        }
+
+        private static async Task EndpointConnect(int endpoint)
+        {
+            // Connect to cached endpoint in our list
+            if (m_endpoints.Count > endpoint)
+            {
+                m_session = await Session.Create(
+                    m_configuration,
+                    m_endpoints[endpoint],
+                    true,
+                    false,
+                    m_configuration.ApplicationName,
+                    60000,
+                    null,
+                    null);
 
                 if (m_session != null)
                 {

From 7974b5cf6e8ef7000daae73e0ca4efed1f209334 Mon Sep 17 00:00:00 2001
From: Martin Regen <mregen@microsoft.com>
Date: Mon, 7 Nov 2016 12:35:09 +0100
Subject: [PATCH 3/6] fix publisher config file typo

---
 .../{Opc.Ua.Publisher.config.xml => Opc.Ua.Publisher.Config.xml}  | 0
 1 file changed, 0 insertions(+), 0 deletions(-)
 rename SampleApplications/Samples/NetCoreConsolePublisher/{Opc.Ua.Publisher.config.xml => Opc.Ua.Publisher.Config.xml} (100%)

diff --git a/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.config.xml b/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.Config.xml
similarity index 100%
rename from SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.config.xml
rename to SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.Config.xml

From 40873fd246ba2bfb2a9a893f2dacfb1361d41cd0 Mon Sep 17 00:00:00 2001
From: Martin Regen <mregen@microsoft.com>
Date: Mon, 7 Nov 2016 20:25:01 +0100
Subject: [PATCH 4/6] refactor amqp connection using amqp async send

---
 .../NetCoreConsolePublisher/AmqpConnection.cs | 123 ++++++++++++------
 1 file changed, 83 insertions(+), 40 deletions(-)

diff --git a/SampleApplications/Samples/NetCoreConsolePublisher/AmqpConnection.cs b/SampleApplications/Samples/NetCoreConsolePublisher/AmqpConnection.cs
index f7923b88ba..6031f2569c 100644
--- a/SampleApplications/Samples/NetCoreConsolePublisher/AmqpConnection.cs
+++ b/SampleApplications/Samples/NetCoreConsolePublisher/AmqpConnection.cs
@@ -80,7 +80,9 @@ namespace Opc.Ua.Publisher
         private DateTime m_currentExpiryTime;
         private Timer m_tokenRenewalTimer;
         private bool m_closed;
-        private int m_counter;
+        private int m_sendCounter;
+        private int m_sendAcceptedCounter;
+        private int m_sendRejectedCounter;
         private object m_sending;
         private int m_sendallthreads;
 
@@ -117,6 +119,9 @@ namespace Opc.Ua.Publisher
             m_sending = new object();
             m_sendallthreads = 0;
             messages = new LinkedList<ArraySegment<byte>>();
+            m_sendCounter = 0;
+            m_sendAcceptedCounter = 0;
+            m_sendRejectedCounter = 0;
         }
 
         #endregion
@@ -127,33 +132,47 @@ namespace Opc.Ua.Publisher
         /// <returns></returns>
         public async Task OpenAsync()
         {
-            Close();
-
-            ConnectionFactory factory = new ConnectionFactory();
-            factory.AMQP.ContainerId = Guid.NewGuid().ToString();
-            if (UseCbs)
+            // make sure only one SendAll or OpenAsync task is active
+            if (Interlocked.Increment(ref m_sendallthreads) != 1)
             {
-                factory.SASL.Profile = SaslProfile.External;
+                Interlocked.Decrement(ref m_sendallthreads);
+                return;
             }
 
-            m_connection = await factory.CreateAsync(GetAddress());
-            m_connection.Closed += new ClosedCallback(OnConnectionClosed);
-
-            if (UseCbs && KeyName != null && KeyValue != null)
+            try
             {
-                await StartCbs();
+                Close();
+
+                ConnectionFactory factory = new ConnectionFactory();
+                factory.AMQP.ContainerId = Guid.NewGuid().ToString();
+                if (UseCbs)
+                {
+                    factory.SASL.Profile = SaslProfile.External;
+                }
+
+                m_connection = await factory.CreateAsync(GetAddress());
+                m_connection.Closed += new ClosedCallback(OnConnectionClosed);
+
+                if (UseCbs && KeyName != null && KeyValue != null)
+                {
+                    StartCbs();
+                }
+                else
+                {
+                    await ResetLinkAsync();
+                }
+
+                Utils.Trace("AMQP Connection opened, connected to '{0}'...", Endpoint);
+
+                m_closed = false;
             }
-            else
+            finally
             {
-                await ResetLinkAsync();
+                Interlocked.Decrement(ref m_sendallthreads);
             }
 
-            Utils.Trace("AMQP Connection opened, connected to '{0}'...", Endpoint);
-
-            m_closed = false;
-
             // Push out the messages we have so far
-            await Task.Run(new Action(SendAll));
+            SendAll();
         }
 
         /// <summary>
@@ -209,7 +228,7 @@ namespace Opc.Ua.Publisher
                     bool sent;
                     lock (m_sending)
                     {
-                        sent = SendOne(onemessage);
+                        sent = SendOneAsync(onemessage, SendAllCallback);
                     }
 
                     if (!sent)
@@ -231,37 +250,61 @@ namespace Opc.Ua.Publisher
             }
         }
 
+        /// <summary>
+        /// Send outcome callback
+        /// </summary>
+        /// <param name="body"></param>
+        /// <returns>Whether message was sent</returns>
+        void SendAllCallback(Message message, Outcome outcome, object state)
+        {
+            if (outcome.Descriptor.Code == 36)
+            {
+                // accepted
+                m_sendAcceptedCounter++;
+            }
+            else 
+            {
+                // rejected or other fail reason
+                m_sendRejectedCounter++;
+            }
+            if (((m_sendRejectedCounter + m_sendAcceptedCounter) % 100) == 0)
+            {
+                Console.WriteLine("Send Statistics: {0} sent {1} accepted {2} rejected", 
+                    m_sendCounter, m_sendAcceptedCounter, m_sendRejectedCounter);
+            }
+        }
+
         /// <summary>
         /// Send message
         /// </summary>
         /// <param name="body"></param>
         /// <returns>Whether message was sent</returns>
-        protected bool SendOne(ArraySegment<byte> body)
+        protected bool SendOneAsync(ArraySegment<byte> body, OutcomeCallback SendCallback)
         {
-            m_counter++;
-
             if (IsClosed())
             {
                 return false;
             }
 
-            var istrm = new MemoryStream(body.Array, body.Offset, body.Count, false);
-
-            Message message = new Message()
+            using (var istrm = new MemoryStream(body.Array, body.Offset, body.Count, false))
             {
-                BodySection = new Data() { Binary = istrm.ToArray() }
-            };
+                Message message = new Message()
+                {
+                    BodySection = new Data() { Binary = istrm.ToArray() }
+                };
 
-            message.Properties = new Properties()
-            {
-                MessageId = Guid.NewGuid().ToString(),
-                ContentType = "application/opcua+json"
-            };
+                message.Properties = new Properties()
+                {
+                    MessageId = Guid.NewGuid().ToString(),
+                    ContentType = "application/opcua+json"
+                };
 
-            if (m_link != null)
-            {
-                m_link.Send(message, 6000);
-                return true;
+                if (m_link != null)
+                {
+                    m_sendCounter++;
+                    m_link.Send(message, SendCallback, null);
+                    return true;
+                }
             }
 
             return false;
@@ -378,7 +421,7 @@ namespace Opc.Ua.Publisher
         /// Start cbs protocol on the underlying connection
         /// </summary>
         /// <returns>Task to wait on</returns>
-        protected async Task StartCbs()
+        protected void StartCbs()
         {
             if (m_connection == null)
             {
@@ -398,7 +441,7 @@ namespace Opc.Ua.Publisher
             // Ensure we have a token
             lock (m_sending)
             {
-                RenewTokenAsync(GenerateSharedAccessToken()).Wait();
+                RenewTokenAsync(GenerateSharedAccessToken()).Wait(TokenLifetime);
             }
 
             // then start the periodic renewal
@@ -483,7 +526,7 @@ namespace Opc.Ua.Publisher
             {
                 lock (m_sending)
                 {
-                    bool result = RenewTokenAsync(GenerateSharedAccessToken()).Wait(60000);
+                    bool result = RenewTokenAsync(GenerateSharedAccessToken()).Wait(TokenLifetime);
                     if (!result)
                     {
                         Utils.Trace("Unexpected timeout error renewing token.");

From a97a3ba7e74de17bf4d66fba49e41208cbb32cb2 Mon Sep 17 00:00:00 2001
From: Martin Regen <mregen@microsoft.com>
Date: Tue, 8 Nov 2016 07:29:39 +0100
Subject: [PATCH 5/6] unify all publishers on port 61410 update to latest amqp
 lib and amqp send code

---
 .../Opc.Ua.Publisher.Config.xml               |   4 +-
 .../Opc.Ua.Publisher.Endpoints.xml            |   2 +-
 .../Samples/Publisher.Net4/AmqpConnection.cs  | 193 +++++++++++++-----
 .../Opc.Ua.Publisher.Endpoints.xml            |  18 +-
 .../Opc.Ua.Publisher.Net4.csproj              |   4 +-
 .../Opc.Ua.Publisher.config.xml               |   4 +-
 .../Samples/Publisher.Net4/packages.config    |   2 +-
 .../Samples/Publisher/AmqpConnection.cs       | 193 +++++++++++++-----
 .../Publisher/Opc.Ua.Publisher.Endpoints.xml  |   2 +-
 .../Publisher/Opc.Ua.Publisher.config.xml     |   4 +-
 .../Samples/Publisher/project.json            |   2 +-
 11 files changed, 301 insertions(+), 127 deletions(-)

diff --git a/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.Config.xml b/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.Config.xml
index 06724f0c7e..06682a623d 100644
--- a/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.Config.xml
+++ b/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.Config.xml
@@ -147,8 +147,8 @@
          For example, a URL used for an endpoint which uses the Basic256 security policy would look like this:
          http://localhost:61211/UA/SampleClient/Basic256 -->
     <BaseAddresses>
-      <ua:String>opc.tcp://localhost:61410/UA/SampleClient</ua:String>
-      <ua:String>https://localhost:61412/UA/SampleClient/</ua:String>
+      <ua:String>opc.tcp://localhost:61410/UA/Publisher</ua:String>
+      <ua:String>https://localhost:61412/UA/Publisher/</ua:String>
     </BaseAddresses>
 
     <!-- 
diff --git a/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.Endpoints.xml b/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.Endpoints.xml
index a73d2d3093..452011680c 100644
--- a/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.Endpoints.xml
+++ b/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.Endpoints.xml
@@ -7,7 +7,7 @@
   <ua:Endpoints>
     <ua:ConfiguredEndpoint>
       <ua:Endpoint>
-        <EndpointUrl>opc.tcp://localhost:61410/UA/SampleClient</EndpointUrl>
+        <EndpointUrl>opc.tcp://localhost:61410/UA/Publisher</EndpointUrl>
         <SecurityMode>SignAndEncrypt_3</SecurityMode>
         <SecurityPolicyUri>http://opcfoundation.org/UA/SecurityPolicy#Basic128Rsa15</SecurityPolicyUri>
         <UserIdentityTokens>
diff --git a/SampleApplications/Samples/Publisher.Net4/AmqpConnection.cs b/SampleApplications/Samples/Publisher.Net4/AmqpConnection.cs
index 4bf24e9869..162ec1fcc9 100644
--- a/SampleApplications/Samples/Publisher.Net4/AmqpConnection.cs
+++ b/SampleApplications/Samples/Publisher.Net4/AmqpConnection.cs
@@ -17,6 +17,10 @@ namespace Opc.Ua.Publisher
     [CollectionDataContract(Name = "ListOfAmqpConnectionConfigurations", Namespace = Namespaces.OpcUaConfig, ItemName = "AmqpConnectionConfiguration")]
     public partial class AmqpConnectionCollection : List<AmqpConnection>
     {
+        public AmqpConnectionCollection()
+        {
+        }
+
         public static AmqpConnectionCollection Load(ApplicationConfiguration configuration)
         {
             return configuration.ParseExtension<AmqpConnectionCollection>();
@@ -71,12 +75,16 @@ namespace Opc.Ua.Publisher
         private Connection m_connection;
         private Session m_session;
         private SenderLink m_link;
-        private Queue<ArraySegment<byte>> messages;
+        private LinkedList<ArraySegment<byte>> messages;
 
         private DateTime m_currentExpiryTime;
         private Timer m_tokenRenewalTimer;
         private bool m_closed;
-        private int m_counter;
+        private int m_sendCounter;
+        private int m_sendAcceptedCounter;
+        private int m_sendRejectedCounter;
+        private object m_sending;
+        private int m_sendallthreads;
 
         #endregion
 
@@ -104,7 +112,16 @@ namespace Opc.Ua.Publisher
         /// </summary>
         private void Initialize()
         {
-            messages = new Queue<ArraySegment<byte>>();
+            m_connection = null;
+            m_session = null;
+            m_link = null;
+            m_closed = true;
+            m_sending = new object();
+            m_sendallthreads = 0;
+            messages = new LinkedList<ArraySegment<byte>>();
+            m_sendCounter = 0;
+            m_sendAcceptedCounter = 0;
+            m_sendRejectedCounter = 0;
         }
 
         #endregion
@@ -115,33 +132,47 @@ namespace Opc.Ua.Publisher
         /// <returns></returns>
         public async Task OpenAsync()
         {
-            Close();
-
-            ConnectionFactory factory = new ConnectionFactory();
-            factory.AMQP.ContainerId = Guid.NewGuid().ToString();
-            if (UseCbs)
+            // make sure only one SendAll or OpenAsync task is active
+            if (Interlocked.Increment(ref m_sendallthreads) != 1)
             {
-                factory.SASL.Profile = SaslProfile.External;
+                Interlocked.Decrement(ref m_sendallthreads);
+                return;
             }
 
-            m_connection = await factory.CreateAsync(GetAddress());
-            m_connection.Closed += new ClosedCallback(OnConnectionClosed);
-
-            if (UseCbs && KeyName != null && KeyValue != null)
+            try
             {
-                await StartCbs();
+                Close();
+
+                ConnectionFactory factory = new ConnectionFactory();
+                factory.AMQP.ContainerId = Guid.NewGuid().ToString();
+                if (UseCbs)
+                {
+                    factory.SASL.Profile = SaslProfile.External;
+                }
+
+                m_connection = await factory.CreateAsync(GetAddress());
+                m_connection.Closed += new ClosedCallback(OnConnectionClosed);
+
+                if (UseCbs && KeyName != null && KeyValue != null)
+                {
+                    StartCbs();
+                }
+                else
+                {
+                    await ResetLinkAsync();
+                }
+
+                Utils.Trace("AMQP Connection opened, connected to '{0}'...", Endpoint);
+
+                m_closed = false;
             }
-            else
+            finally
             {
-                await ResetLinkAsync();
+                Interlocked.Decrement(ref m_sendallthreads);
             }
 
-            Utils.Trace("AMQP Connection opened, connected to '{0}'...", Endpoint);
-
-            m_closed = false;
-
             // Push out the messages we have so far
-            await Task.Run(new Action(SendAll));
+            SendAll();
         }
 
         /// <summary>
@@ -152,10 +183,14 @@ namespace Opc.Ua.Publisher
         {
             lock (messages)
             {
-                messages.Enqueue(body);
+                messages.AddLast(body);
             }
 
-            if (!IsClosed())
+            if (IsClosed())
+            {
+                Task.Run(OpenAsync);
+            }
+            else
             {
                 // Push out the messages we have so far
                 Task.Run(new Action(SendAll));
@@ -167,16 +202,40 @@ namespace Opc.Ua.Publisher
         /// </summary>
         protected void SendAll()
         {
+            // make sure only one send all task is active
+            if (Interlocked.Increment(ref m_sendallthreads) != 1)
+            {
+                Interlocked.Decrement(ref m_sendallthreads);
+                return;
+            }
+
             try
             {
-                lock (messages)
+
+                while (!IsClosed())
                 {
-                    while (!IsClosed() && messages.Count != 0)
+                    ArraySegment<byte> onemessage;
+                    lock (messages)
                     {
-                        bool sent = SendOne(messages.Peek());
-                        if (sent)
+                        if (messages.Count == 0)
                         {
-                            messages.Dequeue();
+                            break;
+                        }
+                        onemessage = messages.First.Value;
+                        messages.RemoveFirst();
+                    }
+
+                    bool sent;
+                    lock (m_sending)
+                    {
+                        sent = SendOneAsync(onemessage, SendAllCallback);
+                    }
+
+                    if (!sent)
+                    {
+                        lock (messages)
+                        {
+                            messages.AddFirst(onemessage);
                         }
                     }
                 }
@@ -184,8 +243,34 @@ namespace Opc.Ua.Publisher
             catch (Exception)
             {
                 Close();
+            }
+            finally
+            {
+                Interlocked.Decrement(ref m_sendallthreads);
+            }
+        }
 
-                Task t = OpenAsync();
+        /// <summary>
+        /// Send outcome callback
+        /// </summary>
+        /// <param name="body"></param>
+        /// <returns>Whether message was sent</returns>
+        void SendAllCallback(Message message, Outcome outcome, object state)
+        {
+            if (outcome.Descriptor.Code == 36)
+            {
+                // accepted
+                m_sendAcceptedCounter++;
+            }
+            else 
+            {
+                // rejected or other fail reason
+                m_sendRejectedCounter++;
+            }
+            if (((m_sendRejectedCounter + m_sendAcceptedCounter) % 100) == 0)
+            {
+                Utils.Trace("Send Statistics: {0} sent {1} accepted {2} rejected", 
+                    m_sendCounter, m_sendAcceptedCounter, m_sendRejectedCounter);
             }
         }
 
@@ -194,32 +279,32 @@ namespace Opc.Ua.Publisher
         /// </summary>
         /// <param name="body"></param>
         /// <returns>Whether message was sent</returns>
-        protected bool SendOne(ArraySegment<byte> body)
+        protected bool SendOneAsync(ArraySegment<byte> body, OutcomeCallback SendCallback)
         {
-            m_counter++;
-
             if (IsClosed())
             {
                 return false;
             }
 
-            var istrm = new MemoryStream(body.Array, body.Offset, body.Count, false);
-
-            Message message = new Message()
+            using (var istrm = new MemoryStream(body.Array, body.Offset, body.Count, false))
             {
-                BodySection = new Data() { Binary = istrm.ToArray() }
-            };
+                Message message = new Message()
+                {
+                    BodySection = new Data() { Binary = istrm.ToArray() }
+                };
 
-            message.Properties = new Properties()
-            {
-                MessageId = Guid.NewGuid().ToString(),
-                ContentType = "application/opcua+json"
-            };
+                message.Properties = new Properties()
+                {
+                    MessageId = Guid.NewGuid().ToString(),
+                    ContentType = "application/opcua+json"
+                };
 
-            if (m_link != null)
-            {
-                m_link.Send(message, 6000);
-                return true;
+                if (m_link != null)
+                {
+                    m_sendCounter++;
+                    m_link.Send(message, SendCallback, null);
+                    return true;
+                }
             }
 
             return false;
@@ -336,7 +421,7 @@ namespace Opc.Ua.Publisher
         /// Start cbs protocol on the underlying connection
         /// </summary>
         /// <returns>Task to wait on</returns>
-        protected async Task StartCbs()
+        protected void StartCbs()
         {
             if (m_connection == null)
             {
@@ -354,7 +439,10 @@ namespace Opc.Ua.Publisher
             }
 
             // Ensure we have a token
-            await RenewTokenAsync(GenerateSharedAccessToken());
+            lock (m_sending)
+            {
+                RenewTokenAsync(GenerateSharedAccessToken()).Wait(TokenLifetime);
+            }
 
             // then start the periodic renewal
             int interval = (int)(TokenLifetime * 0.8);
@@ -436,7 +524,14 @@ namespace Opc.Ua.Publisher
         {
             try
             {
-                RenewTokenAsync(GenerateSharedAccessToken()).Wait(60000);
+                lock (m_sending)
+                {
+                    bool result = RenewTokenAsync(GenerateSharedAccessToken()).Wait(TokenLifetime);
+                    if (!result)
+                    {
+                        Utils.Trace("Unexpected timeout error renewing token.");
+                    }
+                }
             }
             catch (Exception e)
             {
diff --git a/SampleApplications/Samples/Publisher.Net4/Opc.Ua.Publisher.Endpoints.xml b/SampleApplications/Samples/Publisher.Net4/Opc.Ua.Publisher.Endpoints.xml
index 2b2f7c1ec1..452011680c 100644
--- a/SampleApplications/Samples/Publisher.Net4/Opc.Ua.Publisher.Endpoints.xml
+++ b/SampleApplications/Samples/Publisher.Net4/Opc.Ua.Publisher.Endpoints.xml
@@ -7,23 +7,7 @@
   <ua:Endpoints>
     <ua:ConfiguredEndpoint>
       <ua:Endpoint>
-        <EndpointUrl>opc.tcp://localhost:61210/UA/Publisher</EndpointUrl>
-        <SecurityMode>SignAndEncrypt_3</SecurityMode>
-        <SecurityPolicyUri>http://opcfoundation.org/UA/SecurityPolicy#Basic128Rsa15</SecurityPolicyUri>
-        <UserIdentityTokens>
-          <UserTokenPolicy>
-            <TokenType>Anonymous_0</TokenType>
-          </UserTokenPolicy>
-        </UserIdentityTokens>
-        <TransportProfileUri>http://opcfoundation.org/UA/profiles/transport/wsxmlorbinary</TransportProfileUri>
-        <SecurityLevel>0</SecurityLevel>
-      </ua:Endpoint>
-      <ua:UpdateBeforeConnect>true</ua:UpdateBeforeConnect>
-      <ua:SelectedUserTokenPolicy>0</ua:SelectedUserTokenPolicy>
-    </ua:ConfiguredEndpoint>
-    <ua:ConfiguredEndpoint>
-      <ua:Endpoint>
-        <EndpointUrl>opc.tcp://192.168.2.101:51210/UA/SampleServer</EndpointUrl>
+        <EndpointUrl>opc.tcp://localhost:61410/UA/Publisher</EndpointUrl>
         <SecurityMode>SignAndEncrypt_3</SecurityMode>
         <SecurityPolicyUri>http://opcfoundation.org/UA/SecurityPolicy#Basic128Rsa15</SecurityPolicyUri>
         <UserIdentityTokens>
diff --git a/SampleApplications/Samples/Publisher.Net4/Opc.Ua.Publisher.Net4.csproj b/SampleApplications/Samples/Publisher.Net4/Opc.Ua.Publisher.Net4.csproj
index 9ec383a50b..429c6f5ae8 100644
--- a/SampleApplications/Samples/Publisher.Net4/Opc.Ua.Publisher.Net4.csproj
+++ b/SampleApplications/Samples/Publisher.Net4/Opc.Ua.Publisher.Net4.csproj
@@ -63,8 +63,8 @@
     <Prefer32Bit>false</Prefer32Bit>
   </PropertyGroup>
   <ItemGroup>
-    <Reference Include="Amqp.Net, Version=1.1.8.0, Culture=neutral, processorArchitecture=MSIL">
-      <HintPath>..\..\..\packages\AMQPNetLite.1.1.8\lib\net45\Amqp.Net.dll</HintPath>
+    <Reference Include="Amqp.Net, Version=1.2.0.0, Culture=neutral, PublicKeyToken=905a7b1e6458e0c3, processorArchitecture=MSIL">
+      <HintPath>..\..\..\packages\AMQPNetLite.1.2.1\lib\net45\Amqp.Net.dll</HintPath>
       <Private>True</Private>
     </Reference>
     <Reference Include="Opc.Ua.Client, Version=1.2.334.6, Culture=neutral, processorArchitecture=MSIL">
diff --git a/SampleApplications/Samples/Publisher.Net4/Opc.Ua.Publisher.config.xml b/SampleApplications/Samples/Publisher.Net4/Opc.Ua.Publisher.config.xml
index 5528324b90..9beb740567 100644
--- a/SampleApplications/Samples/Publisher.Net4/Opc.Ua.Publisher.config.xml
+++ b/SampleApplications/Samples/Publisher.Net4/Opc.Ua.Publisher.config.xml
@@ -143,8 +143,8 @@
          For example, a URL used for an endpoint which uses the Basic256 security policy would look like this:
          http://localhost:61211/UA/Publisher/Basic256 -->
     <BaseAddresses>
-      <ua:String>opc.tcp://localhost:61210/UA/Publisher</ua:String>
-      <ua:String>https://localhost:61212/UA/Publisher/</ua:String>
+      <ua:String>opc.tcp://localhost:61410/UA/Publisher</ua:String>
+      <ua:String>https://localhost:61412/UA/Publisher/</ua:String>
     </BaseAddresses>
 
     <!-- 
diff --git a/SampleApplications/Samples/Publisher.Net4/packages.config b/SampleApplications/Samples/Publisher.Net4/packages.config
index 7a72ebed0e..01f7afad4d 100644
--- a/SampleApplications/Samples/Publisher.Net4/packages.config
+++ b/SampleApplications/Samples/Publisher.Net4/packages.config
@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="utf-8"?>
 <packages>
-  <package id="AMQPNetLite" version="1.1.8" targetFramework="net46" />
+  <package id="AMQPNetLite" version="1.2.1" targetFramework="net46" />
   <package id="System.Security.Cryptography.Algorithms" version="4.2.0" targetFramework="net46" />
   <package id="System.Security.Cryptography.Encoding" version="4.0.0" targetFramework="net46" />
   <package id="System.Security.Cryptography.Primitives" version="4.0.0" targetFramework="net46" />
diff --git a/SampleApplications/Samples/Publisher/AmqpConnection.cs b/SampleApplications/Samples/Publisher/AmqpConnection.cs
index 4bf24e9869..162ec1fcc9 100644
--- a/SampleApplications/Samples/Publisher/AmqpConnection.cs
+++ b/SampleApplications/Samples/Publisher/AmqpConnection.cs
@@ -17,6 +17,10 @@ namespace Opc.Ua.Publisher
     [CollectionDataContract(Name = "ListOfAmqpConnectionConfigurations", Namespace = Namespaces.OpcUaConfig, ItemName = "AmqpConnectionConfiguration")]
     public partial class AmqpConnectionCollection : List<AmqpConnection>
     {
+        public AmqpConnectionCollection()
+        {
+        }
+
         public static AmqpConnectionCollection Load(ApplicationConfiguration configuration)
         {
             return configuration.ParseExtension<AmqpConnectionCollection>();
@@ -71,12 +75,16 @@ namespace Opc.Ua.Publisher
         private Connection m_connection;
         private Session m_session;
         private SenderLink m_link;
-        private Queue<ArraySegment<byte>> messages;
+        private LinkedList<ArraySegment<byte>> messages;
 
         private DateTime m_currentExpiryTime;
         private Timer m_tokenRenewalTimer;
         private bool m_closed;
-        private int m_counter;
+        private int m_sendCounter;
+        private int m_sendAcceptedCounter;
+        private int m_sendRejectedCounter;
+        private object m_sending;
+        private int m_sendallthreads;
 
         #endregion
 
@@ -104,7 +112,16 @@ namespace Opc.Ua.Publisher
         /// </summary>
         private void Initialize()
         {
-            messages = new Queue<ArraySegment<byte>>();
+            m_connection = null;
+            m_session = null;
+            m_link = null;
+            m_closed = true;
+            m_sending = new object();
+            m_sendallthreads = 0;
+            messages = new LinkedList<ArraySegment<byte>>();
+            m_sendCounter = 0;
+            m_sendAcceptedCounter = 0;
+            m_sendRejectedCounter = 0;
         }
 
         #endregion
@@ -115,33 +132,47 @@ namespace Opc.Ua.Publisher
         /// <returns></returns>
         public async Task OpenAsync()
         {
-            Close();
-
-            ConnectionFactory factory = new ConnectionFactory();
-            factory.AMQP.ContainerId = Guid.NewGuid().ToString();
-            if (UseCbs)
+            // make sure only one SendAll or OpenAsync task is active
+            if (Interlocked.Increment(ref m_sendallthreads) != 1)
             {
-                factory.SASL.Profile = SaslProfile.External;
+                Interlocked.Decrement(ref m_sendallthreads);
+                return;
             }
 
-            m_connection = await factory.CreateAsync(GetAddress());
-            m_connection.Closed += new ClosedCallback(OnConnectionClosed);
-
-            if (UseCbs && KeyName != null && KeyValue != null)
+            try
             {
-                await StartCbs();
+                Close();
+
+                ConnectionFactory factory = new ConnectionFactory();
+                factory.AMQP.ContainerId = Guid.NewGuid().ToString();
+                if (UseCbs)
+                {
+                    factory.SASL.Profile = SaslProfile.External;
+                }
+
+                m_connection = await factory.CreateAsync(GetAddress());
+                m_connection.Closed += new ClosedCallback(OnConnectionClosed);
+
+                if (UseCbs && KeyName != null && KeyValue != null)
+                {
+                    StartCbs();
+                }
+                else
+                {
+                    await ResetLinkAsync();
+                }
+
+                Utils.Trace("AMQP Connection opened, connected to '{0}'...", Endpoint);
+
+                m_closed = false;
             }
-            else
+            finally
             {
-                await ResetLinkAsync();
+                Interlocked.Decrement(ref m_sendallthreads);
             }
 
-            Utils.Trace("AMQP Connection opened, connected to '{0}'...", Endpoint);
-
-            m_closed = false;
-
             // Push out the messages we have so far
-            await Task.Run(new Action(SendAll));
+            SendAll();
         }
 
         /// <summary>
@@ -152,10 +183,14 @@ namespace Opc.Ua.Publisher
         {
             lock (messages)
             {
-                messages.Enqueue(body);
+                messages.AddLast(body);
             }
 
-            if (!IsClosed())
+            if (IsClosed())
+            {
+                Task.Run(OpenAsync);
+            }
+            else
             {
                 // Push out the messages we have so far
                 Task.Run(new Action(SendAll));
@@ -167,16 +202,40 @@ namespace Opc.Ua.Publisher
         /// </summary>
         protected void SendAll()
         {
+            // make sure only one send all task is active
+            if (Interlocked.Increment(ref m_sendallthreads) != 1)
+            {
+                Interlocked.Decrement(ref m_sendallthreads);
+                return;
+            }
+
             try
             {
-                lock (messages)
+
+                while (!IsClosed())
                 {
-                    while (!IsClosed() && messages.Count != 0)
+                    ArraySegment<byte> onemessage;
+                    lock (messages)
                     {
-                        bool sent = SendOne(messages.Peek());
-                        if (sent)
+                        if (messages.Count == 0)
                         {
-                            messages.Dequeue();
+                            break;
+                        }
+                        onemessage = messages.First.Value;
+                        messages.RemoveFirst();
+                    }
+
+                    bool sent;
+                    lock (m_sending)
+                    {
+                        sent = SendOneAsync(onemessage, SendAllCallback);
+                    }
+
+                    if (!sent)
+                    {
+                        lock (messages)
+                        {
+                            messages.AddFirst(onemessage);
                         }
                     }
                 }
@@ -184,8 +243,34 @@ namespace Opc.Ua.Publisher
             catch (Exception)
             {
                 Close();
+            }
+            finally
+            {
+                Interlocked.Decrement(ref m_sendallthreads);
+            }
+        }
 
-                Task t = OpenAsync();
+        /// <summary>
+        /// Send outcome callback
+        /// </summary>
+        /// <param name="body"></param>
+        /// <returns>Whether message was sent</returns>
+        void SendAllCallback(Message message, Outcome outcome, object state)
+        {
+            if (outcome.Descriptor.Code == 36)
+            {
+                // accepted
+                m_sendAcceptedCounter++;
+            }
+            else 
+            {
+                // rejected or other fail reason
+                m_sendRejectedCounter++;
+            }
+            if (((m_sendRejectedCounter + m_sendAcceptedCounter) % 100) == 0)
+            {
+                Utils.Trace("Send Statistics: {0} sent {1} accepted {2} rejected", 
+                    m_sendCounter, m_sendAcceptedCounter, m_sendRejectedCounter);
             }
         }
 
@@ -194,32 +279,32 @@ namespace Opc.Ua.Publisher
         /// </summary>
         /// <param name="body"></param>
         /// <returns>Whether message was sent</returns>
-        protected bool SendOne(ArraySegment<byte> body)
+        protected bool SendOneAsync(ArraySegment<byte> body, OutcomeCallback SendCallback)
         {
-            m_counter++;
-
             if (IsClosed())
             {
                 return false;
             }
 
-            var istrm = new MemoryStream(body.Array, body.Offset, body.Count, false);
-
-            Message message = new Message()
+            using (var istrm = new MemoryStream(body.Array, body.Offset, body.Count, false))
             {
-                BodySection = new Data() { Binary = istrm.ToArray() }
-            };
+                Message message = new Message()
+                {
+                    BodySection = new Data() { Binary = istrm.ToArray() }
+                };
 
-            message.Properties = new Properties()
-            {
-                MessageId = Guid.NewGuid().ToString(),
-                ContentType = "application/opcua+json"
-            };
+                message.Properties = new Properties()
+                {
+                    MessageId = Guid.NewGuid().ToString(),
+                    ContentType = "application/opcua+json"
+                };
 
-            if (m_link != null)
-            {
-                m_link.Send(message, 6000);
-                return true;
+                if (m_link != null)
+                {
+                    m_sendCounter++;
+                    m_link.Send(message, SendCallback, null);
+                    return true;
+                }
             }
 
             return false;
@@ -336,7 +421,7 @@ namespace Opc.Ua.Publisher
         /// Start cbs protocol on the underlying connection
         /// </summary>
         /// <returns>Task to wait on</returns>
-        protected async Task StartCbs()
+        protected void StartCbs()
         {
             if (m_connection == null)
             {
@@ -354,7 +439,10 @@ namespace Opc.Ua.Publisher
             }
 
             // Ensure we have a token
-            await RenewTokenAsync(GenerateSharedAccessToken());
+            lock (m_sending)
+            {
+                RenewTokenAsync(GenerateSharedAccessToken()).Wait(TokenLifetime);
+            }
 
             // then start the periodic renewal
             int interval = (int)(TokenLifetime * 0.8);
@@ -436,7 +524,14 @@ namespace Opc.Ua.Publisher
         {
             try
             {
-                RenewTokenAsync(GenerateSharedAccessToken()).Wait(60000);
+                lock (m_sending)
+                {
+                    bool result = RenewTokenAsync(GenerateSharedAccessToken()).Wait(TokenLifetime);
+                    if (!result)
+                    {
+                        Utils.Trace("Unexpected timeout error renewing token.");
+                    }
+                }
             }
             catch (Exception e)
             {
diff --git a/SampleApplications/Samples/Publisher/Opc.Ua.Publisher.Endpoints.xml b/SampleApplications/Samples/Publisher/Opc.Ua.Publisher.Endpoints.xml
index d7b85ca188..452011680c 100644
--- a/SampleApplications/Samples/Publisher/Opc.Ua.Publisher.Endpoints.xml
+++ b/SampleApplications/Samples/Publisher/Opc.Ua.Publisher.Endpoints.xml
@@ -7,7 +7,7 @@
   <ua:Endpoints>
     <ua:ConfiguredEndpoint>
       <ua:Endpoint>
-        <EndpointUrl>opc.tcp://localhost:61210/UA/SampleClient</EndpointUrl>
+        <EndpointUrl>opc.tcp://localhost:61410/UA/Publisher</EndpointUrl>
         <SecurityMode>SignAndEncrypt_3</SecurityMode>
         <SecurityPolicyUri>http://opcfoundation.org/UA/SecurityPolicy#Basic128Rsa15</SecurityPolicyUri>
         <UserIdentityTokens>
diff --git a/SampleApplications/Samples/Publisher/Opc.Ua.Publisher.config.xml b/SampleApplications/Samples/Publisher/Opc.Ua.Publisher.config.xml
index 5528324b90..9beb740567 100644
--- a/SampleApplications/Samples/Publisher/Opc.Ua.Publisher.config.xml
+++ b/SampleApplications/Samples/Publisher/Opc.Ua.Publisher.config.xml
@@ -143,8 +143,8 @@
          For example, a URL used for an endpoint which uses the Basic256 security policy would look like this:
          http://localhost:61211/UA/Publisher/Basic256 -->
     <BaseAddresses>
-      <ua:String>opc.tcp://localhost:61210/UA/Publisher</ua:String>
-      <ua:String>https://localhost:61212/UA/Publisher/</ua:String>
+      <ua:String>opc.tcp://localhost:61410/UA/Publisher</ua:String>
+      <ua:String>https://localhost:61412/UA/Publisher/</ua:String>
     </BaseAddresses>
 
     <!-- 
diff --git a/SampleApplications/Samples/Publisher/project.json b/SampleApplications/Samples/Publisher/project.json
index 053fe15087..c55b29eefe 100644
--- a/SampleApplications/Samples/Publisher/project.json
+++ b/SampleApplications/Samples/Publisher/project.json
@@ -1,6 +1,6 @@
 {
   "dependencies": {
-    "AMQPNetLite": "1.1.8",
+    "AMQPNetLite": "1.2.1",
     "Microsoft.NETCore.UniversalWindowsPlatform": "5.2.2",
     "System.Xml.XmlDocument": "4.0.1"
   },

From 60786c6fae834e074b3479d0eb2f86383c995051 Mon Sep 17 00:00:00 2001
From: Martin Regen <mregen@microsoft.com>
Date: Tue, 8 Nov 2016 15:50:00 +0100
Subject: [PATCH 6/6] more improvements on amqp startup found with .Net
 publishers

---
 .../NetCoreConsolePublisher/AmqpConnection.cs | 24 ++++---
 .../Samples/Publisher.Net4/AmqpConnection.cs  | 22 +++---
 .../Samples/Publisher.Net4/PublisherForm.cs   | 67 ++++++++++---------
 .../Samples/Publisher/AmqpConnection.cs       | 22 +++---
 .../Samples/Publisher/ClientPage.xaml.cs      | 11 +--
 .../Samples/Publisher/Opc.Ua.Publisher.csproj |  3 +-
 6 files changed, 82 insertions(+), 67 deletions(-)

diff --git a/SampleApplications/Samples/NetCoreConsolePublisher/AmqpConnection.cs b/SampleApplications/Samples/NetCoreConsolePublisher/AmqpConnection.cs
index 6031f2569c..3e6fa169ec 100644
--- a/SampleApplications/Samples/NetCoreConsolePublisher/AmqpConnection.cs
+++ b/SampleApplications/Samples/NetCoreConsolePublisher/AmqpConnection.cs
@@ -17,10 +17,6 @@ namespace Opc.Ua.Publisher
     [CollectionDataContract(Name = "ListOfAmqpConnectionConfigurations", Namespace = Namespaces.OpcUaConfig, ItemName = "AmqpConnectionConfiguration")]
     public partial class AmqpConnectionCollection : List<AmqpConnection>
     {
-        public AmqpConnectionCollection()
-        {
-        }
-
         public static AmqpConnectionCollection Load(ApplicationConfiguration configuration)
         {
             return configuration.ParseExtension<AmqpConnectionCollection>();
@@ -155,7 +151,7 @@ namespace Opc.Ua.Publisher
 
                 if (UseCbs && KeyName != null && KeyValue != null)
                 {
-                    StartCbs();
+                    await StartCbs();
                 }
                 else
                 {
@@ -166,6 +162,10 @@ namespace Opc.Ua.Publisher
 
                 m_closed = false;
             }
+            catch (Exception e)
+            {
+                Utils.Trace("AMQP Connection failed to open, exception: {0}...", e.Message);
+            }
             finally
             {
                 Interlocked.Decrement(ref m_sendallthreads);
@@ -269,7 +269,7 @@ namespace Opc.Ua.Publisher
             }
             if (((m_sendRejectedCounter + m_sendAcceptedCounter) % 100) == 0)
             {
-                Console.WriteLine("Send Statistics: {0} sent {1} accepted {2} rejected", 
+                Utils.Trace("Send Statistics: {0} sent {1} accepted {2} rejected", 
                     m_sendCounter, m_sendAcceptedCounter, m_sendRejectedCounter);
             }
         }
@@ -316,6 +316,11 @@ namespace Opc.Ua.Publisher
         public void Close()
         {
             m_closed = true;
+            if (m_tokenRenewalTimer != null)
+            {
+                m_tokenRenewalTimer.Dispose();
+                m_tokenRenewalTimer = null;
+            }
             Dispose(true);
         }
 
@@ -421,7 +426,7 @@ namespace Opc.Ua.Publisher
         /// Start cbs protocol on the underlying connection
         /// </summary>
         /// <returns>Task to wait on</returns>
-        protected void StartCbs()
+        protected async Task StartCbs()
         {
             if (m_connection == null)
             {
@@ -439,10 +444,7 @@ namespace Opc.Ua.Publisher
             }
 
             // Ensure we have a token
-            lock (m_sending)
-            {
-                RenewTokenAsync(GenerateSharedAccessToken()).Wait(TokenLifetime);
-            }
+            await RenewTokenAsync(GenerateSharedAccessToken());
 
             // then start the periodic renewal
             int interval = (int)(TokenLifetime * 0.8);
diff --git a/SampleApplications/Samples/Publisher.Net4/AmqpConnection.cs b/SampleApplications/Samples/Publisher.Net4/AmqpConnection.cs
index 162ec1fcc9..3e6fa169ec 100644
--- a/SampleApplications/Samples/Publisher.Net4/AmqpConnection.cs
+++ b/SampleApplications/Samples/Publisher.Net4/AmqpConnection.cs
@@ -17,10 +17,6 @@ namespace Opc.Ua.Publisher
     [CollectionDataContract(Name = "ListOfAmqpConnectionConfigurations", Namespace = Namespaces.OpcUaConfig, ItemName = "AmqpConnectionConfiguration")]
     public partial class AmqpConnectionCollection : List<AmqpConnection>
     {
-        public AmqpConnectionCollection()
-        {
-        }
-
         public static AmqpConnectionCollection Load(ApplicationConfiguration configuration)
         {
             return configuration.ParseExtension<AmqpConnectionCollection>();
@@ -155,7 +151,7 @@ namespace Opc.Ua.Publisher
 
                 if (UseCbs && KeyName != null && KeyValue != null)
                 {
-                    StartCbs();
+                    await StartCbs();
                 }
                 else
                 {
@@ -166,6 +162,10 @@ namespace Opc.Ua.Publisher
 
                 m_closed = false;
             }
+            catch (Exception e)
+            {
+                Utils.Trace("AMQP Connection failed to open, exception: {0}...", e.Message);
+            }
             finally
             {
                 Interlocked.Decrement(ref m_sendallthreads);
@@ -316,6 +316,11 @@ namespace Opc.Ua.Publisher
         public void Close()
         {
             m_closed = true;
+            if (m_tokenRenewalTimer != null)
+            {
+                m_tokenRenewalTimer.Dispose();
+                m_tokenRenewalTimer = null;
+            }
             Dispose(true);
         }
 
@@ -421,7 +426,7 @@ namespace Opc.Ua.Publisher
         /// Start cbs protocol on the underlying connection
         /// </summary>
         /// <returns>Task to wait on</returns>
-        protected void StartCbs()
+        protected async Task StartCbs()
         {
             if (m_connection == null)
             {
@@ -439,10 +444,7 @@ namespace Opc.Ua.Publisher
             }
 
             // Ensure we have a token
-            lock (m_sending)
-            {
-                RenewTokenAsync(GenerateSharedAccessToken()).Wait(TokenLifetime);
-            }
+            await RenewTokenAsync(GenerateSharedAccessToken());
 
             // then start the periodic renewal
             int interval = (int)(TokenLifetime * 0.8);
diff --git a/SampleApplications/Samples/Publisher.Net4/PublisherForm.cs b/SampleApplications/Samples/Publisher.Net4/PublisherForm.cs
index cfc0e68284..9aaf7a8b8b 100644
--- a/SampleApplications/Samples/Publisher.Net4/PublisherForm.cs
+++ b/SampleApplications/Samples/Publisher.Net4/PublisherForm.cs
@@ -97,10 +97,6 @@ namespace Opc.Ua.Sample.Controls
             Disconnect();
 
             m_publishers = AmqpConnectionCollection.Load(configuration);
-            foreach (var publisher in m_publishers)
-            {
-                Task t = publisher.OpenAsync();
-            }
 
             this.NotificationsCTRL.ItemsAdded += NotificationsCTRL_ItemsAdded;
         }
@@ -111,34 +107,14 @@ namespace Opc.Ua.Sample.Controls
             {
                 foreach (NotificationMessageListCtrl.ItemData item in e.Items)
                 {
-                    if (item.NotificationMessage == null || item.Subscription.Session == null)
+                    if (item.NotificationMessage != null && item.Subscription.Session != null)
                     {
-                        return;
-                    }
-
-                    JsonEncoder encoder = new JsonEncoder(
-                        item.Subscription.Session.MessageContext, false);
-
-                    foreach (MonitoredItem monitoredItem in item.Subscription.MonitoredItems)
-                    {
-                        encoder.WriteNodeId("MonitoredItem", monitoredItem.ResolvedNodeId);
-                        ((DataChangeNotification)((NotificationData)item.NotificationMessage.NotificationData[0].Body)).MonitoredItems[0].Encode(encoder);
-
-                        string json = encoder.Close();
-                        json = json.Replace("\\", "");
-                        byte[] bytes = new UTF8Encoding(false).GetBytes(json);
-
-                        foreach (var publisher in m_publishers)
+                        IList<MonitoredItemNotification> listOfNotifications = item.NotificationMessage.GetDataChanges(false);
+                        foreach (MonitoredItemNotification notification in listOfNotifications)
                         {
-                            try
-                            {
-                                publisher.Publish(new ArraySegment<byte>(bytes));
-                                Utils.Trace(null, "Publishing: " + json);
-                            }
-                            catch (Exception ex)
-                            {
-                                Utils.Trace(ex, "Failed to publish message, dropping....");
-                            }
+                            // lookup the monitored item.
+                            MonitoredItem monitoredItem = item.Subscription.FindItemByClientHandle(notification.ClientHandle);
+                            this.CreateAmqpMessage(monitoredItem, notification.Value);
                         }
                     }
                 }
@@ -149,6 +125,37 @@ namespace Opc.Ua.Sample.Controls
             }
         }
 
+        private void CreateAmqpMessage(MonitoredItem monitoredItem, DataValue value)
+        {
+            JsonEncoder encoder = new JsonEncoder(
+                monitoredItem.Subscription.Session.MessageContext, false);
+
+            string hostname = monitoredItem.Subscription.Session.ConfiguredEndpoint.EndpointUrl.DnsSafeHost;
+            if (hostname == "localhost")
+            {
+                hostname = Utils.GetHostName();
+            }
+            encoder.WriteString("HostName", hostname);
+            encoder.WriteNodeId("MonitoredItem", monitoredItem.ResolvedNodeId);
+            encoder.WriteDataValue("Value", value);
+
+            var json = encoder.Close();
+            var bytes = new UTF8Encoding(false).GetBytes(json);
+
+            foreach (var publisher in m_publishers)
+            {
+                try
+                {
+                    publisher.Publish(new ArraySegment<byte>(bytes));
+                    Utils.Trace(null, "New JSON Message: " + json);
+                }
+                catch (Exception ex)
+                {
+                    Utils.Trace(ex, "Failed to add message, dropping....");
+                }
+            }
+        }
+
         /// <summary>
         /// Opens a new form.
         /// </summary>
diff --git a/SampleApplications/Samples/Publisher/AmqpConnection.cs b/SampleApplications/Samples/Publisher/AmqpConnection.cs
index 162ec1fcc9..3e6fa169ec 100644
--- a/SampleApplications/Samples/Publisher/AmqpConnection.cs
+++ b/SampleApplications/Samples/Publisher/AmqpConnection.cs
@@ -17,10 +17,6 @@ namespace Opc.Ua.Publisher
     [CollectionDataContract(Name = "ListOfAmqpConnectionConfigurations", Namespace = Namespaces.OpcUaConfig, ItemName = "AmqpConnectionConfiguration")]
     public partial class AmqpConnectionCollection : List<AmqpConnection>
     {
-        public AmqpConnectionCollection()
-        {
-        }
-
         public static AmqpConnectionCollection Load(ApplicationConfiguration configuration)
         {
             return configuration.ParseExtension<AmqpConnectionCollection>();
@@ -155,7 +151,7 @@ namespace Opc.Ua.Publisher
 
                 if (UseCbs && KeyName != null && KeyValue != null)
                 {
-                    StartCbs();
+                    await StartCbs();
                 }
                 else
                 {
@@ -166,6 +162,10 @@ namespace Opc.Ua.Publisher
 
                 m_closed = false;
             }
+            catch (Exception e)
+            {
+                Utils.Trace("AMQP Connection failed to open, exception: {0}...", e.Message);
+            }
             finally
             {
                 Interlocked.Decrement(ref m_sendallthreads);
@@ -316,6 +316,11 @@ namespace Opc.Ua.Publisher
         public void Close()
         {
             m_closed = true;
+            if (m_tokenRenewalTimer != null)
+            {
+                m_tokenRenewalTimer.Dispose();
+                m_tokenRenewalTimer = null;
+            }
             Dispose(true);
         }
 
@@ -421,7 +426,7 @@ namespace Opc.Ua.Publisher
         /// Start cbs protocol on the underlying connection
         /// </summary>
         /// <returns>Task to wait on</returns>
-        protected void StartCbs()
+        protected async Task StartCbs()
         {
             if (m_connection == null)
             {
@@ -439,10 +444,7 @@ namespace Opc.Ua.Publisher
             }
 
             // Ensure we have a token
-            lock (m_sending)
-            {
-                RenewTokenAsync(GenerateSharedAccessToken()).Wait(TokenLifetime);
-            }
+            await RenewTokenAsync(GenerateSharedAccessToken());
 
             // then start the periodic renewal
             int interval = (int)(TokenLifetime * 0.8);
diff --git a/SampleApplications/Samples/Publisher/ClientPage.xaml.cs b/SampleApplications/Samples/Publisher/ClientPage.xaml.cs
index f981e41f5a..96cf42fbb4 100644
--- a/SampleApplications/Samples/Publisher/ClientPage.xaml.cs
+++ b/SampleApplications/Samples/Publisher/ClientPage.xaml.cs
@@ -99,11 +99,6 @@ namespace Opc.Ua.Publisher
 
             m_publishers = AmqpConnectionCollection.Load(configuration);
 
-            foreach (var publisher in m_publishers)
-            {
-                Task t = publisher.OpenAsync();
-            }
-
             m_MonitoredItem_Notification = new MonitoredItemNotificationEventHandler(MonitoredItem_Notification);
         }
 
@@ -545,6 +540,12 @@ namespace Opc.Ua.Publisher
 
                 JsonEncoder encoder = new JsonEncoder(
                     monitoredItem.Subscription.Session.MessageContext, false);
+                string hostname = monitoredItem.Subscription.Session.ConfiguredEndpoint.EndpointUrl.DnsSafeHost;
+                if (hostname == "localhost")
+                {
+                    hostname = Utils.GetHostName();
+                }
+                encoder.WriteString("HostName", hostname);
                 encoder.WriteNodeId("MonitoredItem", monitoredItem.ResolvedNodeId);
                 e.NotificationValue.Encode(encoder);
 
diff --git a/SampleApplications/Samples/Publisher/Opc.Ua.Publisher.csproj b/SampleApplications/Samples/Publisher/Opc.Ua.Publisher.csproj
index 917fef8c50..233866113e 100644
--- a/SampleApplications/Samples/Publisher/Opc.Ua.Publisher.csproj
+++ b/SampleApplications/Samples/Publisher/Opc.Ua.Publisher.csproj
@@ -205,7 +205,8 @@
   </PropertyGroup>
   <Import Project="$(MSBuildExtensionsPath)\Microsoft\WindowsXaml\v$(VisualStudioVersion)\Microsoft.Windows.UI.Xaml.CSharp.targets" />
   <PropertyGroup>
-    <PostBuildEvent>xcopy "$(ProjectDir)OPC Foundation\*.*" "$(TargetDir)AppX\OPC Foundation" /I/E/R/Y</PostBuildEvent>
+    <PostBuildEvent>mkdir "$(ProjectDir)OPC Foundation"
+xcopy "$(ProjectDir)OPC Foundation\*.*" "$(TargetDir)AppX\OPC Foundation" /I/E/R/Y</PostBuildEvent>
   </PropertyGroup>
   <!-- To modify your build process, add your task inside one of the targets below and uncomment it. 
        Other similar extension points exist, see Microsoft.Common.targets.