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] 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)
                 {