Browse Source

增加 TcpServer 和 TcpClient。

master
王厅 1 week ago
parent
commit
0787d32825
  1. 25
      Apewer/Network/Extension.cs
  2. 25
      Apewer/Network/TcpBuffer.cs
  3. 354
      Apewer/Network/TcpClient.cs
  4. 99
      Apewer/Network/TcpInstance.cs
  5. 430
      Apewer/Network/TcpServer.cs

25
Apewer/Network/Extension.cs

@ -4,7 +4,6 @@ using System.Linq;
using System.Net;
using System.Net.NetworkInformation;
using System.Net.Sockets;
using System.Text;
namespace Apewer.Network
{
@ -156,6 +155,30 @@ namespace Apewer.Network
return json;
}
/// <summary>从套接字接收数据。</summary>
/// <exception cref="ArgumentNullException"></exception>
/// <exception cref="ArgumentOutOfRangeException"></exception>
/// <exception cref="SocketException"></exception>
/// <exception cref="ObjectDisposedException"></exception>
/// <exception cref="System.Security.SecurityException"></exception>
public static byte[] Receive(this Socket socket, int maxLength = 1024)
{
if (socket == null) throw new ArgumentNullException(nameof(socket));
var buffer = new byte[maxLength];
var received = socket.Receive(buffer, 0, maxLength, SocketFlags.None);
if (received < maxLength)
{
var newBuffer = new byte[received];
Array.Copy(buffer, newBuffer, received);
return newBuffer;
}
else
{
return buffer;
}
}
}
}

25
Apewer/Network/TcpBuffer.cs

@ -1,25 +0,0 @@
using System;
using System.Collections.Generic;
using System.Net.Sockets;
using System.Text;
namespace Apewer
{
internal class TcpBuffer
{
// TCP 传输缓冲区大小。
public const int Size = 8192;
public Socket Socket;
public byte[] Buffer;
public TcpBuffer()
{
Buffer = new byte[Size];
}
}
}

354
Apewer/Network/TcpClient.cs

