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

Refactoring in TcpMessageSocket

Browse Source
This commit is contained in:
Eduard Gilmutdinov
2017-08-10 14:15:28 -07:00
parent e88d3ab5e3
commit 6df6828828
2 changed files with 87 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs
+86 -166
View File
@@ -11,6 +11,7 @@
*/
using System;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Threading.Tasks;
@@ -171,8 +172,6 @@ namespace Opc.Ua.Bindings
m_sink = sink;
m_socket = null;
m_socketV4 = null;
m_socketV6 = null;
m_bufferManager = bufferManager;
m_receiveBufferSize = receiveBufferSize;
m_incomingMessageSize = -1;
@@ -242,7 +241,7 @@ namespace Opc.Ua.Bindings
/// <summary>
/// Connects to an endpoint.
/// </summary>
public async Task<bool> BeginConnect(Uri endpointUrl, EventHandler<IMessageSocketAsyncEventArgs> callback, object state)
public async Task BeginConnect(Uri endpointUrl, EventHandler<IMessageSocketAsyncEventArgs> callback, object state)
{
if (endpointUrl == null) throw new ArgumentNullException("endpointUrl");
@@ -251,151 +250,92 @@ namespace Opc.Ua.Bindings
throw new InvalidOperationException("The socket is already connected.");
}
// Get DNS host information.
// Get DNS host information
IPAddress[] hostAdresses = await Dns.GetHostAddressesAsync(endpointUrl.DnsSafeHost);
// try IPv4 and IPv6 address
IPAddress addressV4 = null;
IPAddress addressV6 = null;
// Get IPv4 and IPv6 address
IPAddress addressV4 = hostAdresses.FirstOrDefault(a => a.AddressFamily == AddressFamily.InterNetwork);
IPAddress addressV6 = hostAdresses.FirstOrDefault(a => a.AddressFamily == AddressFamily.InterNetworkV6);
foreach (IPAddress address in hostAdresses)
// Get port
int port = endpointUrl.Port;
if (port <= 0 || port > UInt16.MaxValue)
{
if (address.AddressFamily == AddressFamily.InterNetworkV6)
{
if (addressV6 == null)
{
addressV6 = address;
}
}
if (address.AddressFamily == AddressFamily.InterNetwork)
{
if (addressV4 == null)
{
addressV4 = address;
}
}
if ((addressV4 != null) && (addressV6 != null))
{
break;
}
port = Utils.UaTcpDefaultPort;
}
SocketError error = SocketError.NotInitialized;
TaskCompletionSource<SocketError> tcs = new TaskCompletionSource<SocketError>();
SocketAsyncEventArgs argsV4 = null;
SocketAsyncEventArgs argsV6 = null;
m_socketResponses = 0;
Action doCallback = () => callback(this, new TcpMessageSocketAsyncEventArgs { UserToken = state });
lock (m_socketLock)
{
// ensure a valid port.
int port = endpointUrl.Port;
if (port <= 0 || port > UInt16.MaxValue)
{
port = Utils.UaTcpDefaultPort;
}
// create sockets if IP address was provided
if (addressV6 != null)
{
m_socketResponses++;
m_socketV6 = new Socket(AddressFamily.InterNetworkV6, SocketType.Stream, ProtocolType.Tcp);
argsV6 = new SocketAsyncEventArgs()
{
UserToken = state,
RemoteEndPoint = new IPEndPoint(addressV6, port),
};
argsV6.Completed += (o, e) =>
{
lock (m_socketLock)
{
m_socketResponses--;
if (m_socketV6 != null)
{
if (m_socket == null &&
(m_socketResponses == 0 || e.SocketError == SocketError.Success))
{
m_socket = m_socketV6;
tcs.SetResult(e.SocketError);
callback(o, new TcpMessageSocketAsyncEventArgs { UserToken = e.UserToken });
}
else
{
m_socketV6.Dispose();
}
m_socketV6 = null;
}
}
e.Dispose();
};
}
if (addressV4 != null)
{
m_socketResponses++;
m_socketV4 = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
argsV4 = new SocketAsyncEventArgs()
{
UserToken = state,
RemoteEndPoint = new IPEndPoint(addressV4, port),
};
argsV4.Completed += (o, e) =>
{
lock (m_socketLock)
{
m_socketResponses--;
if (m_socketV4 != null)
{
if (m_socket == null &&
(m_socketResponses == 0 || e.SocketError == SocketError.Success))
{
m_socket = m_socketV4;
tcs.SetResult(e.SocketError);
callback(o, new TcpMessageSocketAsyncEventArgs { UserToken = e.UserToken });
}
else
{
m_socketV4.Dispose();
}
m_socketV4 = null;
}
}
e.Dispose();
};
BeginConnect(addressV6, AddressFamily.InterNetworkV6, port, doCallback);
}
bool connectV6Sync = true;
bool connectV4Sync = true;
if (m_socketV6 != null)
if (addressV4 != null && m_socket == null)
{
connectV6Sync = !m_socketV6.ConnectAsync(argsV6);
if (connectV6Sync)
{
// I/O completed synchronously
callback(m_socketV6, new TcpMessageSocketAsyncEventArgs { UserToken = state });
error = argsV6.SocketError;
}
}
if (m_socketV4 != null && error != SocketError.Success)
{
connectV4Sync = !m_socketV4.ConnectAsync(argsV4);
if (connectV4Sync)
{
// I/O completed synchronously
callback(m_socketV4, new TcpMessageSocketAsyncEventArgs { UserToken = state });
error = argsV4.SocketError;
}
}
if (connectV4Sync && connectV6Sync)
{
return (error == SocketError.Success) ? true : false;
BeginConnect(addressV4, AddressFamily.InterNetwork, port, doCallback);
}
}
}
error = await tcs.Task;
/// <summary>
/// Try to connect to endpoint and do callback if connected successfully
/// </summary>
/// <param name="address">Endpoint address</param>
/// <param name="addressFamily">Endpoint address family</param>
/// <param name="port">Endpoint port</param>
/// <param name="callback">Callback that must be executed if the connection would be established</param>
private void BeginConnect(IPAddress address, AddressFamily addressFamily, int port, Action callback)
{
var socket = new Socket(addressFamily, SocketType.Stream, ProtocolType.Tcp);
var args = new SocketAsyncEventArgs()
{
UserToken = callback,
RemoteEndPoint = new IPEndPoint(address, port),
};
args.Completed += new EventHandler<SocketAsyncEventArgs>(OnSocketConnected);
return (error == SocketError.Success) ? true : false;
var result = !socket.ConnectAsync(args);
if (result)
{
// I/O completed synchronously
OnSocketConnected(socket, args);
}
}
/// <summary>
/// Handle socket connection event
/// </summary>
/// <param name="sender"></param>
/// <param name="args"></param>
private void OnSocketConnected(object sender, SocketAsyncEventArgs args)
{
lock (m_socketLock)
{
var socket = sender as Socket;
if (!m_closed && m_socket == null && args.SocketError == SocketError.Success)
{
m_socket = socket;
((Action)args.UserToken)();
}
else
{
try
{
if (socket.Connected)
{
socket.Shutdown(SocketShutdown.Both);
}
}
finally
{
socket.Dispose();
}
}
}
args.Dispose();
}
/// <summary>
@@ -403,48 +343,27 @@ namespace Opc.Ua.Bindings
/// </summary>
public void Close()
{
// get the socket.
Socket socket = null;
Socket socketV4 = null;
Socket socketV6 = null;
lock (m_socketLock)
{
socket = m_socket;
socketV4 = m_socketV4;
socketV6 = m_socketV6;
m_socket = null;
m_socketV6 = null;
m_socketV4 = null;
}
m_closed = true;
// shutdown sockets which may still be active
// due to a timeout during ConnectAsync
if (socketV4 != null)
{
socketV4.Dispose();
}
if (socketV6 != null)
{
socketV6.Dispose();
}
// shutdown the socket.
if (socket != null)
{
try
// Shutdown the socket.
if (m_socket != null)
{
if (socket.Connected)
try
{
socket.Shutdown(SocketShutdown.Both);
}
if (m_socket.Connected)
{
m_socket.Shutdown(SocketShutdown.Both);
}
socket.Dispose();
}
catch (Exception e)
{
Utils.Trace(e, "Unexpected error closing socket.");
m_socket.Dispose();
m_socket = null;
}
catch (Exception e)
{
Utils.Trace(e, "Unexpected error closing socket.");
}
}
}
}
@@ -718,6 +637,7 @@ namespace Opc.Ua.Bindings
private object m_socketLock = new object();
private Socket m_socket;
private bool m_closed = false;
private object m_readLock = new object();
private byte[] m_receiveBuffer;
Stack/Opc.Ua.Core/Stack/Transport/IMessageSocket.cs
+1 -1
View File
@@ -137,7 +137,7 @@ namespace Opc.Ua.Bindings
/// <summary>
/// Connects to an endpoint.
/// </summary>
Task<bool> BeginConnect(Uri endpointUrl, EventHandler<IMessageSocketAsyncEventArgs> callback, object state);
Task BeginConnect(Uri endpointUrl, EventHandler<IMessageSocketAsyncEventArgs> callback, object state);
/// <summary>
/// Forcefully closes the socket.