Merge pull request #8 from OPCFoundation/prototyping

Merge Publisher fixes

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>();
@@ -75,12 +71,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
 
@@ -108,7 +108,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
@@ -119,33 +128,51 @@ 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)
+                {
+                    await StartCbs();
+                }
+                else
+                {
+                    await ResetLinkAsync();
+                }
+
+                Utils.Trace("AMQP Connection opened, connected to '{0}'...", Endpoint);
+
+                m_closed = false;
             }
-            else
+            catch (Exception e)
             {
-                await ResetLinkAsync();
+                Utils.Trace("AMQP Connection failed to open, exception: {0}...", e.Message);
+            }
+            finally
+            {
+                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>
@@ -156,7 +183,7 @@ namespace Opc.Ua.Publisher
         {
             lock (messages)
             {
-                messages.Enqueue(body);
+                messages.AddLast(body);
             }
 
             if (IsClosed())
@@ -175,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);
                         }
                     }
                 }
@@ -193,6 +244,34 @@ namespace Opc.Ua.Publisher
             {
                 Close();
             }
+            finally
+            {
+                Interlocked.Decrement(ref m_sendallthreads);
+            }
+        }
+
+        /// <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);
+            }
         }
 
         /// <summary>
@@ -200,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;
@@ -237,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);
         }
 
@@ -442,12 +526,12 @@ namespace Opc.Ua.Publisher
         {
             try
             {
-                lock (messages)
+                lock (m_sending)
                 {
-                    bool result = RenewTokenAsync(GenerateSharedAccessToken()).Wait(60000);
+                    bool result = RenewTokenAsync(GenerateSharedAccessToken()).Wait(TokenLifetime);
                     if (!result)
                     {
-                        Utils.Trace( "Unexpected timeout error renewing token.");
+                        Utils.Trace("Unexpected timeout error renewing token.");
                     }
                 }
             }

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/Publisher</ua:String>
+      <ua:String>https://localhost:61412/UA/Publisher/</ua:String>
     </BaseAddresses>
 
     <!-- 

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/Publisher</EndpointUrl>
         <SecurityMode>SignAndEncrypt_3</SecurityMode>
         <SecurityPolicyUri>http://opcfoundation.org/UA/SecurityPolicy#Basic128Rsa15</SecurityPolicyUri>
         <UserIdentityTokens>

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