dotnet/taobao-sdk-netcore-auto_1719572210878-20240628-source/Tmc/TmcClient.cs

using System;
using System.Collections;
using System.Collections.Generic;
using System.Diagnostics;
using System.Text;
using System.Threading;
using Taobao.Top.Link.Channel;
using Taobao.Top.Link.Endpoints;
using Taobao.Top.Link.Util;
using Top.Api;
using Top.Api.Util;

namespace Top.Tmc
{
    /// <summary>消息服务客户端</summary>
    public class TmcClient
    {
        // sign parameters
        private const string GROUP_NAME = "group_name";
        private const string SDK = "sdk";
        private const string INTRANET_IP = "intranet_ip";

        private TmcClientIdentity _id;
        private string _appSecret;
        private string _uri;
        private Endpoint _endpoint;
        private EndpointProxy _serverProxy;

        private volatile bool running;

        private int _heartbeatInterval = 45000; // 心跳频率（单位：毫秒）
        private int _reconnectIntervalSeconds = 15; // 重连周期（单位：秒）
        private int _pullRequestIntervalSeconds = 30; // 定时获取消息周期（单位：秒）
        private Timer _reconnectTimer;
        private Timer _pullRequestTimer;

        public event EventHandler<MessageArgs> OnMessage;

        /// <summary>获取或设置Log</summary>
        public ITopLogger Log { get; set; }

        private bool enableTraceLog = true;

        public bool EnableTraceLog
        {
            get { return this.enableTraceLog;}
            set { this.enableTraceLog = value; }
        }

        /// <summary>获取或设置定时发送拉取请求的周期（单位：秒）</summary>
        public int PullRequestIntervalSeconds
        {
            get { return this._pullRequestIntervalSeconds; }
            set
            {
                this._pullRequestIntervalSeconds = value;
                if (this._pullRequestTimer != null)
                    this._pullRequestTimer.Change(TimeSpan.FromSeconds(this._pullRequestIntervalSeconds)
                        , TimeSpan.FromSeconds(this._pullRequestIntervalSeconds));
            }
        }

        /// <summary>获取或设置自动重连间隔（单位：秒）</summary>
        public int ReconnectIntervalSeconds
        {
            get { return this._reconnectIntervalSeconds; }
            set
            {
                this._reconnectIntervalSeconds = value;
                if (this._reconnectTimer != null)
                    this._reconnectTimer.Change(TimeSpan.FromSeconds(this._reconnectIntervalSeconds)
                        , TimeSpan.FromSeconds(this._reconnectIntervalSeconds));
            }
        }

        /// <summary>以默认分组，初始化TMC客户端</summary>
        public TmcClient(string appKey, string appSecret) : this(appKey, appSecret, "default") { }

        /// <summary>初始化TMC客户端</summary>
        public TmcClient(string appKey, string appSecret, string groupName)
        {
            this._appSecret = appSecret;
            this._id = new TmcClientIdentity(appKey, groupName);
            this.PrepareEndpoint();
        }

        /// <summary>连接到 TMC Server</summary>
        /// <param name="uri">TMC server address, eg: ws://mc.api.taobao.com/</param>
        public void Connect(string uri)
        {
            this.running = true;
            doConnect(uri);
            this.StartReconnect();
            this.StartPullRequest();
        }

        private void doConnect(string uri)
        {
            var signHeader = new Dictionary<string, string>();
            var connHeader = new Dictionary<string, object>();
            signHeader.Add(Constants.APP_KEY, this._id.AppKey);
            connHeader.Add(Constants.APP_KEY, signHeader[Constants.APP_KEY]);

            signHeader.Add(GROUP_NAME, this._id.GroupName);
            connHeader.Add(GROUP_NAME, signHeader[GROUP_NAME]);

            signHeader.Add(Constants.TIMESTAMP, DateTime.Now.Ticks.ToString());
            connHeader.Add(Constants.TIMESTAMP, signHeader[Constants.TIMESTAMP]);

            connHeader.Add(Constants.SIGN, TopUtils.SignTopRequest(signHeader, this._appSecret, Constants.SIGN_METHOD_MD5));
            //extra fields
            connHeader.Add(SDK, Constants.SDK_VERSION);
            connHeader.Add(INTRANET_IP, TopUtils.GetIntranetIp());
            this._serverProxy = this._endpoint.GetEndpoint(new TmcServerIdentity(), uri, connHeader);
            this._uri = uri;
            this.Log.Info("connected to tmc server: {0}", uri);
        }

