//------------------------------------------------------------------------------
// 此代码版权(除特别声明或在XREF结尾的命名空间的代码)归作者本人若汝棋茗所有
// 源代码使用协议遵循本仓库的开源协议及附加协议,若本仓库没有设置,则按MIT开源协议授权
// CSDN博客:https://blog.csdn.net/qq_40374647
// 哔哩哔哩视频:https://space.bilibili.com/94253567
// Gitee源代码仓库:https://gitee.com/RRQM_Home
// Github源代码仓库:https://github.com/RRQM
// API首页:https://www.yuque.com/rrqm/touchsocket/index
// 交流QQ群:234762506
// 感谢您的下载和使用
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
using System;
using System.Collections.Generic;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using TouchSocket.Core;
using TouchSocket.Core.ByteManager;
using TouchSocket.Core.Config;
using TouchSocket.Core.Dependency;
using TouchSocket.Core.Log;
using TouchSocket.Http;
using TouchSocket.Sockets;
namespace TouchSocket.Rpc.TouchRpc
{
///
/// HttpRpcClient
///
public class HttpTouchRpcClient : HttpClientBase, IHttpTouchRpcClient
{
private readonly ActionMap m_actionMap;
private int m_failCount = 0;
private readonly RpcActor m_rpcActor;
private RpcStore m_rpcStore;
private Timer m_timer;
///
/// 创建一个HttpTouchRpcClient实例。
///
public HttpTouchRpcClient()
{
this.m_actionMap = new ActionMap();
this.m_rpcActor = new RpcActor(false)
{
OutputSend = this.RpcActorSend,
OnHandshaked = this.OnRpcActorHandshaked,
OnReceived = this.OnRpcActorReceived,
OnClose = this.OnRpcServiceClose,
OnStreamTransfering = this.OnRpcActorStreamTransfering,
OnStreamTransfered = this.OnRpcActorStreamTransfered,
OnFileTransfering = this.OnRpcActorFileTransfering,
OnFileTransfered = this.OnRpcActorFileTransfered,
GetInvokeMethod = this.GetInvokeMethod,
Caller = this
};
}
///
/// 服务器映射
///
public ActionMap ActionMap { get => this.m_actionMap; }
///
///
///
public string ID => this.m_rpcActor.ID;
///
///
///
public bool IsHandshaked => this.m_rpcActor != null && this.m_rpcActor.IsHandshaked;
///
///
///
public ResponseType ResponseType { get => this.m_rpcActor.ResponseType; set => this.m_rpcActor.ResponseType = value; }
///
///
///
public string RootPath { get => this.m_rpcActor.RootPath; set => this.m_rpcActor.RootPath = value; }
///
///
///
public RpcActor RpcActor => this.m_rpcActor;
///
///
///
public RpcStore RpcStore => this.m_rpcStore;
///
///
///
public SerializationSelector SerializationSelector => this.m_rpcActor.SerializationSelector;
///
///
///
public Func TryCanInvoke { get; set; }
///
///
///
///
///
public bool ChannelExisted(int id)
{
return this.m_rpcActor.ChannelExisted(id);
}
///
/// ,以“rrqm”连接
///
///
///
public override ITcpClient Connect(int timeout = 5000)
{
if (this.IsHandshaked)
{
return this;
}
if (!this.Online)
{
base.Connect(timeout);
}
HttpRequest httpRequest = new HttpRequest();
httpRequest.InitHeaders()
.AsMethod(TouchRpcUtility.TouchRpc);
var response = this.RequestContent(httpRequest);
if (response.StatusCode == "200")
{
this.SwitchProtocolToTouchRpc();
this.m_rpcActor.Handshake(this.Config.GetValue(TouchRpcConfigExtensions.VerifyTokenProperty), default,
timeout, this.Config.GetValue(TouchRpcConfigExtensions.MetadataProperty));
return this;
}
else
{
throw new Exception(response.StatusMessage);
}
}
///
///
///
///
public Channel CreateChannel()
{
return this.m_rpcActor.CreateChannel();
}
///
///
///
///
///
public Channel CreateChannel(int id)
{
return this.m_rpcActor.CreateChannel(id);
}
///
///
///
///
///
public Channel CreateChannel(string clientID)
{
return this.m_rpcActor.CreateChannel(clientID);
}
///
///
///
///
///
///
public Channel CreateChannel(string clientID, int id)
{
return this.m_rpcActor.CreateChannel(clientID, id);
}
///
///
///
///
///
///
public void Invoke(string method, IInvokeOption invokeOption, params object[] parameters)
{
this.m_rpcActor.Invoke(method, invokeOption, parameters);
}
///
///
///
///
///
///
///
///
public T Invoke(string method, IInvokeOption invokeOption, params object[] parameters)
{
return this.m_rpcActor.Invoke(method, invokeOption, parameters);
}
///
///
///
///
///
///
///
///
///
public T Invoke(string method, IInvokeOption invokeOption, ref object[] parameters, Type[] types)
{
return this.m_rpcActor.Invoke(method, invokeOption, ref parameters, types);
}
///
///
///
///
///
///
///
public void Invoke(string method, IInvokeOption invokeOption, ref object[] parameters, Type[] types)
{
this.m_rpcActor.Invoke(method, invokeOption, ref parameters, types);
}
///
///
///
///
///
///
///
public void Invoke(string id, string method, IInvokeOption invokeOption, params object[] parameters)
{
this.m_rpcActor.Invoke(id, method, invokeOption, parameters);
}
///
///
///
///
///
///
///
///
///
public T Invoke(string id, string method, IInvokeOption invokeOption, params object[] parameters)
{
return this.m_rpcActor.Invoke(id, method, invokeOption, parameters);
}
///
///
///
///
///
///
///
///
public void Invoke(string targetID, string method, IInvokeOption invokeOption, ref object[] parameters, Type[] types)
{
this.m_rpcActor.Invoke(targetID, method, invokeOption, ref parameters, types);
}
///
///
///
///
///
///
///
///
///
///
public T Invoke(string targetID, string method, IInvokeOption invokeOption, ref object[] parameters, Type[] types)
{
return this.m_rpcActor.Invoke(targetID, method, invokeOption, ref parameters, types);
}
///
///
///
///
///
///
///
public Task InvokeAsync(string method, IInvokeOption invokeOption, params object[] parameters)
{
return this.m_rpcActor.InvokeAsync(method, invokeOption, parameters);
}
///
///
///
///
///
///
///
///
public Task InvokeAsync(string method, IInvokeOption invokeOption, params object[] parameters)
{
return this.m_rpcActor.InvokeAsync(method, invokeOption, parameters);
}
///
///
///
///
///
///
///
///
public Task InvokeAsync(string id, string method, IInvokeOption invokeOption, params object[] parameters)
{
return this.m_rpcActor.InvokeAsync(id, method, invokeOption, parameters);
}
///
///
///
///
///
///
///
///
///
public Task InvokeAsync(string id, string method, IInvokeOption invokeOption, params object[] parameters)
{
return this.m_rpcActor.InvokeAsync(id, method, invokeOption, parameters);
}
///
///
///
///
///
public bool Ping(int timeout = 5000)
{
return this.m_rpcActor.Ping(timeout);
}
///
///
///
///
///
///
///
public Result PullFile(FileRequest fileRequest, FileOperator fileOperator, Metadata metadata = null)
{
return this.m_rpcActor.PullFile(fileRequest, fileOperator, metadata);
}
///
///
///
///
///
///
///
///
public Result PullFile(string clientID, FileRequest fileRequest, FileOperator fileOperator, Metadata metadata = null)
{
return this.m_rpcActor.PullFile(clientID, fileRequest, fileOperator, metadata);
}
///
///
///
///
///
///
///
public Task PullFileAsync(FileRequest fileRequest, FileOperator fileOperator, Metadata metadata = null)
{
return this.m_rpcActor.PullFileAsync(fileRequest, fileOperator, metadata);
}
///
///
///
///
///
///
///
///
public Task PullFileAsync(string clientID, FileRequest fileRequest, FileOperator fileOperator, Metadata metadata = null)
{
return this.m_rpcActor.PullFileAsync(clientID, fileRequest, fileOperator, metadata);
}
///
///
///
///
///
///
///
public Result PushFile(FileRequest fileRequest, FileOperator fileOperator, Metadata metadata = null)
{
return this.m_rpcActor.PushFile(fileRequest, fileOperator, metadata);
}
///
///
///
///
///
///
///
///
public Result PushFile(string clientID, FileRequest fileRequest, FileOperator fileOperator, Metadata metadata = null)
{
return this.m_rpcActor.PushFile(clientID, fileRequest, fileOperator, metadata);
}
///
///
///
///
///
///
///
public Task PushFileAsync(FileRequest fileRequest, FileOperator fileOperator, Metadata metadata = null)
{
return this.m_rpcActor.PushFileAsync(fileRequest, fileOperator, metadata);
}
///
///
///
///
///
///
///
///
public Task PushFileAsync(string clientID, FileRequest fileRequest, FileOperator fileOperator, Metadata metadata = null)
{
return this.m_rpcActor.PushFileAsync(clientID, fileRequest, fileOperator, metadata);
}
///
///
///
///
///
public void ResetID(string id, CancellationToken cancellationToken = default)
{
this.m_rpcActor.ResetID(id, cancellationToken);
}
///
///
///
///
///
///
///
public Result SendStream(Stream stream, StreamOperator streamOperator, Metadata metadata = null)
{
return this.m_rpcActor.SendStream(stream, streamOperator, metadata);
}
///
///
///
///
///
///
///
public Task SendStreamAsync(Stream stream, StreamOperator streamOperator, Metadata metadata = null)
{
return this.m_rpcActor.SendStreamAsync(stream, streamOperator, metadata);
}
///
///
///
///
///
///
public bool TrySubscribeChannel(int id, out Channel channel)
{
return this.m_rpcActor.TrySubscribeChannel(id, out channel);
}
///
///
///
///
protected override void Dispose(bool disposing)
{
this.m_rpcActor.SafeDispose();
base.Dispose(disposing);
}
///
///
///
///
///
protected override void HandleReceivedData(ByteBlock byteBlock, IRequestInfo requestInfo)
{
if (this.Protocol == TouchRpcUtility.TouchRpcProtocol && byteBlock != null)
{
this.m_rpcActor.InputReceivedData(byteBlock);
return;
}
base.HandleReceivedData(byteBlock, requestInfo);
}
///
///
///
///
protected override void LoadConfig(TouchSocketConfig config)
{
base.LoadConfig(config);
this.m_rpcActor.Logger = this.Container.Resolve();
this.RootPath = this.Config.GetValue(TouchRpcConfigExtensions.RootPathProperty);
this.ResponseType = this.Config.GetValue(TouchRpcConfigExtensions.ResponseTypeProperty);
this.m_rpcActor.SerializationSelector = this.Config.GetValue(TouchRpcConfigExtensions.SerializationSelectorProperty);
if (config.GetValue(RpcConfigExtensions.RpcStoreProperty) is RpcStore rpcStore)
{
rpcStore.AddRpcParser(this.GetType().Name, this);
}
else
{
new RpcStore(config.Container).AddRpcParser(this.GetType().Name, this);
}
}
#region 发送
///
///
///
///
///
public void Send(short protocol, byte[] buffer)
{
this.m_rpcActor.Send(protocol, buffer);
}
///
///
///
///
///
///
///
public void Send(short protocol, byte[] buffer, int offset, int length)
{
this.m_rpcActor.Send(protocol, buffer, offset, length);
}
///
///
///
///
///
public void Send(short protocol, ByteBlock dataByteBlock)
{
this.m_rpcActor.Send(protocol, dataByteBlock);
}
///
///
///
///
public void Send(short protocol)
{
this.m_rpcActor.Send(protocol);
}
///
/// 不允许直接发送
///
///
///
///
public override void Send(byte[] buffer, int offset, int length)
{
throw new Exception("不允许直接发送,请指定任意大于0的协议,然后发送。");
}
///
/// 不允许直接发送
///
///
public override void Send(IList> transferBytes)
{
throw new Exception("不允许直接发送,请指定任意大于0的协议,然后发送。");
}
///
///
///
///
///
public void SendAsync(short protocol, byte[] buffer)
{
this.m_rpcActor.SendAsync(protocol, buffer);
}
///
///
///
///
///
///
///
public void SendAsync(short protocol, byte[] buffer, int offset, int length)
{
this.m_rpcActor.SendAsync(protocol, buffer, offset, length);
}
///
///
///
///
///
public void SendAsync(short protocol, ByteBlock dataByteBlock)
{
this.m_rpcActor.SendAsync(protocol, dataByteBlock);
}
///
///
///
///
public void SendAsync(short protocol)
{
this.m_rpcActor.SendAsync(protocol);
}
///
/// 不允许直接发送
///
///
///
///
public override void SendAsync(byte[] buffer, int offset, int length)
{
throw new Exception("不允许直接发送,请指定任意大于0的协议,然后发送。");
}
///
/// 不允许直接发送
///
///
public override void SendAsync(IList> transferBytes)
{
throw new Exception("不允许直接发送,请指定任意大于0的协议,然后发送。");
}
#endregion 发送
///
///
///
///
protected override void OnDisconnected(ClientDisconnectedEventArgs e)
{
this.m_timer.SafeDispose();
this.m_rpcActor?.Close(e.Message);
base.OnDisconnected(e);
}
#region 内部委托绑定
private MethodInstance GetInvokeMethod(string arg)
{
return this.ActionMap.GetMethodInstance(arg);
}
private void OnRpcActorFileTransfered(RpcActor actor, FileTransferStatusEventArgs e)
{
if (this.UsePlugin && this.PluginsManager.Raise(nameof(ITouchRpcPlugin.OnFileTransfered), this, e))
{
return;
}
this.OnFileTransfered(e);
}
private void OnRpcActorFileTransfering(RpcActor actor, FileOperationEventArgs e)
{
if (this.UsePlugin && this.PluginsManager.Raise(nameof(ITouchRpcPlugin.OnFileTransfering), this, e))
{
return;
}
this.OnFileTransfering(e);
}
private void OnRpcActorHandshaked(RpcActor actor, VerifyOptionEventArgs e)
{
if (this.UsePlugin && this.PluginsManager.Raise(nameof(ITouchRpcPlugin.OnHandshaked), this, e))
{
return;
}
this.OnHandshaked(e);
}
private void OnRpcActorReceived(RpcActor actor, short protocol, ByteBlock byteBlock)
{
if (this.UsePlugin && this.PluginsManager.Raise(nameof(ITouchRpcPlugin.OnReceivedProtocolData), this, new ProtocolDataEventArgs(protocol, byteBlock)))
{
return;
}
this.OnReceived(protocol, byteBlock);
}
private void OnRpcActorStreamTransfered(RpcActor actor, StreamStatusEventArgs e)
{
if (this.UsePlugin && this.PluginsManager.Raise(nameof(ITouchRpcPlugin.OnStreamTransfered), this, e))
{
return;
}
this.OnStreamTransfered(e);
}
private void OnRpcActorStreamTransfering(RpcActor actor, StreamOperationEventArgs e)
{
if (this.UsePlugin && this.PluginsManager.Raise(nameof(ITouchRpcPlugin.OnStreamTransfering), this, e))
{
return;
}
this.OnStreamTransfering(e);
}
private void OnRpcServiceClose(RpcActor actor, string arg2)
{
this.Close(arg2);
}
private void RpcActorSend(RpcActor actor, bool isAsync, ArraySegment[] transferBytes)
{
if (isAsync)
{
base.SendAsync(transferBytes);
}
else
{
base.Send(transferBytes);
}
}
#endregion 内部委托绑定
#region 事件触发
///
/// 当文件传输结束之后。并不意味着完成传输,请通过属性值进行判断。
///
///
protected virtual void OnFileTransfered(FileTransferStatusEventArgs e)
{
}
///
/// 在文件传输即将进行时触发。
///
///
protected virtual void OnFileTransfering(FileOperationEventArgs e)
{
}
///
/// 在完成握手连接时
///
///
protected virtual void OnHandshaked(MsgEventArgs e)
{
this.m_timer.SafeDispose();
if (this.Config.GetValue(TouchRpcConfigExtensions.HeartbeatFrequencyProperty) is HeartbeatValue heartbeat)
{
this.m_timer = new Timer((obj) =>
{
if (this.Ping())
{
Interlocked.Exchange(ref this.m_failCount, 0);
}
else
{
if (Interlocked.Increment(ref this.m_failCount) > heartbeat.MaxFailCount)
{
this.Close("自动心跳失败次数达到最大,已清理连接。");
this.m_timer.SafeDispose();
}
}
}, null, heartbeat.Interval, heartbeat.Interval);
}
}
///
/// 收到数据。
///
///
///
protected virtual void OnReceived(short protocol, ByteBlock byteBlock)
{
}
///
/// 流数据处理,用户需要在此事件中对e.Bucket手动释放。
///
///
protected virtual void OnStreamTransfered(StreamStatusEventArgs e)
{
}
///
/// 即将接收流数据,用户需要在此事件中对e.Bucket初始化。
///
///
protected virtual void OnStreamTransfering(StreamOperationEventArgs e)
{
}
#endregion 事件触发
#region RPC解析器
void IRpcParser.OnRegisterServer(MethodInstance[] methodInstances)
{
foreach (var methodInstance in methodInstances)
{
if (methodInstance.GetAttribute() is TouchRpcAttribute attribute)
{
this.ActionMap.Add(attribute.GetInvokenKey(methodInstance), methodInstance);
}
}
}
void IRpcParser.OnUnregisterServer(MethodInstance[] methodInstances)
{
foreach (var methodInstance in methodInstances)
{
if (methodInstance.GetAttribute() is TouchRpcAttribute attribute)
{
this.m_actionMap.Remove(attribute.GetInvokenKey(methodInstance));
}
}
}
void IRpcParser.SetRpcStore(RpcStore rpcStore)
{
this.m_rpcActor.RpcStore = rpcStore;
this.m_rpcStore = rpcStore;
}
#endregion RPC解析器
}
}