g82tt/ThingsGateway
1
0
Fork 0
mirror of https://gitee.com/ThingsGateway/ThingsGateway.git synced 2025-10-24 12:20:28 +08:00
Code Issues Packages Projects Releases Wiki Activity

更新touchsocket依赖正式版本

Browse Source
This commit is contained in:
Diego2098
2024-03-09 15:19:19 +08:00
parent 8d0881632c
commit d98d405009
15 changed files with 114 additions and 1829 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/Directory.Build.props
View File

@@ -5,7 +5,7 @@
<LangVersion>11.0</LangVersion>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
<Version>5.0.2.23</Version>
<Version>5.0.3.0</Version>
<Authors>Diego</Authors>
<Company>Diego</Company>
<Product>Diego</Product>

40
src/ThingsGateway.Gateway.ApiController/ThingsGateway.Gateway.ApiController.xml
View File

@@ -292,6 +292,11 @@
<param name="fromType">注册类型</param>
<param name="toType">实例类型</param>
</member>
<member name="T:TouchSocket.Core.GeneratorPackageAttribute">
<summary>
标识源生成<see cref="T:TouchSocket.Core.IPackage"/>的实现。
</summary>
</member>
<member name="T:TouchSocket.Core.GeneratorPluginAttribute">
<summary>
使用源生成插件的调用。
@@ -303,11 +308,6 @@
</summary>
<param name="pluginName">插件名称，一般建议使用nameof()解决。</param>
</member>
<member name="T:TouchSocket.Rpc.GeneratorRpcMethodAttribute">
<summary>
标识该接口方法将自动生成调用的扩展方法
</summary>
</member>
<member name="T:TouchSocket.Rpc.GeneratorRpcProxyAttribute">
<summary>
标识该接口将使用源生成自动生成调用的代理类
@@ -343,11 +343,6 @@
生成代码
</summary>
</member>
<member name="P:TouchSocket.Rpc.GeneratorRpcProxyAttribute.MethodFlags">
<summary>
函数标识
</summary>
</member>
<member name="P:TouchSocket.Rpc.GeneratorRpcProxyAttribute.InheritedInterface">
<summary>
继承接口
@@ -358,5 +353,30 @@
标识将通过源生成器生成Rpc服务的调用委托。
</summary>
</member>
<member name="T:TouchSocket.Rpc.GeneratorRpcServerRegisterAttribute">
<summary>
标识将通过源生成器生成Rpc服务的注册代码。
</summary>
</member>
<member name="P:TouchSocket.Rpc.GeneratorRpcServerRegisterAttribute.MethodName">
<summary>
方法名称。默认是“RegisterAllFrom+AssemblyName”
</summary>
</member>
<member name="P:TouchSocket.Rpc.GeneratorRpcServerRegisterAttribute.ClassName">
<summary>
扩展类类名，默认是“RegisterRpcServerFrom+AssemblyName+Extension”
</summary>
</member>
<member name="P:TouchSocket.Rpc.GeneratorRpcServerRegisterAttribute.Accessibility">
<summary>
访问修饰。
<para>
如果为<see cref="F:TouchSocket.Rpc.Accessibility.Both"/>，将生成注册公共Rpc服务与非公共服务两个方法。其中非公共方法会在<see cref="P:TouchSocket.Rpc.GeneratorRpcServerRegisterAttribute.MethodName"/>之前以Internal开头。
如果为<see cref="F:TouchSocket.Rpc.Accessibility.Internal"/>，将只生成注册非公共Rpc服务。
如果为<see cref="F:TouchSocket.Rpc.Accessibility.Public"/>，将只生成注册公共Rpc服务。
</para>
</summary>
</member>
</members>
</doc>

4
src/ThingsGateway.Gateway.Application/Plugin/DriverBase.cs
View File

@@ -169,11 +169,11 @@ public abstract class DriverBase : DisposableObject
}
try
{
await DisposeAsync();
Dispose();
}
catch (Exception ex)
{
Logger?.LogError(ex, $"插件执行{nameof(DisposeAsync)}方法失败");
Logger?.LogError(ex, $"插件执行{nameof(Dispose)}方法失败");
}
Logger?.LogInformation($"{DeviceName}：线程已停止");
}

2
src/ThingsGateway.Gateway.Application/ThingsGateway.Gateway.Application.csproj
View File

@@ -25,7 +25,7 @@
<PackageReference Include="CS-Script" Version="4.8.14" />
<!--CS-Script与Furion冲突，直接安装覆盖版本-->
<PackageReference Include="Microsoft.CodeAnalysis.CSharp.Scripting" Version="4.9.2" />
<PackageReference Include="TouchSocket.Dmtp" Version="2.0.0-rc.2" />
<PackageReference Include="TouchSocket.Dmtp" Version="2.0.0" />
</ItemGroup>

8
src/foundation/ThingsGateway.Foundation.AdapterConsle/src/ThingsGateway.Foundation.AdapterConsle.csproj
View File

@@ -6,10 +6,10 @@
<DocumentationFile></DocumentationFile>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="ThingsGateway.Foundation.Modbus" Version="5.0.2.20" />
<PackageReference Include="ThingsGateway.Foundation.Dlt645" Version="5.0.2.20" />
<PackageReference Include="ThingsGateway.Foundation.SiemensS7" Version="5.0.2.20" />
<PackageReference Include="ThingsGateway.Foundation.Variable" Version="5.0.2.20" />
<PackageReference Include="ThingsGateway.Foundation.Modbus" Version="5.0.2.23" />
<PackageReference Include="ThingsGateway.Foundation.Dlt645" Version="5.0.2.23" />
<PackageReference Include="ThingsGateway.Foundation.SiemensS7" Version="5.0.2.23" />
<PackageReference Include="ThingsGateway.Foundation.Variable" Version="5.0.2.23" />
</ItemGroup>
<!--<ItemGroup>
<ProjectReference Include="..\..\ThingsGateway.Foundation.Dlt645\src\ThingsGateway.Foundation.Dlt645.csproj" />

14
src/foundation/ThingsGateway.Foundation/src/Channel/TcpClient/Extension/TcpClientConfigExtension.cs
View File

@@ -32,18 +32,6 @@ namespace ThingsGateway.Foundation
return tgTcpClient;
}
/// <summary>
/// 获取一个新的Tcp服务端通道。传入绑定地址
/// </summary>
/// <returns></returns>
public static TgTcpService GetTcpServiceWithBindIPHost(this TouchSocketConfig config, IPHost bindUrl)
{
if (bindUrl == null) throw new ArgumentNullException(nameof(IPHost));
config.SetListenIPHosts(new IPHost[] { bindUrl });
//载入配置
TgTcpService tgTcpService = new TgTcpService();
tgTcpService.Setup(config);
return tgTcpService;
}
}
}

34
src/foundation/ThingsGateway.Foundation/src/Channel/TcpClient/TgTcpClientBase.cs
View File