@ -13,341 +13,55 @@ namespace Apewer.Network
{
/// <summary>TCP 客户端。</summary>
public class TcpClient
public class TcpClient : IDisposable
{
#region event
Socket _client = null;
/// <summary>Exception。</summary>
public Event<Exception> Excepted { get; set; }
/// <summary>已发送数据。</summary>
public Event<byte[]> Sent { get; set; }
/// <summary>已接收数据。</summary>
public Event<byte[]> Received { get; set; }
/// <summary>套接字实例。</summary>
public Socket Socket { get => _client; }
/// <summary>已连接。</summary>
public Event Connected { get; set; }
/// <summary>已断开。</summary>
public Event Closed { get; set; }
#endregion
#region definition
private Thread _listener = null;
private Thread _provider = null;
private Socket _socket = null;
private Queue<byte[]> _queue = null;
// private AutoResetEvent _are = null;
private bool _break = false;
private int _timeout = 0;
private string _localip, _remoteip;
private int _localport = 0, _remoteport = 0;
private bool _background = true;
/// <summary></summary>
internal object Tag { get; set; }
/// <summary>构造函数。</summary>
public TcpClient()
{
RemoteIP = "127.0.0.1";
RemotePort = 0;
}
/// <summary>构造函数。</summary>
public TcpClient(string ip, int port, int timeout = 0)
{
RemoteIP = ip;
RemotePort = port;
Timeout = timeout;
}
#endregion
#region accessor
/// <summary>获取或设置监听线程是否为后台线程,默认为“是”。</summary>
public bool Background
{
get { return _background; }
set
{
_background = value;
try { if (_listener != null) _listener.IsBackground = value; }
catch (Exception ex) { Excepted?.Invoke(this, ex); }
}
}
/// <summary>获取或设置远程计算机的 IP 地址。</summary>
public string RemoteIP
{
get { return _remoteip; }
set
{
string ip = value;
if (!NetworkUtility.IsIP(ip)) ip = NetworkUtility.Resolve(ip);
if (!NetworkUtility.IsIP(ip)) ip = "127.0.0.1";
_remoteip = ip;
}
}
/// <summary>获取或设置远程计算机的 TCP 端口号。</summary>
public int RemotePort
{
get { return _remoteport; }
set
{
int port = value;
if (port < 0) port = 0;
if (port > 65535) port = 65535;
_remoteport = port;
}
}
/// <summary>获取或设置本地计算机的 IP 地址。</summary>
public string LocalIP
{
get { return _localip; }
private set { _localip = string.IsNullOrEmpty(value) ? "" : value; }
}
/// <summary>获取或设置本地计算机的 TCP 端口号。</summary>
public int LocalPort
{
get { return _localport; }
private set
{
int port = value;
if (port < 0) port = 0;
if (port > 65535) port = 65535;
_localport = port;
}
}
/// <summary>获取或设置超时时间。</summary>
public int Timeout
{
get { return _timeout; }
set { _timeout = (value > 0) ? value : 0; }
}
/// <summary>开始连接,并初始化发送队列。</summary>
public void Start(bool inBackground = false)
{
Close(false);
_queue = new Queue<byte[]>();
if (!inBackground)
{
_provider = new Thread(Provider);
_provider.IsBackground = true;
_provider.Start();
return;
}
Provider();
// _are.WaitOne(1000);
}
/// <summary>断开连接。</summary>
/// <param name="event">是否引导 Closed 事件。</param>
public void Close(bool @event = true)
{
CloseThread(ref _provider);
CloseThread(ref _listener);
CloseSocket();
CloseQueue();
_queue = null;
// _are = null;
if (@event) Closed?.Invoke(this);
}
/// <summary>向服务端发送数据。</summary>
/// <param name="bytes">字节数组。</param>
public bool Send(byte[] bytes)
{
if (_socket != null)
{
_queue.Enqueue(bytes);
return true;
}
else
{
var ex = new SocketException(10057);
Excepted?.Invoke(this, ex);
return false;
}
}
public bool Connected { get => _client != null && _client.Connected; }
/// <summary>是否已连接。</summary>
public bool Online
{
get
{
try { if (_socket != null) return _socket.Connected; }
catch (Exception ex) { Excepted?.Invoke(this, ex); }
return false;
}
}
#endregion
#region logic
private void Listener()
{
var buffer = new byte[TcpBuffer.Size];
var length = 0;
while (true)
{
try
{
if (_socket.Poll(50, SelectMode.SelectWrite))
{
_socket.Blocking = true;
length = _socket.Receive(buffer);
if (length > 0)
{
var bytes = new byte[length];
Array.Copy(buffer, bytes, length);
Received?.Invoke(this, bytes);
}
else
{
if (_socket != null) break;
}
}
}
catch (SocketException ex)
{
if (ex.ErrorCode != 10060) Excepted?.Invoke(this, ex);
if (ex.ErrorCode == 10054) break;
}
catch (Exception ex)
{
Excepted?.Invoke(this, ex);
// break;
}
}
if (_provider != null)
{
if (_provider.IsAlive) _provider.Abort();
_provider = null;
}
if (_socket != null)
{
_socket.Disconnect(true);
_socket.Close();
_socket = null;
}
CloseThread(ref _provider);
CloseSocket();
CloseQueue();
Closed?.Invoke(this);
}
private void Provider()
{
_break = false;
_socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
if (_timeout > 0) _socket.SendTimeout = _timeout;
if (_timeout > 0) _socket.ReceiveTimeout = _timeout;
//_are = new AutoResetEvent(false);
try
{
_socket.Blocking = true;
_socket.Connect(_remoteip, _remoteport);
if (_socket.Connected)
{
_socket.SendBufferSize = TcpBuffer.Size;
_socket.ReceiveBufferSize = TcpBuffer.Size;
var rep = (IPEndPoint)_socket.RemoteEndPoint;
var lep = (IPEndPoint)_socket.LocalEndPoint;
_remoteip = rep.Address.ToString();
_localip = lep.Address.ToString();
_localport = lep.Port;
Connected?.Invoke(this);
_listener = new Thread(Listener);
_listener.Start();
// _are.Set();
}
else
{
Close();
return;
}
}
catch (Exception ex)
{
Excepted?.Invoke(this, ex);
}
while ((!_break) && (_socket != null))
{
/// <summary>启动客户端,并连接到服务端。</summary>
public TcpClient(string ip, int port) : this(IPAddress.Parse(ip), port) { }
if (_queue.Count > 0)
{
var bytes = _queue.Dequeue();
if (bytes.Length > 0)
{
try
{
_socket.Send(bytes);
Sent?.Invoke(this, bytes);
}
catch (Exception ex)
{
Excepted?.Invoke(this, ex);
}
}
else
{
Thread.Sleep(1);
}
}
}
}
/// <summary>启动客户端,并连接到服务端。</summary>
public TcpClient(IPAddress address, int port) : this(new IPEndPoint(address, port)) { }
private void CloseSocket()
/// <summary>启动客户端,并连接到服务端。</summary>
public TcpClient(IPEndPoint endpoint)
{
if (_socket != null)
{
try
{
_socket.Disconnect(false);
_socket.Close();
}
catch (SocketException ex) { Excepted?.Invoke(this, ex); }
catch (Exception ex) { Excepted?.Invoke(this, ex); }
_socket = null;
}
_client = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
_client.Connect(endpoint);
_client.SendTimeout = 5000;
}
private void CloseThread(ref Thread thread)
/// <summary>关闭连接,释放系统资源。</summary>
public void Dispose()
{
if (thread != null)
{
if (thread.IsAlive) thread.Abort();
thread = null;
}
try { _client.Disconnect(false); } catch { }
try { _client.Close(100); } catch { }
#if !NET20
try { _client.Dispose(); } catch { }
#endif
}
private void CloseQueue()
{
if (_queue != null) _queue.Clear();
}
/// <summary>接收。</summary>
/// <exception cref="ArgumentNullException"></exception>
/// <exception cref="ArgumentOutOfRangeException"></exception>
/// <exception cref="SocketException"></exception>
/// <exception cref="ObjectDisposedException"></exception>
/// <exception cref="System.Security.SecurityException"></exception>
public byte[] Receive(int maxLength = 1024) => _client.Receive(maxLength);
#endregion
/// <summary>发送。</summary>
/// <exception cref="ArgumentNullException" />
/// <exception cref="SocketException" />
/// <exception cref="ObjectDisposedException" />
public void Send(byte[] data) => _client.Send(data);
}
}