        /// <summary>向指定的主题发布一条与用户无关的消息。</summary>
        /// <param name="topic">主题名称</param>
        /// <param name="content">严格根据主题定义的消息内容（JSON/XML）</param>
        public void Send(string topic, string content)
        {
            if (string.IsNullOrEmpty(topic))
                throw new ArgumentNullException("topic");
            if (string.IsNullOrEmpty(content))
                throw new ArgumentNullException("content");

            IDictionary<string, object> msg = new Dictionary<string, object>();
            msg.Add(MessageFields.KIND, MessageKind.Data);
            msg.Add(MessageFields.DATA_TOPIC, topic);
            msg.Add(MessageFields.DATA_CONTENT, content);
            this._serverProxy.SendAndWait(msg, 2000);
        }

        /// <summary>向指定的主题发布一条与用户相关的消息。</summary>
        /// <param name="topic">主题名称</param>
        /// <param name="content">严格根据主题定义的消息内容（JSON/XML）</param>
        /// <param name="session">用户授权码</param>
        public void Send(string topic, string content, string session)
        {
            if (string.IsNullOrEmpty(topic))
                throw new ArgumentNullException("topic");
            if (string.IsNullOrEmpty(content))
                throw new ArgumentNullException("content");
            if (string.IsNullOrEmpty(session))
                throw new ArgumentNullException("session");

            IDictionary<string, object> msg = new Dictionary<string, object>();
            msg.Add(MessageFields.KIND, MessageKind.Data);
            msg.Add(MessageFields.DATA_TOPIC, topic);
            msg.Add(MessageFields.DATA_CONTENT, content);
            msg.Add(MessageFields.DATA_INCOMING_USER_SESSION, session);
            this._serverProxy.SendAndWait(msg, 2000);
        }

        /// <summary>向服务端发送消息拉取请求</summary>
        protected internal void PullRequest()
        {
            IDictionary<string, object> msg = new Dictionary<string, object>();
            msg.Add(MessageFields.KIND, MessageKind.PullRequest);
            this._serverProxy.Send(msg);
        }

        /// <summary>确认消息</summary>
        protected internal void Confirm(long id)
        {
            IDictionary<string, object> msg = new Dictionary<string, object>();
            msg.Add(MessageFields.KIND, MessageKind.Confirm);
            msg.Add(MessageFields.CONFIRM_ID, id);
            this._serverProxy.Send(msg);
        }

        /// <summary>确认消息</summary>
        protected internal void Fail(long id,string errorMsg)
        {
            IDictionary<string, object> msg = new Dictionary<string, object>();
            msg.Add(MessageFields.KIND, MessageKind.Failed);
            msg.Add(MessageFields.CONFIRM_ID, id);
            msg.Add(MessageFields.CONFIRM_MSG, errorMsg);

            this._serverProxy.Send(msg);
        }

        private void PrepareEndpoint()
        {
            this.Log = Top.Api.Log.Instance;
            this._endpoint = new Endpoint(Log, this._id);
            this._endpoint.ChannelSelector = new ClientChannelSharedSelector(Log) { HeartbeatPeriod = this._heartbeatInterval };
            this._endpoint.OnMessage += new EventHandler<EndpointContext>(InternalOnMessage);
            this._endpoint.OnAckMessage += new EventHandler<AckMessageArgs>(InternalOnAckMessage);
        }

        private void InternalOnMessage(object sender, EndpointContext context)
        {
            if (enableTraceLog)
            {
                this.Log.Info("messsage from {0}: {1}", context.MessageFrom, this.Dump(context.Message));
            }
            
            if (this.OnMessage == null)
                return;

            ThreadPool.QueueUserWorkItem(o =>
            {
                if (!this.running)
                {
                    this.Log.Info(string.Format("message dropped as client closed: {0}", this.Dump(context.Message)));
                    return;
                }

                Message msg = this.ParseMessage(context.Message);
                var args = new MessageArgs(msg, m => this.Confirm(m.Id));
                var sw = new Stopwatch();
                try
                {
                    sw.Start();
                    this.OnMessage(this, args);
                    sw.Stop();
                }
                catch (Exception e)
                {
                    args.Fail(e.Message);
                }

                if (args._isFail)
                {
                    this.Log.Info("process message error: {0}", args._reason);

                    try
                    {
                        this.Fail(msg.Id,args._reason.Length > 128 ? args._reason.Substring(0,128) : args._reason);
                    }
                    catch (Exception e)
                    {
                        this.Log.Warn(string.Format("try report fail reason {0} error {1}", this.Dump(context.Message), e.StackTrace));
                    }

                    return;
                }

                // prevent confirm attach
                if (sw.ElapsedMilliseconds <= 1)
                {
                    Thread.Sleep(10);
                }

                if (args._isConfirmed){
                    return;
                }

                try
                {
                    this.Confirm(msg.Id);

                    if (enableTraceLog)
                    {
                        this.Log.Info("confirm message topic: {0}, dataid: {1}", msg.Topic, msg.Dataid);
                    }
                }
                catch (Exception e)
                {
                    this.Log.Warn(string.Format("confirm message {0} error {1}", this.Dump(context.Message), e.StackTrace));
                }
            });
        }