@@ -28,7 +28,6 @@ namespace ThingsGateway.Foundation
{
this.Protocol = Protocol.Tcp;
}
/// <summary>
/// Tcp客户端
/// </summary>
@@ -36,14 +35,13 @@ namespace ThingsGateway.Foundation
{
Dispose(true);
}
#region 变量
private DelaySender m_delaySender;
private volatile bool m_online;
private readonly EasyLock m_semaphoreForConnect = new();
private readonly TcpCore m_tcpCore = new();
private readonly object m_lock = new object();
#endregion 变量
#region 事件
@@ -215,10 +213,6 @@ namespace ThingsGateway.Foundation
/// <inheritdoc/>
public int Port { get; private set; }
/// <inheritdoc/>
[Obsolete("该配置已被弃用，正式版发布时会直接删除", true)]
public ReceiveType ReceiveType { get; private set; }
/// <inheritdoc/>
public bool UseSsl => this.GetTcpCore().UseSsl;
@@ -238,7 +232,7 @@ namespace ThingsGateway.Foundation
/// <inheritdoc/>
public virtual void Close(string msg)
{
lock (this.GetTcpCore())
lock (this.m_lock)
{
if (this.m_online)
{
@@ -256,14 +250,20 @@ namespace ThingsGateway.Foundation
protected override void Dispose(bool disposing)
{
if (DisposedValue) return;
lock (this.GetTcpCore())
if (disposing)
{
if (this.m_online)
lock (this.m_lock)
{
this.PrivateOnDisconnecting(new DisconnectEventArgs(true, string.Format(FoundationConst.ProactivelyDisconnect, nameof(Dispose)))).GetFalseAwaitResult();
this.BreakOut(true, string.Format(FoundationConst.ProactivelyDisconnect, nameof(Dispose)));
if (this.m_online)
{
this.PrivateOnDisconnecting(new DisconnectEventArgs(true, string.Format(FoundationConst.ProactivelyDisconnect, nameof(Dispose)))).GetFalseAwaitResult();
this.BreakOut(true, string.Format(FoundationConst.ProactivelyDisconnect, nameof(Dispose)));
}
}
this.m_tcpCore.SafeDispose();
}
base.Dispose(disposing);
}
@@ -457,7 +457,7 @@ namespace ThingsGateway.Foundation
/// <inheritdoc/>
public IReceiver CreateReceiver()
{
return this.m_receiver ??= new TgReceiver(this);
return this.m_receiver ??= new(this);
}
/// <inheritdoc/>
@@ -467,7 +467,6 @@ namespace ThingsGateway.Foundation
}
#endregion Receiver
/// <inheritdoc/>
public override string ToString()
{
@@ -486,7 +485,7 @@ namespace ThingsGateway.Foundation
/// <param name="msg"></param>
protected void BreakOut(bool manual, string msg)
{
lock (this.GetTcpCore())
lock (this.m_lock)
{
if (this.m_online)
{
@@ -839,6 +838,7 @@ namespace ThingsGateway.Foundation
{
this.m_delaySender = new DelaySender(delaySenderOption, this.GetTcpCore().Send);
}
this.m_tcpCore.Reset(socket);
this.m_tcpCore.OnReceived = this.HandleReceived;
this.m_tcpCore.OnBreakOut = this.TcpCoreBreakOut;
@@ -892,9 +892,5 @@ namespace ThingsGateway.Foundation
}
return Task.FromResult(false);
}
/// <inheritdoc/>
[Obsolete("该配置已被弃用，正式版发布时会直接删除", true)]
public Stream? GetStream() => default;
}
}

32
src/foundation/ThingsGateway.Foundation/src/Channel/TcpService/Extension/TcpServiceConfigExtension.cs Normal file
View File

@@ -0,0 +1,32 @@
//------------------------------------------------------------------------------
// 此代码版权声明为全文件覆盖，如有原作者特别声明，会在下方手动补充
// 此代码版权（除特别声明外的代码）归作者本人Diego所有
// 源代码使用协议遵循本仓库的开源协议及附加协议
// Gitee源代码仓库：https://gitee.com/diego2098/ThingsGateway
// Github源代码仓库：https://github.com/kimdiego2098/ThingsGateway
// 使用文档：https://diego2098.gitee.io/thingsgateway-docs/
// QQ群：605534569
//------------------------------------------------------------------------------
namespace ThingsGateway.Foundation
{
/// <summary>
/// TcpServiceConfigExtension
/// </summary>
public static class TcpServiceConfigExtension
{
/// <summary>
/// 获取一个新的Tcp服务端通道。传入绑定地址
/// </summary>
/// <returns></returns>
public static TgTcpService GetTcpServiceWithBindIPHost(this TouchSocketConfig config, IPHost bindUrl)
{
if (bindUrl == null) throw new ArgumentNullException(nameof(IPHost));
config.SetListenIPHosts(new IPHost[] { bindUrl });
//载入配置
TgTcpService tgTcpService = new TgTcpService();
tgTcpService.Setup(config);
return tgTcpService;
}
}
}

84
src/foundation/ThingsGateway.Foundation/src/Channel/TcpService/SocketClientCollection.cs
View File

@@ -1,84 +0,0 @@
//------------------------------------------------------------------------------
// 此代码版权声明为全文件覆盖，如有原作者特别声明，会在下方手动补充
// 此代码版权（除特别声明外的代码）归作者本人Diego所有
// 源代码使用协议遵循本仓库的开源协议及附加协议
// Gitee源代码仓库：https://gitee.com/diego2098/ThingsGateway
// Github源代码仓库：https://github.com/kimdiego2098/ThingsGateway
// 使用文档：https://diego2098.gitee.io/thingsgateway-docs/
// QQ群：605534569
//------------------------------------------------------------------------------
using System.Collections.Concurrent;
using System.Diagnostics;
namespace ThingsGateway.Foundation
{
[DebuggerDisplay("Count={Count}")]
internal class SocketClientCollection : ConcurrentDictionary<string, ISocketClient>, ISocketClientCollection
{
public IEnumerable<ISocketClient> GetClients()
{
return this.Values;
}
public IEnumerable<string> GetIds()
{
return this.Keys;
}
public bool SocketClientExist(string id)
{
return string.IsNullOrEmpty(id) ? false : this.ContainsKey(id);
}
public bool TryGetSocketClient(string id, out ISocketClient socketClient)
{
if (string.IsNullOrEmpty(id))
{
socketClient = null;
return false;
}
return this.TryGetValue(id, out socketClient);
}
public bool TryGetSocketClient<TClient>(string id, out TClient socketClient) where TClient : ISocketClient
{
if (string.IsNullOrEmpty(id))
{
socketClient = default;
return false;
}
if (this.TryGetValue(id, out var client))
{
socketClient = (TClient)client;
return true;
}
socketClient = default;
return false;
}
internal bool TryAdd(ISocketClient socketClient)
{
return this.TryAdd(socketClient.Id, socketClient);
}
internal bool TryRemove<TClient>(string id, out TClient socketClient) where TClient : ISocketClient
{
if (string.IsNullOrEmpty(id))
{
socketClient = default;
return false;
}
if (this.TryRemove(id, out var client))
{
socketClient = (TClient)client;
return true;
}
socketClient = default;
return false;
}
}
}

768
src/foundation/ThingsGateway.Foundation/src/Channel/TcpService/TcpServiceBase.cs
View File

@@ -1,768 +0,0 @@
//------------------------------------------------------------------------------
// 此代码版权声明为全文件覆盖，如有原作者特别声明，会在下方手动补充
// 此代码版权（除特别声明外的代码）归作者本人Diego所有
// 源代码使用协议遵循本仓库的开源协议及附加协议
// Gitee源代码仓库：https://gitee.com/diego2098/ThingsGateway
// Github源代码仓库：https://github.com/kimdiego2098/ThingsGateway
// 使用文档：https://diego2098.gitee.io/thingsgateway-docs/
// QQ群：605534569
//------------------------------------------------------------------------------
using System.Net.Sockets;
using TouchSocket.Resources;
namespace ThingsGateway.Foundation
{
/// <summary>
/// Tcp泛型服务器，由使用者自己指定<see cref="TgSocketClientBase"/>类型。
/// </summary>
public class TcpServiceBase<TClient> : TcpServiceBase, ITcpService<TClient> where TClient : TgSocketClientBase, new()
{
/// <summary>
/// Tcp泛型服务器，由使用者自己指定<see cref="TgSocketClientBase"/>类型。
/// </summary>
public TcpServiceBase()
{
this.m_getDefaultNewId = this.GetDefaultNewId;
}
#region 变量
private readonly List<TcpNetworkMonitor> m_monitors = new List<TcpNetworkMonitor>();
private readonly SocketClientCollection m_socketClients = new SocketClientCollection();
private Func<string> m_getDefaultNewId;
private int m_maxCount;
private long m_nextId;
private ServerState m_serverState;
#endregion 变量
#region 属性
/// <inheritdoc/>
public override int MaxCount => this.m_maxCount;
/// <inheritdoc/>
public override IEnumerable<TcpNetworkMonitor> Monitors => this.m_monitors.ToArray();
/// <inheritdoc/>
public override string ServerName => this.Config?.GetValue(TouchSocketConfigExtension.ServerNameProperty);
/// <inheritdoc/>
public override ServerState ServerState => this.m_serverState;
/// <inheritdoc/>
public override ISocketClientCollection SocketClients => this.m_socketClients;
#endregion 属性
#region 事件
/// <summary>
/// 用户连接完成
/// </summary>
public ConnectedEventHandler<TClient> Connected { get; set; }
/// <summary>
/// 有用户连接的时候
/// </summary>
public ConnectingEventHandler<TClient> Connecting { get; set; }
/// <summary>
/// 有用户断开连接
/// </summary>
public DisconnectEventHandler<TClient> Disconnected { get; set; }
/// <summary>
/// 即将断开连接(仅主动断开时有效)。
/// </summary>
public DisconnectEventHandler<TClient> Disconnecting { get; set; }
internal Task OnInternalConnected(ISocketClient socketClient, ConnectedEventArgs e)
{
return this.OnClientConnected(socketClient, e);
}
internal Task OnInternalConnecting(ISocketClient socketClient, ConnectingEventArgs e)
{
return this.OnClientConnecting(socketClient, e);
}
internal Task OnInternalDisconnected(ISocketClient socketClient, DisconnectEventArgs e)
{
return this.OnClientDisconnected(socketClient, e);
}
internal Task OnInternalDisconnecting(ISocketClient socketClient, DisconnectEventArgs e)
{
return this.OnClientDisconnecting(socketClient, e);
}
internal Task OnInternalReceivedData(ISocketClient socketClient, ReceivedDataEventArgs e)
{
return this.OnClientReceivedData(socketClient, e);
}
/// <summary>
/// <inheritdoc/>
/// </summary>
/// <param name="socketClient"></param>
/// <param name="e"></param>
protected override sealed Task OnClientConnected(ISocketClient socketClient, ConnectedEventArgs e)
{
return this.OnConnected((TClient)socketClient, e);
}
/// <summary>
/// <inheritdoc/>
/// </summary>
/// <param name="socketClient"></param>
/// <param name="e"></param>
protected override sealed Task OnClientConnecting(ISocketClient socketClient, ConnectingEventArgs e)
{
return this.OnConnecting((TClient)socketClient, e);
}
/// <summary>
/// <inheritdoc/>
/// </summary>
/// <param name="socketClient"></param>
/// <param name="e"></param>
protected override sealed Task OnClientDisconnected(ISocketClient socketClient, DisconnectEventArgs e)
{
return this.OnDisconnected((TClient)socketClient, e);
}
/// <summary>
/// <inheritdoc/>
/// </summary>
/// <param name="socketClient"></param>
/// <param name="e"></param>
protected override sealed Task OnClientDisconnecting(ISocketClient socketClient, DisconnectEventArgs e)
{
return this.OnDisconnecting((TClient)socketClient, e);
}
/// <summary>
/// <inheritdoc/>
/// </summary>
/// <param name="socketClient"></param>
/// <param name="e"></param>
protected override sealed Task OnClientReceivedData(ISocketClient socketClient, ReceivedDataEventArgs e)
{
return this.OnReceived((TClient)socketClient, e);
}
/// <summary>
/// 客户端连接完成，覆盖父类方法将不会触发事件。
/// </summary>
/// <param name="socketClient"></param>
/// <param name="e"></param>
protected virtual async Task OnConnected(TClient socketClient, ConnectedEventArgs e)
{
if (this.Connected != null)
{
await this.Connected.Invoke(socketClient, e);
}
}
/// <summary>
/// 客户端请求连接，覆盖父类方法将不会触发事件。
/// </summary>
/// <param name="socketClient"></param>
/// <param name="e"></param>
protected virtual async Task OnConnecting(TClient socketClient, ConnectingEventArgs e)
{
if (this.Connecting != null)
{
await this.Connecting.Invoke(socketClient, e);
}
}
/// <summary>
/// 客户端断开连接，覆盖父类方法将不会触发事件。
/// </summary>
/// <param name="socketClient"></param>
/// <param name="e"></param>
protected virtual async Task OnDisconnected(TClient socketClient, DisconnectEventArgs e)
{
if (this.Disconnected != null)
{
await this.Disconnected.Invoke(socketClient, e);
}
}
/// <summary>
/// 即将断开连接(仅主动断开时有效)。
/// </summary>
/// <param name="socketClient"></param>
/// <param name="e"></param>
protected virtual async Task OnDisconnecting(TClient socketClient, DisconnectEventArgs e)
{
if (this.Disconnecting != null)
{
await this.Disconnecting.Invoke(socketClient, e);
}
}
/// <summary>
/// 当收到适配器数据。
/// </summary>
/// <param name="socketClient"></param>
/// <param name="e"></param>
protected virtual Task OnReceived(TClient socketClient, ReceivedDataEventArgs e)
{
return EasyTask.CompletedTask;
}
#endregion 事件
private string GetDefaultNewId()
{
return Interlocked.Increment(ref this.m_nextId).ToString();
}
/// <summary>
/// 获取下一个新Id
/// </summary>
/// <returns></returns>
protected string GetNextNewId()
{
try
{
return this.m_getDefaultNewId.Invoke();
}
catch (Exception ex)
{
this.Logger.Exception(ex);
}
return this.GetDefaultNewId();
}
/// <inheritdoc/>
/// <exception cref="ArgumentNullException"></exception>
/// <exception cref="ObjectDisposedException"></exception>
public override void AddListen(TcpListenOption option)
{
if (option is null)
{
throw new ArgumentNullException(nameof(option));
}
this.ThrowIfDisposed();
if (option.IpHost is null)
{
throw new ArgumentNullException(nameof(option.IpHost));
}
var socket = new Socket(option.IpHost.EndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
if (option.ReuseAddress)
{
socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true);
}
var e = new SocketAsyncEventArgs();
var networkMonitor = new TcpNetworkMonitor(option, socket, e);
this.PreviewBind(networkMonitor);
socket.Bind(option.IpHost.EndPoint);
socket.Listen(option.Backlog);
e.UserToken = networkMonitor;
e.Completed += this.Args_Completed;
if (!networkMonitor.Socket.AcceptAsync(e))
{
this.OnAccepted(e);
}
this.m_monitors.Add(networkMonitor);
}
/// <inheritdoc/>
public override void Clear()
{
foreach (var item in this.GetIds())
{
if (this.TryGetSocketClient(item, out var client))
{
client.SafeDispose();
}
}
}
/// <summary>
/// 获取当前在线的所有客户端
/// </summary>
/// <returns></returns>
public IEnumerable<TClient> GetClients()
{
return this.m_socketClients.GetClients()
.Select(a => (TClient)a);
}
/// <inheritdoc/>
/// <returns></returns>
/// <exception cref="ArgumentNullException"></exception>
public override bool RemoveListen(TcpNetworkMonitor monitor)
{
if (monitor is null)
{
throw new ArgumentNullException(nameof(monitor));
}
if (this.m_monitors.Remove(monitor))
{
monitor.SocketAsyncEvent.SafeDispose();
monitor.Socket.SafeDispose();
return true;
}
return false;
}
/// <summary>
/// <inheritdoc/>
/// </summary>
/// <param name="oldId"></param>
/// <param name="newId"></param>
/// <exception cref="ClientNotFindException"></exception>
/// <exception cref="Exception"></exception>
public override void ResetId(string oldId, string newId)
{
if (string.IsNullOrEmpty(oldId))
{
throw new ArgumentException($"“{nameof(oldId)}”不能为 null 或空。", nameof(oldId));
}
if (string.IsNullOrEmpty(newId))
{
throw new ArgumentException($"“{nameof(newId)}”不能为 null 或空。", nameof(newId));
}
if (oldId == newId)
{
return;
}
if (this.m_socketClients.TryGetSocketClient(oldId, out TClient socketClient))
{
socketClient.ResetId(newId);
}
else
{
throw new ClientNotFindException(TouchSocketResource.ClientNotFind.GetDescription(oldId));
}
}
/// <summary>
///<inheritdoc/>
/// </summary>
/// <param name="id"></param>
/// <returns></returns>
public override bool SocketClientExist(string id)
{
return this.SocketClients.SocketClientExist(id);
}
/// <inheritdoc/>
public override void Start()
{
if (this.Config is null)
{
throw new ArgumentNullException(nameof(this.Config), "Config为null，请先执行Setup");
}
try
{
var optionList = new List<TcpListenOption>();
if (this.Config.GetValue(TouchSocketConfigExtension.ListenOptionsProperty) is Action<List<TcpListenOption>> action)
{
action.Invoke(optionList);
}
var iPHosts = this.Config.GetValue(TouchSocketConfigExtension.ListenIPHostsProperty);
if (iPHosts != null)
{
foreach (var item in iPHosts)
{
var option = new TcpListenOption
{
IpHost = item,
ServiceSslOption = this.Config.GetValue(TouchSocketConfigExtension.SslOptionProperty) as ServiceSslOption,
ReuseAddress = this.Config.GetValue(TouchSocketConfigExtension.ReuseAddressProperty),
NoDelay = this.Config.GetValue(TouchSocketConfigExtension.NoDelayProperty),
Adapter = this.Config.GetValue(TouchSocketConfigExtension.TcpDataHandlingAdapterProperty),
};
option.Backlog = this.Config.GetValue(TouchSocketConfigExtension.BacklogProperty) ?? option.Backlog;
option.SendTimeout = this.Config.GetValue(TouchSocketConfigExtension.SendTimeoutProperty);
optionList.Add(option);
}
}
switch (this.m_serverState)
{
case ServerState.None:
{
this.BeginListen(optionList);
break;
}
case ServerState.Running:
{
return;
}
case ServerState.Stopped:
{
this.BeginListen(optionList);
break;
}
case ServerState.Disposed:
{
throw new ObjectDisposedException(this.GetType().Name);
}
}
this.m_serverState = ServerState.Running;
this.PluginManager.Raise(nameof(IServerStartedPlugin.OnServerStarted), this, new ServiceStateEventArgs(this.m_serverState, default));
return;
}
catch (Exception ex)
{
this.m_serverState = ServerState.Exception;
this.PluginManager.Raise(nameof(IServerStartedPlugin.OnServerStarted), this, new ServiceStateEventArgs(this.m_serverState, ex) { Message = ex.Message });
throw;
}
}
/// <inheritdoc/>
public override async Task StartAsync()
{
if (this.Config is null)
{
throw new ArgumentNullException(nameof(this.Config), "Config为null，请先执行Setup");
}
try
{
var optionList = new List<TcpListenOption>();
if (this.Config.GetValue(TouchSocketConfigExtension.ListenOptionsProperty) is Action<List<TcpListenOption>> action)
{
action.Invoke(optionList);
}
var iPHosts = this.Config.GetValue(TouchSocketConfigExtension.ListenIPHostsProperty);
if (iPHosts != null)
{
foreach (var item in iPHosts)
{
var option = new TcpListenOption
{
IpHost = item,
ServiceSslOption = this.Config.GetValue(TouchSocketConfigExtension.SslOptionProperty) as ServiceSslOption,
ReuseAddress = this.Config.GetValue(TouchSocketConfigExtension.ReuseAddressProperty),
NoDelay = this.Config.GetValue(TouchSocketConfigExtension.NoDelayProperty),
Adapter = this.Config.GetValue(TouchSocketConfigExtension.TcpDataHandlingAdapterProperty),
};
option.Backlog = this.Config.GetValue(TouchSocketConfigExtension.BacklogProperty) ?? option.Backlog;
option.SendTimeout = this.Config.GetValue(TouchSocketConfigExtension.SendTimeoutProperty);
optionList.Add(option);
}
}
switch (this.m_serverState)
{
case ServerState.None:
{
this.BeginListen(optionList);
break;
}
case ServerState.Running:
{
return;
}
case ServerState.Stopped:
{
this.BeginListen(optionList);
break;
}
case ServerState.Disposed:
{
throw new ObjectDisposedException(this.GetType().Name);
}
}
this.m_serverState = ServerState.Running;
await this.PluginManager.RaiseAsync(nameof(IServerStartedPlugin.OnServerStarted), this, new ServiceStateEventArgs(this.m_serverState, default)).ConfigureFalseAwait();
return;
}
catch (Exception ex)
{
this.m_serverState = ServerState.Exception;
await this.PluginManager.RaiseAsync(nameof(IServerStartedPlugin.OnServerStarted), this, new ServiceStateEventArgs(this.m_serverState, ex) { Message = ex.Message }).ConfigureFalseAwait();
throw;
}
}
/// <inheritdoc/>
public override void Stop()
{
foreach (var item in this.m_monitors)
{
item.Socket.SafeDispose();
item.SocketAsyncEvent.SafeDispose();
}
this.m_monitors.Clear();
this.Clear();
this.m_serverState = ServerState.Stopped;
this.PluginManager.Raise(nameof(IServerStopedPlugin.OnServerStoped), this, new ServiceStateEventArgs(this.m_serverState, default));
return;
}
/// <inheritdoc/>
public override async Task StopAsync()
{
foreach (var item in this.m_monitors)
{
item.Socket.SafeDispose();
item.SocketAsyncEvent.SafeDispose();
}
this.m_monitors.Clear();
this.Clear();
this.m_serverState = ServerState.Stopped;
await this.PluginManager.RaiseAsync(nameof(IServerStopedPlugin.OnServerStoped), this, new ServiceStateEventArgs(this.m_serverState, default));
}
/// <summary>
/// 尝试获取TClient
/// </summary>
/// <param name="id">Id</param>
/// <param name="socketClient">TClient</param>
/// <returns></returns>
public bool TryGetSocketClient(string id, out TClient socketClient)
{
return this.m_socketClients.TryGetSocketClient(id, out socketClient);
}
/// <summary>
/// <inheritdoc/>
/// </summary>
/// <param name="disposing"></param>
protected override void Dispose(bool disposing)
{
if (this.DisposedValue)
{
return;
}
if (disposing)
{
foreach (var item in this.m_monitors)
{
item.Socket.SafeDispose();
item.SocketAsyncEvent.SafeDispose();
}
this.m_monitors.Clear();
this.Clear();
this.m_serverState = ServerState.Disposed;
this.PluginManager.Raise(nameof(IServerStopedPlugin.OnServerStoped), this, new ServiceStateEventArgs(this.m_serverState, default));
}
base.Dispose(disposing);
}
/// <summary>
/// 初始化客户端实例。
/// </summary>
/// <returns></returns>
protected virtual TClient GetClientInstence(Socket socket, TcpNetworkMonitor monitor)
{
return new TClient();
}
/// <inheritdoc/>
protected override void LoadConfig(TouchSocketConfig config)
{
if (config.GetValue(TouchSocketConfigExtension.GetDefaultNewIdProperty) is Func<string> fun)
{
this.m_getDefaultNewId = fun;
}
this.m_maxCount = config.GetValue(TouchSocketConfigExtension.MaxCountProperty);
}
/// <summary>
/// 在验证Ssl发送错误时。
/// </summary>
/// <param name="ex"></param>
protected virtual void OnAuthenticatingError(Exception ex)
{
}
/// <inheritdoc/>
protected virtual void PreviewBind(TcpNetworkMonitor monitor)
{
}
private void Args_Completed(object? sender, SocketAsyncEventArgs e)
{
this.OnAccepted(e);
}
private void BeginListen(List<TcpListenOption> optionList)
{
foreach (var item in optionList)
{
this.AddListen(item);
}
}
private void OnAccepted(SocketAsyncEventArgs e)
{
if (this.DisposedValue)
{
return;
}
if (e.LastOperation == SocketAsyncOperation.Accept && e.SocketError == SocketError.Success && e.AcceptSocket != null)
{
var socket = e.AcceptSocket;
if (this.SocketClients.Count < this.m_maxCount)
{
this.OnClientSocketInit(Tuple.Create(socket, (TcpNetworkMonitor)e.UserToken)).GetFalseAwaitResult();
//Task.Factory.StartNew(this.OnClientSocketInit, Tuple.Create(socket, (TcpNetworkMonitor)e.UserToken));
}
else
{
socket.SafeDispose();
this.Logger.Warning(this, "连接客户端数量已达到设定最大值");
}
}
if (this.m_serverState == ServerState.Running)
{
e.AcceptSocket = null;
try
{
if (!((TcpNetworkMonitor)e.UserToken).Socket.AcceptAsync(e))
{
this.OnAccepted(e);
}
}
catch (ObjectDisposedException)
{
}
catch (Exception ex)
{
this.Logger.Exception(ex);
e.SafeDispose();
return;
}
}
}
private SingleStreamDataHandlingAdapter GetAdapter(TcpNetworkMonitor monitor)
{
try
{
return monitor.Option.Adapter.Invoke();
}
catch (Exception ex)
{
this.Logger.Exception(ex);
}
return new NormalDataHandlingAdapter();
}
private async Task OnClientSocketInit(object obj)
{
var tuple = (Tuple<Socket, TcpNetworkMonitor>)obj;
var socket = tuple.Item1;
var monitor = tuple.Item2;
try
{
if (monitor.Option.NoDelay != null)
{
socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.NoDelay, monitor.Option.NoDelay);
}
socket.SendTimeout = monitor.Option.SendTimeout;
var client = this.GetClientInstence(socket, monitor);
client.InternalSetService(this);
client.InternalSetContainer(this.Resolver);
client.InternalSetListenOption(monitor.Option);
client.InternalSetSocket(socket);
client.InternalSetPluginManager(this.PluginManager);
if (client.CanSetDataHandlingAdapter)
{
client.SetDataHandlingAdapter(this.GetAdapter(monitor));
}
await client.InternalInitialized().ConfigureFalseAwait();
var args = new ConnectingEventArgs(socket)
{
Id = this.GetNextNewId()
};
await client.InternalConnecting(args).ConfigureFalseAwait();//Connecting
if (args.IsPermitOperation)
{
client.InternalSetId(args.Id);
if (!socket.Connected)
{
socket.SafeDispose();
return;
}
if (this.m_socketClients.TryAdd(client))
{
_ = client.InternalConnected(new ConnectedEventArgs());
if (!socket.Connected)
{
return;
}
if (monitor.Option.UseSsl)
{
try
{
await client.AuthenticateAsync(monitor.Option.ServiceSslOption).ConfigureFalseAwait();
_ = client.BeginReceiveSsl();
}
catch (Exception ex)
{
this.OnAuthenticatingError(ex);
throw;
}
}
else
{
client.BeginReceive();
}
}
else
{
throw new Exception($"Id={client.Id}重复");
}
}
else
{
socket.SafeDispose();
}
}
catch (Exception ex)
{
socket.SafeDispose();
this.Logger.Log(LogLevel.Error, this, "接收新连接错误", ex);
}
}
}
}

866
src/foundation/ThingsGateway.Foundation/src/Channel/TcpService/TgSocketClientBase.cs
View File

@@ -1,866 +0,0 @@
//------------------------------------------------------------------------------
// 此代码版权声明为全文件覆盖，如有原作者特别声明，会在下方手动补充
// 此代码版权（除特别声明外的代码）归作者本人Diego所有
// 源代码使用协议遵循本仓库的开源协议及附加协议
// Gitee源代码仓库：https://gitee.com/diego2098/ThingsGateway
// Github源代码仓库：https://github.com/kimdiego2098/ThingsGateway
// 使用文档：https://diego2098.gitee.io/thingsgateway-docs/
// QQ群：605534569
//------------------------------------------------------------------------------
using System.Diagnostics;
using System.Net.Sockets;
using TouchSocket.Resources;
namespace ThingsGateway.Foundation
{
/// <summary>
/// SocketClient
/// </summary>
[DebuggerDisplay("Id={Id},IPAdress={IP}:{Port}")]
public class TgSocketClientBase : ConfigObject, ISocketClient
{
/// <summary>
/// 构造函数
/// </summary>
public TgSocketClientBase()
{
this.Protocol = Protocol.Tcp;
}
#region 变量
private DelaySender m_delaySender;
private TcpCore m_tcpCore;
#endregion 变量
#region 属性
/// <inheritdoc/>
public bool CanSend => this.Online;
/// <inheritdoc/>
public virtual bool CanSetDataHandlingAdapter => true;
/// <inheritdoc/>
public IResolver Resolver { get; private set; }
/// <inheritdoc/>
public SingleStreamDataHandlingAdapter DataHandlingAdapter { get; private set; }
/// <inheritdoc/>
public string Id { get; private set; }
/// <inheritdoc/>
public string IP { get; private set; }
/// <inheritdoc/>
public bool IsClient => false;
/// <inheritdoc/>
public DateTime LastReceivedTime => this.GetTcpCore().ReceiveCounter.LastIncrement;
/// <inheritdoc/>
public DateTime LastSendTime => this.GetTcpCore().SendCounter.LastIncrement;
/// <inheritdoc/>
public TcpListenOption ListenOption { get; private set; }
/// <inheritdoc/>
public Socket MainSocket { get; private set; }
/// <inheritdoc/>
public bool Online { get; private set; }
/// <inheritdoc/>
public IPluginManager PluginManager { get; private set; }
/// <inheritdoc/>
public int Port { get; private set; }
/// <inheritdoc/>
public Protocol Protocol { get; set; }
/// <inheritdoc/>
[Obsolete("该配置已被弃用，正式版发布时会直接删除", true)]
public ReceiveType ReceiveType { get; private set; }
/// <inheritdoc/>
public TcpServiceBase Service { get; private set; }
/// <inheritdoc/>
public TgTcpService TgService => (TgTcpService)Service;
/// <inheritdoc/>
public string ServiceIP { get; private set; }
/// <inheritdoc/>
public int ServicePort { get; private set; }
/// <inheritdoc/>
public bool UseSsl { get; private set; }
/// <inheritdoc/>
public override TouchSocketConfig Config => this.TgService?.Config;
#endregion 属性
#region Internal
internal Task AuthenticateAsync(ServiceSslOption sslOption)
{
return this.m_tcpCore.AuthenticateAsync(sslOption);
}
internal void BeginReceive()
{
try
{
this.m_tcpCore.BeginIocpReceive();
}
catch (Exception ex)
{
this.BreakOut(false, ex.Message);
}
}
internal Task BeginReceiveSsl()
{
return this.m_tcpCore.BeginSslReceive();
}
internal Task InternalConnected(ConnectedEventArgs e)
{
this.Online = true;
return this.OnConnected(e);
}
internal Task InternalConnecting(ConnectingEventArgs e)
{
return this.OnConnecting(e);
}
internal Task InternalInitialized()
{
return this.OnInitialized();
}
internal void InternalSetContainer(IResolver containerProvider)
{
this.Resolver = containerProvider;
this.Logger ??= containerProvider.Resolve<ILog>();
}
internal void InternalSetId(string id)
{
this.Id = id;
}
internal void InternalSetListenOption(TcpListenOption option)
{
this.ListenOption = option;
//this.ReceiveType = option.ReceiveType;
}
internal void InternalSetPluginManager(IPluginManager pluginManager)
{
this.PluginManager = pluginManager;
}
internal void InternalSetService(TcpServiceBase serviceBase)
{
this.Service = serviceBase;
}
internal void InternalSetSocket(Socket socket)
{
this.MainSocket = socket ?? throw new ArgumentNullException(nameof(socket));
this.IP = socket.RemoteEndPoint.GetIP();
this.Port = socket.RemoteEndPoint.GetPort();
this.ServiceIP = socket.LocalEndPoint.GetIP();
this.ServicePort = socket.LocalEndPoint.GetPort();
if (this.Config.GetValue(TouchSocketConfigExtension.DelaySenderProperty) is DelaySenderOption senderOption)
{
this.m_delaySender = new DelaySender(senderOption, this.GetTcpCore().Send);
}
var tcpCore = this.TgService.RentTcpCore();
tcpCore.Reset(socket);
tcpCore.OnReceived = this.HandleReceived;
tcpCore.OnBreakOut = this.TcpCoreBreakOut;
if (this.Config.GetValue(TouchSocketConfigExtension.MinBufferSizeProperty) is int minValue)
{
tcpCore.MinBufferSize = minValue;
}
if (this.Config.GetValue(TouchSocketConfigExtension.MaxBufferSizeProperty) is int maxValue)
{
tcpCore.MaxBufferSize = maxValue;
}
this.m_tcpCore = tcpCore;
}
/// <summary>
/// 中断连接
/// </summary>
/// <param name="manual"></param>
/// <param name="msg"></param>
protected void BreakOut(bool manual, string msg)
{
if (this.GetSocketCliectCollection().TryRemove(this.Id, out _))
{
if (this.Online)
{
this.Online = false;
this.MainSocket.SafeDispose();
this.m_delaySender.SafeDispose();
this.DataHandlingAdapter.SafeDispose();
Task.Factory.StartNew(this.PrivateOnDisconnected, new DisconnectEventArgs(manual, msg));
}
}
}
private void HandleReceived(TcpCore core, ByteBlock byteBlock)
{
try
{
if (this.DisposedValue)
{
return;
}
if (this.ReceivingData(byteBlock).GetFalseAwaitResult())
{
return;
}
if (this.DataHandlingAdapter == null)
{
this.Logger.Error(this, TouchSocketResource.NullDataAdapter.GetDescription());
return;
}
this.DataHandlingAdapter.ReceivedInput(byteBlock);
}
catch (Exception ex)
{
this.Logger.Log(LogLevel.Error, this, "在处理数据时发生错误", ex);
}
}
private void TcpCoreBreakOut(TcpCore core, bool manual, string msg)
{
this.BreakOut(manual, msg);
}
#endregion Internal
#region 事件&委托
/// <inheritdoc/>
public DisconnectEventHandler<ITcpClientBase> Disconnected { get; set; }
/// <inheritdoc/>
public DisconnectEventHandler<ITcpClientBase> Disconnecting { get; set; }
/// <summary>
/// 当客户端完整建立Tcp连接。
/// </summary>
/// <param name="e"></param>
protected virtual async Task OnConnected(ConnectedEventArgs e)
{
if (await this.PluginManager.RaiseAsync(nameof(ITcpConnectedPlugin.OnTcpConnected), this, e))
{
return;
}
await this.TgService.OnInternalConnected(this, e);
}
/// <summary>
/// 客户端正在连接。
/// </summary>
protected virtual async Task OnConnecting(ConnectingEventArgs e)
{
if (await this.PluginManager.RaiseAsync(nameof(ITcpConnectingPlugin.OnTcpConnecting), this, e))
{
return;
}
await this.TgService.OnInternalConnecting(this, e);
}
/// <summary>
/// 客户端已断开连接。
/// </summary>
/// <param name="e"></param>
protected virtual async Task OnDisconnected(DisconnectEventArgs e)
{
if (this.Disconnected != null)
{
await this.Disconnected.Invoke(this, e);
if (e.Handled)
{
return;
}
}
if (await this.PluginManager.RaiseAsync(nameof(ITcpDisconnectedPlugin.OnTcpDisconnected), this, e))
{
return;
}
await this.TgService.OnInternalDisconnected(this, e);
}
/// <summary>
/// 即将断开连接(仅主动断开时有效)。
/// </summary>
/// <param name="e"></param>
protected virtual async Task OnDisconnecting(DisconnectEventArgs e)
{
try
{
if (this.Disconnecting != null)
{
await this.Disconnecting.Invoke(this, e);
if (e.Handled)
{
return;
}
}
if (await this.PluginManager.RaiseAsync(nameof(ITcpDisconnectingPlugin.OnTcpDisconnecting), this, e))
{
return;
}
await this.TgService.OnInternalDisconnecting(this, e);
}
catch (Exception ex)
{
this.Logger.Log(LogLevel.Error, this, $"在事件{nameof(this.Disconnecting)}中发生错误。", ex);
}
}
/// <summary>
/// 当初始化完成时，执行在<see cref="OnConnecting(ConnectingEventArgs)"/>之前。
/// </summary>
protected virtual Task OnInitialized()
{
return EasyTask.CompletedTask;
}
private async Task PrivateOnDisconnected(object? obj)
{
this.m_receiver?.TryInputReceive(default, default);
var e = (DisconnectEventArgs)obj;
try
{
await this.OnDisconnected(e);
}
catch (Exception)
{
}
finally
{
var tcp = this.m_tcpCore;
this.m_tcpCore = null;
this.TgService.ReturnTcpCore(tcp);
base.Dispose(true);
}
}
private Task PrivateOnDisconnecting(object? obj)
{
return this.OnDisconnecting((DisconnectEventArgs)obj);
}
#endregion 事件&委托
/// <inheritdoc/>
public virtual void Close(string msg)
{
lock (this.GetTcpCore())
{
if (this.Online)
{
Task.Factory.StartNew(this.PrivateOnDisconnecting, new DisconnectEventArgs(true, msg));
this.MainSocket.TryClose();
this.BreakOut(true, msg);
}
}
}
/// <summary>
/// <inheritdoc/>
/// </summary>
/// <returns></returns>
/// <exception cref="NotImplementedException"></exception>
[Obsolete("该方法已被弃用，正式版发布时会直接删除", true)]
public Stream GetStream()
{
throw new NotImplementedException();
}
/// <summary>
/// <inheritdoc/>
/// </summary>
/// <param name="newId"></param>
/// <exception cref="ArgumentException"></exception>
/// <exception cref="ClientNotFindException"></exception>
/// <exception cref="Exception"></exception>
public virtual void ResetId(string newId)
{
this.DirectResetId(newId);
}
/// <inheritdoc/>
public virtual void SetDataHandlingAdapter(SingleStreamDataHandlingAdapter adapter)
{
this.ThrowIfDisposed();
if (!this.CanSetDataHandlingAdapter)
{
throw new Exception($"不允许自由调用{nameof(SetDataHandlingAdapter)}进行赋值。");
}
this.SetAdapter(adapter);
}
/// <summary>
/// 直接重置内部Id。
/// </summary>
/// <param name="newId"></param>
protected void DirectResetId(string newId)
{
this.ThrowIfDisposed();
if (string.IsNullOrEmpty(newId))
{
throw new ArgumentException($"“{nameof(newId)}”不能为 null 或空。", nameof(newId));
}
if (this.Id == newId)
{
return;
}
var oldId = this.Id;
if (this.GetSocketCliectCollection().TryRemove(this.Id, out TgSocketClient socketClient))
{
socketClient.Id = newId;
if (this.GetSocketCliectCollection().TryAdd(socketClient))
{
this.IdChanged(oldId, newId).ConfigureAwait(false).GetAwaiter().GetResult();
return;
}
else
{
socketClient.Id = oldId;
if (this.GetSocketCliectCollection().TryAdd(socketClient))
{
throw new Exception("Id重复");
}
else
{
socketClient.Close("修改新Id时操作失败，且回退旧Id时也失败。");
}
}
}
else
{
throw new ClientNotFindException(TouchSocketResource.ClientNotFind.GetDescription(oldId));
}
}
/// <inheritdoc/>
protected override void Dispose(bool disposing)
{
lock (this.GetTcpCore())
{
if (this.Online)
{
Task.Factory.StartNew(this.PrivateOnDisconnecting, new DisconnectEventArgs(true, $"{nameof(Dispose)}主动断开"));
this.BreakOut(true, $"{nameof(Dispose)}主动断开");
}
}
}
/// <summary>
/// 处理已接收到的数据。
/// <para>根据不同的数据处理适配器，会传递不同的数据</para>
/// </summary>
/// <param name="byteBlock">以二进制流形式传递</param>
/// <param name="requestInfo">以解析的数据对象传递</param>
/// <returns>如果返回<see langword="true"/>则表示数据已被处理，且不会再向下传递。</returns>
[Obsolete("此方法已被弃用，请使用ReceivedData替代。本方法将在正式版发布时删除", true)]
protected virtual bool HandleReceivedData(ByteBlock byteBlock, IRequestInfo requestInfo)
{
return false;
}
/// <summary>
/// 当Id更新的时候触发
/// </summary>
/// <param name="oldId"></param>
/// <param name="newId"></param>
/// <returns></returns>
protected Task IdChanged(string oldId, string newId)
{
return this.PluginManager.RaiseAsync(nameof(IIdChangedPlugin.OnIdChanged), this, new IdChangedEventArgs(oldId, newId));
}
/// <summary>
/// 当收到适配器处理的数据时。
/// </summary>
/// <returns>如果返回<see langword="true"/>则表示数据已被处理，且不会再向下传递。</returns>
protected virtual async Task ReceivedData(ReceivedDataEventArgs e)
{
if (e.Handled)
{
return;
}
await this.PluginManager.RaiseAsync(nameof(ITcpReceivedPlugin.OnTcpReceived), this, e);
if (e.Handled)
{
return;
}
await this.TgService.OnInternalReceivedData(this, e);
}
/// <summary>
/// 当收到原始数据
/// </summary>
/// <param name="byteBlock"></param>
/// <returns>如果返回<see langword="true"/>则表示数据已被处理，且不会再向下传递。</returns>
protected virtual Task<bool> ReceivingData(ByteBlock byteBlock)
{
if (this.PluginManager.GetPluginCount(nameof(ITcpReceivingPlugin.OnTcpReceiving)) > 0)
{
return this.PluginManager.RaiseAsync(nameof(ITcpReceivingPlugin.OnTcpReceiving), this, new ByteBlockEventArgs(byteBlock));
}
return Task.FromResult(false);
}
/// <summary>
/// 当即将发送时，如果覆盖父类方法，则不会触发插件。
/// </summary>
/// <param name="buffer">数据缓存区</param>
/// <param name="offset">偏移</param>
/// <param name="length">长度</param>
/// <returns>返回值表示是否允许发送</returns>
protected virtual async Task<bool> SendingData(byte[] buffer, int offset, int length)
{
if (this.PluginManager.GetPluginCount(nameof(ITcpSendingPlugin.OnTcpSending)) > 0)
{
var args = new SendingEventArgs(buffer, offset, length);
await this.PluginManager.RaiseAsync(nameof(ITcpSendingPlugin.OnTcpSending), this, args).ConfigureAwait(false);
return args.IsPermitOperation;
}
return true;
}
/// <summary>
/// 设置适配器，该方法不会检验<see cref="CanSetDataHandlingAdapter"/>的值。
/// </summary>
/// <param name="adapter"></param>
protected void SetAdapter(SingleStreamDataHandlingAdapter adapter)
{
if (adapter is null)
{
throw new ArgumentNullException(nameof(adapter));
}
if (this.Config != null)
{
adapter.Config(this.Config);
}
adapter.Logger = this.Logger;
adapter.OnLoaded(this);
adapter.ReceivedCallBack = this.PrivateHandleReceivedData;
adapter.SendCallBack = this.DefaultSend;
adapter.SendAsyncCallBack = this.DefaultSendAsync;
this.DataHandlingAdapter = adapter;
}
private SocketClientCollection GetSocketCliectCollection()
{
return this.TgService.SocketClients as SocketClientCollection;
}
private TcpCore GetTcpCore()
{
this.ThrowIfDisposed();
return this.m_tcpCore ?? throw new ObjectDisposedException(this.GetType().Name);
}
private void PrivateHandleReceivedData(ByteBlock byteBlock, IRequestInfo requestInfo)
{
if (this.m_receiver != null)
{
if (this.m_receiver.TryInputReceive(byteBlock, requestInfo))
{
return;
}
}
this.ReceivedData(new ReceivedDataEventArgs(byteBlock, requestInfo)).GetFalseAwaitResult();
}
#region 发送
/// <inheritdoc/>
public void DefaultSend(byte[] buffer, int offset, int length)
{
if (this.SendingData(buffer, offset, length).GetFalseAwaitResult())
{
if (this.m_delaySender != null)
{
this.m_delaySender.Send(new QueueDataBytes(buffer, offset, length));
return;
}
this.GetTcpCore().Send(buffer, offset, length);
}
}
/// <inheritdoc/>
public async Task DefaultSendAsync(byte[] buffer, int offset, int length)
{
if (await this.SendingData(buffer, offset, length))
{
await this.GetTcpCore().SendAsync(buffer, offset, length);
}
}
#region 同步发送
/// <summary>
/// <inheritdoc/>
/// </summary>
/// <param name="requestInfo"></param>
/// <exception cref="NotConnectedException"></exception>
/// <exception cref="OverlengthException"></exception>
/// <exception cref="Exception"></exception>
public virtual void Send(IRequestInfo requestInfo)
{
this.ThrowIfDisposed();
if (this.DataHandlingAdapter == null)
{
throw new ArgumentNullException(nameof(this.DataHandlingAdapter), TouchSocketResource.NullDataAdapter.GetDescription());
}
if (!this.DataHandlingAdapter.CanSendRequestInfo)
{
throw new NotSupportedException($"当前适配器不支持对象发送。");
}
this.DataHandlingAdapter.SendInput(requestInfo);
}
/// <summary>
/// 发送字节流
/// </summary>
/// <param name="buffer"></param>
/// <param name="offset"></param>
/// <param name="length"></param>
/// <exception cref="NotConnectedException"></exception>
/// <exception cref="OverlengthException"></exception>
/// <exception cref="Exception"></exception>
public virtual void Send(byte[] buffer, int offset, int length)
{
this.ThrowIfDisposed();
if (this.DataHandlingAdapter == null)
{
throw new ArgumentNullException(nameof(this.DataHandlingAdapter), TouchSocketResource.NullDataAdapter.GetDescription());
}
this.DataHandlingAdapter.SendInput(buffer, offset, length);
}
/// <summary>
/// <inheritdoc/>
/// </summary>
/// <param name="transferBytes"></param>
public virtual void Send(IList<ArraySegment<byte>> transferBytes)
{
this.ThrowIfDisposed();
if (this.DataHandlingAdapter == null)
{
throw new ArgumentNullException(nameof(this.DataHandlingAdapter), TouchSocketResource.NullDataAdapter.GetDescription());
}
if (this.DataHandlingAdapter.CanSplicingSend)
{
this.DataHandlingAdapter.SendInput(transferBytes);
}
else
{
var length = 0;
foreach (var item in transferBytes)
{
length += item.Count;
}
using (var byteBlock = new ByteBlock(length))
{
foreach (var item in transferBytes)
{
byteBlock.Write(item.Array, item.Offset, item.Count);
}
this.DataHandlingAdapter.SendInput(byteBlock.Buffer, 0, byteBlock.Len);
}
}
}
#endregion 同步发送
#region 异步发送
/// <summary>
/// <inheritdoc/>
/// </summary>
/// <param name="buffer"></param>
/// <param name="offset"></param>
/// <param name="length"></param>
/// <exception cref="NotConnectedException"></exception>
/// <exception cref="OverlengthException"></exception>
/// <exception cref="Exception"></exception>
public virtual Task SendAsync(byte[] buffer, int offset, int length)
{
this.ThrowIfDisposed();
if (this.DataHandlingAdapter == null)
{
throw new ArgumentNullException(nameof(this.DataHandlingAdapter), TouchSocketResource.NullDataAdapter.GetDescription());
}
return this.DataHandlingAdapter.SendInputAsync(buffer, offset, length);
}
/// <summary>
/// <inheritdoc/>
/// </summary>
/// <param name="requestInfo"></param>
/// <exception cref="NotConnectedException"></exception>
/// <exception cref="OverlengthException"></exception>
/// <exception cref="Exception"></exception>
public virtual Task SendAsync(IRequestInfo requestInfo)
{
this.ThrowIfDisposed();
if (this.DataHandlingAdapter == null)
{
throw new ArgumentNullException(nameof(this.DataHandlingAdapter), TouchSocketResource.NullDataAdapter.GetDescription());
}
if (!this.DataHandlingAdapter.CanSendRequestInfo)
{
throw new NotSupportedException($"当前适配器不支持对象发送。");
}
return this.DataHandlingAdapter.SendInputAsync(requestInfo);
}
/// <summary>
/// <inheritdoc/>
/// </summary>
/// <param name="transferBytes"></param>
/// <returns></returns>
/// <exception cref="ArgumentNullException"></exception>
public virtual Task SendAsync(IList<ArraySegment<byte>> transferBytes)
{
this.ThrowIfDisposed();
if (this.DataHandlingAdapter == null)
{
throw new ArgumentNullException(nameof(this.DataHandlingAdapter), TouchSocketResource.NullDataAdapter.GetDescription());
}
if (this.DataHandlingAdapter.CanSplicingSend)
{
return this.DataHandlingAdapter.SendInputAsync(transferBytes);
}
else
{
var length = 0;
foreach (var item in transferBytes)
{
length += item.Count;
}
using (var byteBlock = new ByteBlock(length))
{
foreach (var item in transferBytes)
{
byteBlock.Write(item.Array, item.Offset, item.Count);
}
return this.DataHandlingAdapter.SendInputAsync(byteBlock.Buffer, 0, byteBlock.Len);
}
}
}
#endregion 异步发送
#region Id发送
/// <summary>
/// 发送字节流
/// </summary>
/// <param name="id">用于检索TcpSocketClient</param>
/// <param name="buffer"></param>
/// <param name="offset"></param>
/// <param name="length"></param>
/// <exception cref="KeyNotFoundException"></exception>
/// <exception cref="NotConnectedException"></exception>
/// <exception cref="OverlengthException"></exception>
/// <exception cref="Exception"></exception>
public void Send(string id, byte[] buffer, int offset, int length)
{
this.TgService.Send(id, buffer, offset, length);
}
/// <summary>
/// <inheritdoc/>
/// </summary>
/// <param name="id"></param>
/// <param name="requestInfo"></param>
public void Send(string id, IRequestInfo requestInfo)
{
this.TgService.Send(id, requestInfo);
}
/// <summary>
/// 发送字节流
/// </summary>
/// <param name="id">用于检索TcpSocketClient</param>
/// <param name="buffer"></param>
/// <param name="offset"></param>
/// <param name="length"></param>
/// <exception cref="KeyNotFoundException"></exception>
/// <exception cref="NotConnectedException"></exception>
/// <exception cref="OverlengthException"></exception>
/// <exception cref="Exception"></exception>
public Task SendAsync(string id, byte[] buffer, int offset, int length)
{
return this.TgService.SendAsync(id, buffer, offset, length);
}
/// <summary>
/// <inheritdoc/>
/// </summary>
/// <param name="id"></param>
/// <param name="requestInfo"></param>
public Task SendAsync(string id, IRequestInfo requestInfo)
{
return this.TgService.SendAsync(id, requestInfo);
}
#endregion Id发送
#endregion 发送
#region TgReceiver
private TgReceiver m_receiver;
/// <inheritdoc/>
public void ClearReceiver()
{
this.m_receiver = null;
}
/// <inheritdoc/>
public IReceiver CreateReceiver()
{
return this.m_receiver ??= new TgReceiver(this);
}
#endregion TgReceiver
}
}

4
src/foundation/ThingsGateway.Foundation/src/Channel/TgSocketClient.cs
View File

@@ -16,7 +16,7 @@ namespace ThingsGateway.Foundation
/// TgSocketClient
/// </summary>
[DebuggerDisplay("Id={Id},IPAdress={IP}:{Port}")]
public class TgSocketClient : TgSocketClientBase, IClientChannel
public class TgSocketClient : SocketClient, IClientChannel
{
/// <summary>
/// TgSocketClient
@@ -126,7 +126,7 @@ namespace ThingsGateway.Foundation
return base.OnDisconnected(e);
}
#region 无
#region
/// <inheritdoc/>
public void Setup(TouchSocketConfig config)

17
src/foundation/ThingsGateway.Foundation/src/Channel/TgTcpService.cs
View File

@@ -13,7 +13,7 @@ using System.Net.Sockets;
namespace ThingsGateway.Foundation
{
/// <inheritdoc/>
public class TgTcpServiceBase<TClient> : TcpServiceBase<TClient>, ITcpService<TClient> where TClient : TgSocketClient, new()
public class TgTcpServiceBase<TClient> : TcpService<TClient>, ITcpService<TClient> where TClient : TgSocketClient, new()
{
/// <inheritdoc/>
~TgTcpServiceBase()
@@ -165,16 +165,13 @@ namespace ThingsGateway.Foundation
{
return;
}
if (disposing)
{
if (Monitors.Count() > 0)
{
if (Monitors.Count() == 0)
Logger.Info($"{this}{FoundationConst.ServiceStoped}");
}
}
var count = Monitors.Count();
base.Dispose(disposing);
if (count > 0)
{
if (Monitors.Count() == 0)
Logger.Info($"{this}{FoundationConst.ServiceStoped}");
}
}
}

66
src/foundation/ThingsGateway.Foundation/src/Channel/UdpSession/TgUdpSessionBase.cs
View File

@@ -41,7 +41,6 @@ namespace ThingsGateway.Foundation
{
Dispose(true);
}
/// <summary>
/// <inheritdoc/>
/// </summary>
@@ -314,10 +313,9 @@ namespace ThingsGateway.Foundation
this.m_socketAsyncs.Clear();
this.PluginManager?.Raise(nameof(IServerStartedPlugin.OnServerStarted), this, new ServiceStateEventArgs(this.m_serverState, default));
this.PluginManager.SafeDispose();
this.PluginManager?.SafeDispose();
}
}
base.Dispose(disposing);
}
@@ -377,6 +375,7 @@ namespace ThingsGateway.Foundation
throw new ArgumentNullException(nameof(adapter));
}
if (adapter.Owner != null && adapter.Owner != this)
{
throw new Exception(FoundationConst.CannotUseAdapterAgain);
@@ -390,6 +389,7 @@ namespace ThingsGateway.Foundation
adapter.OnLoaded(this);
adapter.ReceivedCallBack = this.PrivateHandleReceivedData;
adapter.SendCallBack = this.DefaultSend;
adapter.SendCallBackAsync = this.DefaultSendAsync;
this.DataHandlingAdapter = adapter;
}
@@ -408,6 +408,7 @@ namespace ThingsGateway.Foundation
#region Windows下UDP连接被重置错误10054
if (NewLife.Runtime.Windows)
{
const int SIP_UDP_CONNRESET = -1744830452;
@@ -489,7 +490,7 @@ namespace ThingsGateway.Foundation
{
try
{
this.LastReceivedTime = DateTimeUtil.Now;
this.LastReceivedTime = DateTime.Now;
if (this.DisposedValue)
{
return;
@@ -513,7 +514,7 @@ namespace ThingsGateway.Foundation
private void IO_Completed(object? sender, SocketAsyncEventArgs e)
{
this.ProcessReceive((Socket)sender!, e);
this.ProcessReceive((Socket)sender, e);
}
private void PrivateHandleReceivedData(EndPoint remoteEndPoint, ByteBlock byteBlock, IRequestInfo requestInfo)
@@ -546,10 +547,6 @@ namespace ThingsGateway.Foundation
{
throw new ArgumentNullException(nameof(this.DataHandlingAdapter), TouchSocketResource.NullDataAdapter.GetDescription());
}
if (!this.DataHandlingAdapter.CanSendRequestInfo)
{
throw new NotSupportedException(FoundationConst.CannotSendIRequestInfo);
}
}
#endregion Throw
@@ -678,9 +675,9 @@ namespace ThingsGateway.Foundation
if (e.SocketError == SocketError.Success)
{
var byteBlock = (ByteBlock)e.UserToken;
byteBlock!.SetLength(e.BytesTransferred);
byteBlock.SetLength(e.BytesTransferred);
this.HandleBuffer(e.RemoteEndPoint!, byteBlock);
this.HandleBuffer(e.RemoteEndPoint, byteBlock);
var newByteBlock = new ByteBlock(1024 * 64);
e.UserToken = newByteBlock;
@@ -708,14 +705,10 @@ namespace ThingsGateway.Foundation
#region DefaultSend
/// <summary>
/// <inheritdoc/>
/// </summary>
/// <param name="buffer"></param>
/// <param name="offset"></param>
/// <param name="length"></param>
public void DefaultSend(byte[] buffer, int offset, int length)
{
this.ThorwIfRemoteIPHostNull();
this.DefaultSend(this.RemoteIPHost.EndPoint, buffer, offset, length);
}
@@ -728,14 +721,11 @@ namespace ThingsGateway.Foundation
/// <param name="length"></param>
public void DefaultSend(EndPoint endPoint, byte[] buffer, int offset, int length)
{
this.ThrowIfDisposed();
if (this.HandleSendingData(endPoint, buffer, offset, length))
{
if (this.CanSend)
{
this.Monitor.Socket.SendTo(buffer, offset, length, SocketFlags.None, endPoint);
}
this.LastSendTime = DateTimeUtil.Now;
this.Monitor.Socket.SendTo(buffer, offset, length, SocketFlags.None, endPoint);
this.LastSendTime = DateTime.Now;
}
}
@@ -751,10 +741,7 @@ namespace ThingsGateway.Foundation
/// <param name="length"></param>
public Task DefaultSendAsync(byte[] buffer, int offset, int length)
{
if (this.RemoteIPHost == null)
{
throw new ArgumentNullException(nameof(RemoteIPHost));
}
this.ThorwIfRemoteIPHostNull();
return this.DefaultSendAsync(this.RemoteIPHost.EndPoint, buffer, offset, length);
}
@@ -802,10 +789,7 @@ namespace ThingsGateway.Foundation
/// <param name="transferBytes"></param>
public void Send(IList<ArraySegment<byte>> transferBytes)
{
if (this.RemoteIPHost == null)
{
throw new ArgumentNullException(nameof(RemoteIPHost));
}
this.ThorwIfRemoteIPHostNull();
this.Send(this.RemoteIPHost.EndPoint, transferBytes);
}
@@ -816,10 +800,7 @@ namespace ThingsGateway.Foundation
/// <param name="transferBytes"></param>
public void Send(EndPoint endPoint, IList<ArraySegment<byte>> transferBytes)
{
if (this.DataHandlingAdapter == null)
{
throw new ArgumentNullException(nameof(this.DataHandlingAdapter), TouchSocketResource.NullDataAdapter.GetDescription());
}
ThorwIfDataHandlingAdapterNull();
if (!this.DataHandlingAdapter.CanSplicingSend)
{
@@ -834,10 +815,7 @@ namespace ThingsGateway.Foundation
/// <param name="transferBytes"></param>
public Task SendAsync(IList<ArraySegment<byte>> transferBytes)
{
if (this.RemoteIPHost == null)
{
throw new ArgumentNullException(nameof(RemoteIPHost));
}
this.ThorwIfRemoteIPHostNull();
return this.SendAsync(this.RemoteIPHost.EndPoint, transferBytes);
}
@@ -848,15 +826,7 @@ namespace ThingsGateway.Foundation
/// <param name="transferBytes"></param>
public Task SendAsync(EndPoint endPoint, IList<ArraySegment<byte>> transferBytes)
{
if (this.DataHandlingAdapter == null)
{
throw new ArgumentNullException(nameof(this.DataHandlingAdapter), TouchSocketResource.NullDataAdapter.GetDescription());
}
if (!this.DataHandlingAdapter.CanSplicingSend)
{
throw new NotSupportedException(FoundationConst.CannotSplicingSend);
}
ThorwIfDataHandlingAdapterNull();
return this.DataHandlingAdapter.SendInputAsync(endPoint, transferBytes);
}
@@ -869,7 +839,7 @@ namespace ThingsGateway.Foundation
/// <inheritdoc/>
public IReceiver CreateReceiver()
{
return this.m_receiver ??= new TgReceiver(this);
return this.m_receiver ??= new(this);
}
/// <inheritdoc/>

2
src/foundation/ThingsGateway.Foundation/src/ThingsGateway.Foundation.csproj
View File

@@ -13,7 +13,7 @@
<ItemGroup>
<PackageReference Include="NewLife.Core" Version="10.8.2024.303" />
<PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
<PackageReference Include="TouchSocket" Version="2.0.0-rc.2" />
<PackageReference Include="TouchSocket" Version="2.0.0" />
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)'!='net45'">