diff --git a/SampleApplications/Samples/NetCoreConsolePublisher/AmqpConnection.cs b/SampleApplications/Samples/NetCoreConsolePublisher/AmqpConnection.cs index a3edd4e6f1..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 { - public AmqpConnectionCollection() - { - } - public static AmqpConnectionCollection Load(ApplicationConfiguration configuration) { return configuration.ParseExtension(); @@ -75,12 +71,16 @@ namespace Opc.Ua.Publisher private Connection m_connection; private Session m_session; private SenderLink m_link; - private Queue> messages; + private LinkedList> 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 /// private void Initialize() { - messages = new Queue>(); + m_connection = null; + m_session = null; + m_link = null; + m_closed = true; + m_sending = new object(); + m_sendallthreads = 0; + messages = new LinkedList>(); + m_sendCounter = 0; + m_sendAcceptedCounter = 0; + m_sendRejectedCounter = 0; } #endregion @@ -119,33 +128,51 @@ namespace Opc.Ua.Publisher /// 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(); } /// @@ -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 /// 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 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); + } + } + + /// + /// Send outcome callback + /// + /// + /// Whether message was sent + 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); + } } /// @@ -200,32 +279,32 @@ namespace Opc.Ua.Publisher /// /// /// Whether message was sent - protected bool SendOne(ArraySegment body) + protected bool SendOneAsync(ArraySegment 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."); } } } diff --git a/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.config.xml b/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.Config.xml similarity index 98% rename from SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.config.xml rename to SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.Config.xml index 586cead167..06682a623d 100644 --- a/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.config.xml +++ b/SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.Config.xml @@ -91,6 +91,10 @@ OPC Foundation/CertificateStores/RejectedCertificates + + true + @@ -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 --> - opc.tcp://localhost:61210/UA/SampleClient - https://localhost:61212/UA/SampleClient/ + opc.tcp://localhost:61410/UA/Publisher + https://localhost:61412/UA/Publisher/ - opc.tcp://localhost:61210/UA/Publisher - https://localhost:61212/UA/Publisher/ + opc.tcp://localhost:61410/UA/Publisher + https://localhost:61412/UA/Publisher/ - opc.tcp://localhost:61210/UA/Publisher - https://localhost:61212/UA/Publisher/ + opc.tcp://localhost:61410/UA/Publisher + https://localhost:61412/UA/Publisher/