99
Apewer/Network/TcpInstance.cs

@ -1,99 +0,0 @@
using Apewer.Network;
using System;
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;
namespace Apewer
{
internal class TcpInstance
{
private TcpServer _parent;
private Socket _server;
private Socket _client;
private IPEndPoint _ep;
private string _ip;
private int _port;
private ManualResetEvent _done = new ManualResetEvent(false);
private int _length = 0;
private byte[] _data;
public TcpInstance(TcpServer parent, Socket server, Socket client)
{
_parent = parent;
_server = server;
_client = client;
_ep = (IPEndPoint)_client.RemoteEndPoint;
_ip = _ep.Address.ToString();
_port = _ep.Port;
}
public void Process()
{
try
{
var vm = new TcpBuffer();
vm.Socket = _client;
vm.Socket.BeginReceive(vm.Buffer, 0, vm.Buffer.Length, SocketFlags.None, Callback, vm);
_done.WaitOne();
}
catch (Exception argex) { _parent.RaiseExcepted(argex); }
}
private void Callback(IAsyncResult ar)
{
try
{
var buffer = (TcpBuffer)ar.AsyncState;
if (buffer.Socket.Connected)
{
_length = buffer.Socket.EndReceive(ar);
if (_length > 0)
{
_data = new byte[_length];
Array.Copy(buffer.Buffer, 0, _data, 0, _length);
_parent.RaiseReceived(_ip, _port, _data);
buffer.Socket.BeginReceive(buffer.Buffer, 0, buffer.Buffer.Length, SocketFlags.None, Callback, buffer);
}
else
{
if (_client != null)
{
if (_client.Connected) _client.Close();
_client = null;
}
if (_ep != null)
{
_parent.Remove(_ip + "-" + _port.ToString());
_parent.RaiseClosed(_ip, _port);
}
}
}
else
{
if (!string.IsNullOrEmpty(_ip)) _parent.Close(_ip, _port);
}
}
catch (Exception ex)
{
_parent.RaiseExcepted(ex);
if (_client != null)
{
if (_client.Connected) _client.Close();
_client = null;
}
_parent.Close(_ip, _port);
}
_done.Set();
}
}
}

