GithubStar/UA-.NETStandard
1
0
Fork 0
mirror of https://github.com/OPCFoundation/UA-.NETStandard.git synced 2026-10-22 09:37:12 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix threading issues in console publisher

Browse Source
This commit is contained in:
Martin Regen
2016-11-07 11:28:27 +01:00
parent a650ff5c78
commit d49eca0ae3
4 changed files with 162 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
SampleApplications/Samples/NetCoreConsolePublisher/AmqpConnection.cs
+50 -11
View File
@@ -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.");
}
}
}
SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.Endpoints.xml
+1 -17
View File
@@ -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>
SampleApplications/Samples/NetCoreConsolePublisher/Opc.Ua.Publisher.config.xml
+6 -2
View File
@@ -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>
<!--
SampleApplications/Samples/NetCoreConsolePublisher/Program.cs
+105 -83
View File
@@ -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)
{