        private void InternalOnAckMessage(object sender, AckMessageArgs e)
        {
            if (this.Log.IsDebugEnabled())
                this.Log.Debug("ack messsage from {0}: {1}", e.MessageFrom, e.Message);
        }

        private void StartReconnect()
        {
            if (this._reconnectTimer != null) return;
            this._reconnectTimer = new Timer(o =>
            {
                try
                {
                    if (!this._serverProxy.hasValidSender())
                    {
                        this.Log.Info("reconning...");
                        Thread.Sleep(1000);
                        // 尝试再次关闭移除tunnel, 未知异常导致断开重连时，不会移除出tunnel
                        this._serverProxy.Close(this._uri, "client closed");
                        Thread.Sleep(1000);
                        this.doConnect(this._uri);
                    }
                }
                catch (Exception e)
                {
                    this.Log.Warn("reconnect error", e);
                }
            }, null
            , TimeSpan.FromSeconds(this._reconnectIntervalSeconds)
            , TimeSpan.FromSeconds(this._reconnectIntervalSeconds));
        }

        private void StartPullRequest()
        {
            if (this._pullRequestTimer != null) return;
            this._pullRequestTimer = new Timer(o =>
            {
                try
                {
                    if (this._serverProxy.hasValidSender())
                    {
                        this.PullRequest();
                    }
                }
                catch (Exception e)
                {
                    this.Log.Warn("pull request error", e);
                }
            }
            , null
            , TimeSpan.FromMilliseconds(500)
            , TimeSpan.FromSeconds(this.PullRequestIntervalSeconds));
        }

        private Message ParseMessage(IDictionary<string, object> raw)
        {
            var msg = new Message();
            msg.Id = this.GetValue<long>(raw, MessageFields.OUTGOING_ID);
            msg.Topic = this.GetValue<string>(raw, MessageFields.DATA_TOPIC);
            msg.PubAppKey = this.GetValue<string>(raw, MessageFields.DATA_OUTGOING_PUBLISHER);
            msg.PubTime = this.GetValue<DateTime>(raw, MessageFields.DATA_PUBLISH_TIME);
            msg.UserId = this.GetValue<long>(raw, MessageFields.DATA_OUTGOING_USER_ID);
            msg.UserNick = this.GetValue<string>(raw, MessageFields.DATA_OUTGOING_USER_NICK);
            msg.OutgoingTime = this.GetValue<DateTime>(raw, MessageFields.DATA_ATTACH_OUTGOING_TIME);
            msg.Dataid = this.GetValue<object>(raw, MessageFields.DATA_DATAID);

            if (!raw.ContainsKey(MessageFields.DATA_CONTENT))
                return msg;
            msg.Content = raw[MessageFields.DATA_CONTENT] is byte[]
                ? Encoding.UTF8.GetString(GZIPHelper.Unzip(raw[MessageFields.DATA_CONTENT] as byte[]))
                : (string)raw[MessageFields.DATA_CONTENT];

            return msg;
        }

        private T GetValue<T>(IDictionary<string, object> raw, string key)
        {
            if (raw.ContainsKey(key))
            {
                object value = raw[key];
                if (value != null)
                {
                    return (T)value;
                }
            }
            return default(T);
        }

        private string Dump(IDictionary<string, object> raw)
        {
            var buf = new StringBuilder();
            foreach (var i in raw)
                buf.AppendFormat("{0}={1}|", i.Key, i.Value);
            return buf.ToString();
        }

        public void Close()
        {
            this.running = false;
            if (this._pullRequestTimer != null)
            {
                this._pullRequestTimer.Dispose();
                this._pullRequestTimer = null;
            }
            if (this._reconnectTimer != null)
            {
                this._reconnectTimer.Dispose();
                this._reconnectTimer = null;
            }
            this._serverProxy.Close(this._uri, "client closed");
            this.Log.Warn("tmc client closed");
        }

        public bool Online
        {
            get
            {
                return this._serverProxy != null && this._serverProxy.hasValidSender();
            }
        }
    }
}