430
Apewer/Network/TcpServer.cs

@ -1,401 +1,171 @@
using System;
using System.Collections;
using System.Text;
using Apewer.Internals;
using System;
using System.Collections.Generic;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using System.IO;
using Apewer.Internals;
using Apewer;
namespace Apewer.Network
{
/// <summary>TCP 服务端。</summary>
internal class TcpServer
public sealed class TcpServer : IDisposable
{
#region event
Socket _server = null;
List<Socket> _clients = new List<Socket>(1024);
/// <summary>Exception。</summary>
public Event<Exception> Excepted { get; set; }
/// <summary>获取当前监听的端口。</summary>
public int Port { get => GetLocalPort(); }
/// <summary>服务端已启动。</summary>
public Event<SocketEndPoint> Started { get; set; }
/// <summary>启动服务端,并在所有接口上监听端口。</summary>
public TcpServer(int port) : this(IPAddress.Any, port) { }
/// <summary>服务端已关闭。</summary>
public Event<SocketEndPoint> Quitted { get; set; }
/// <summary>启动服务端,并监听端口。</summary>
public TcpServer(string ip, int port) : this(IPAddress.Parse(ip), port) { }
/// <summary>客户端已连接。</summary>
public Event<SocketEndPoint> Connected { get; set; }
/// <summary>启动服务端,并监听端口。</summary>
public TcpServer(IPAddress address, int port) : this(new IPEndPoint(address, port)) { }
/// <summary>客户端已断开。</summary>
public Event<SocketEndPoint> Closed { get; set; }
/// <summary>已收到客户端数据。</summary>
public Event<SocketReceived> Received { get; set; }
#region raise
internal void RaiseExcepted(Exception exception)
/// <summary>启动服务端,并监听端口。</summary>
public TcpServer(IPEndPoint endpoint)
{
if (Excepted != null) Excepted(this, exception);
_server = new Socket(endpoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
_server.Bind(endpoint);
_server.Listen(500);
_server.SendTimeout = 5000;
}
internal void RaiseStarted()
/// <summary>关闭连接,释放系统资源。</summary>
public void Dispose()
{
if (Started != null)
lock (_clients)
{
var ip = _endpoint == null ? "" : _endpoint.Address.ToString();
var port = _endpoint == null ? 0 : _endpoint.Port;
Started?.Invoke(this, new SocketEndPoint(ip, port));
}
}
internal void RaiseConnected(string ip, int port)
{
Connected?.Invoke(this, new SocketEndPoint(ip, port));
}
internal void RaiseClosed(string ip, int port)
{
Closed?.Invoke(this, new SocketEndPoint(ip, port));
}
internal void RaiseQuitted()
{
var quitted = Quitted;
if (quitted != null)
{
var ip = _endpoint == null ? "" : _endpoint.Address.ToString();
var port = _endpoint == null ? 0 : _endpoint.Port;
quitted(this, new SocketEndPoint(ip, port));
}
}
internal void RaiseReceived(string ip, int port, byte[] bytes)
{
Received?.Invoke(this, new SocketReceived(ip, port, bytes));
}
#endregion
#endregion
#region definition
private Socket _socket = null;
private Thread _listener = null;
private SortedList _client = null;
private IPEndPoint _endpoint = null;
private int _port = 0;
private int _max = 0;
private int _timeout = 0;
private bool _state = false;
private bool _background = true;
/// <summary>构造函数。</summary>
public TcpServer(int port = 0)
{
Port = port;
Max = 1000;
Timeout = 1000;
}
#endregion
#region accessor
/// <summary>获取或设置监听线程是否为后台线程,默认为“是”。</summary>
public bool Background
{
get { return _background; }
set
{
_background = value;
try { if (_listener != null) _listener.IsBackground = value; }
catch (Exception ex) { RaiseExcepted(ex); }
}
}
/// <summary>端口。</summary>
public int Port
{
get
{
int port = _port;
if (port > 65535) port = 65535;
if (port < 0) port = 0;
return port;
}
set
{
int port = value;
if (port < 0) port = 0;
if (port > 65535) port = 65535;
_port = port;
}
}
/// <summary>最大客户端数量。</summary>
public int Max
{
get { return _max; }
set { _max = (value > 0) ? value : 0; }
}
/// <summary>发送数据和接收数据的超时时间。</summary>
public int Timeout
{
get { return _timeout; }
set { _timeout = (value > 0) ? value : 0; }
}
/// <summary>服务端正在运行。</summary>
public bool Alive
{
get { return (_listener != null) ? _listener.IsAlive : false; }
}
/// <summary>已连接的客户端数量。</summary>
public int Count
{
get { return (_client != null) ? _client.Count : 0; }
}
/// <summary>启动服务端。</summary>
public bool Start(bool inBackground = true)
{
try
{
_endpoint = new IPEndPoint(IPAddress.Any, Port);
_client = new SortedList();
_socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
_socket.Bind(_endpoint);
_socket.Listen(Max);
_socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.AcceptConnection, 1);
_socket.ReceiveTimeout = Timeout;
_socket.SendTimeout = Timeout;
if (inBackground)
{
_listener = new Thread(Listener);
_listener.IsBackground = Background;
_listener.Start();
_state = _listener.IsAlive;
var cep = (IPEndPoint)_socket.LocalEndPoint;
_port = cep.Port;
RaiseStarted();
return true;
}
else
foreach (var client in _clients)
{
_state = true;
var cep = (IPEndPoint)_socket.LocalEndPoint;
_port = cep.Port;
RaiseStarted();
Listener();
return true;
try { client.Disconnect(false); } catch { }
try { client.Close(100); } catch { }
#if !NET20
try { client.Dispose(); } catch { }
#endif
}
_clients.Clear();
}
catch (Exception ex)
if (_server != null)
{
_endpoint = null;
RaiseExcepted(ex);
if ((_socket != null) && _socket.Connected) _socket.Close();
_state = false;
Quit();
return false;
try { _server.Disconnect(false); } catch { }
try { _server.Close(100); } catch { }
#if !NET20
try { _server.Dispose(); } catch { }
#endif
}
}
/// <summary>关闭服务端。</summary>
public void Quit()
/// <summary>获取当前所有客户端连接。</summary>
public Socket[] GetClients()
{
Quit(true);
}
/// <summary>断开与所有客户端的连接。</summary>
public void Close()
{
if (_client != null)
lock (_clients)
{
foreach (Socket socket in _client.Values)
{
try
{
var ep = (IPEndPoint)socket.RemoteEndPoint;
Close(ep.Address.ToString(), ep.Port);
}
catch (Exception ex)
{
RaiseExcepted(ex);
}
}
var array = _clients.ToArray();
return array;
}
}
/// <summary>断开与指定客户端的连接。</summary>
public void Close(string ip, int port)
/// <summary>等待接收客户端连接。此方法将会阻塞当前线程直到关闭监听的端口。</summary>
/// <param name="callback">处理客户端连接,当此回调结束时将释放此连接。</param>
/// <exception cref="ArgumentNullException"></exception>
public void Accept(Action<Socket> callback)
{
if (port < 0) port = 0;
if (port > 65535) port = 65535;
if (callback == null) throw new ArgumentNullException(nameof(callback));
if (!string.IsNullOrEmpty(ip))
while (true)
{
var ck = ip + "-" + port.ToString();
var cs = Client(ip, port);
var socket = null as Socket;
try
{
if (cs != null) cs.Close();
Remove(ck);
RaiseClosed(ip, port);
socket = _server.Accept();
if (socket != null)
{
Callback.Enqueue(this, socket, callback);
}
}
catch (Exception ex)
catch
{
RaiseExcepted(ex);
if (socket != null)
{
try { socket.Close(); } catch { }
try { socket.Disconnect(false); } catch { }
#if !NET20
try { socket.Dispose(); } catch { }
#endif
}
}
}
}
/// <summary>向所有客户端广播数据。</summary>
public void Send(byte[] bytes)
int GetLocalPort()
{
var length = bytes.Length;
if ((_client.Count > 0) && (length > 0))
var socket = _server;
if (socket != null)
{
foreach (Socket i in _client.Values) Send(bytes, i);
var endpoint = socket.LocalEndPoint as IPEndPoint;
if (endpoint != null) return endpoint.Port;
}
return 0;
}
/// <summary>向指定客户端发送数据。</summary>
public bool Send(byte[] bytes, string ip, int port)
sealed class Callback
{
if (port < 0) port = 0;
if (port > 65535) port = 65535;
if (string.IsNullOrEmpty(ip)) return false;
return Send(bytes, Client(ip, port));
}
#endregion
#region logic
TcpServer _server;
Socket _socket;
Action<Socket> _action;
/// <summary>
///
/// </summary>
/// <param name="event"></param>
private void Quit(bool @event)
{
Close();
if (_listener != null)
{
if (_listener.IsAlive) _listener.Abort();
_listener = null;
}
if (_socket != null)
public static void Enqueue(TcpServer server, Socket socket, Action<Socket> callback)
{
_socket.Close();
_socket = null;
}
if (_client != null)
{
_client.Clear();
_client = null;
}
_endpoint = null;
if (@event) RaiseQuitted();
}
var state = new Callback();
state._server = server ?? throw new ArgumentNullException(nameof(server));
state._socket = socket ?? throw new ArgumentNullException(nameof(socket));
state._action = callback ?? throw new ArgumentNullException(nameof(callback));
private void Close(Socket socket)
{
if (socket != null)
{
try { socket.Close(); }
catch (Exception ex) { RaiseExcepted(ex); }
ThreadPool.QueueUserWorkItem(InThreadPool, state);
}
}
internal void Remove(string key)
{
if ((_client != null) && (!string.IsNullOrEmpty(key)))
static void InThreadPool(object state)
{
if (_client.ContainsKey(key))
var instance = state as Callback;
if (instance == null) return;
if (instance._server == null) return;
if (instance._socket == null) return;
if (instance._action == null) return;
var socket = instance._socket;
lock (instance._server._clients)
{
try
{
Close((Socket)_client[key]);
_client.Remove(key);
}
catch (Exception ex)
{
RaiseExcepted(ex);
}
instance._server._clients.Add(instance._socket);
}
}
}
private bool Send(byte[] bytes, Socket client)
{
var length = bytes.Length;
if ((client != null) && (length > 0))
{
try
{
return (client.Send(bytes, length, SocketFlags.None) > 0) ? true : false;
instance._action.Invoke(instance._socket);
}
catch (Exception ex) { RaiseExcepted(ex); }
}
return false;
}
private void Listener()
{
while (_socket != null)
{
try
catch
{ }
finally
{
var socket = _socket.Accept();
if (socket != null)
lock (instance._server._clients)
{
var ep = (IPEndPoint)socket.RemoteEndPoint;
var key = ep.Address.ToString() + "-" + ep.Port.ToString();
_client.Add(key, socket);
var instance = new TcpInstance(this, _socket, socket);
var thread = new Thread(instance.Process);
thread.IsBackground = true;
thread.Name = key;
thread.Start();
RaiseConnected(ep.Address.ToString(), ep.Port);
instance._server._clients.Remove(instance._socket);
}
}
catch (Exception ex)
{
RaiseExcepted(ex);
try { socket.Disconnect(false); } catch { }
try { socket.Close(100); } catch { }
#if !NET20
try { socket.Dispose(); } catch { }
#endif
}
}
Quit(false);
}
private Socket Client(string ip, int port)
{
if (port < 0) port = 0;
if (port > 65535) port = 65535;
try
{
var ck = ip + "-" + port.ToString();
if (_client.ContainsKey(ck)) return (Socket)_client[ck];
}
catch (Exception ex)
{
RaiseExcepted(ex);
}
return null;
}
#endregion
}
}

Loading…
Cancel
Save