MasstransferExporter/MasstransferInfrastructure/Process/Client/ProcessHelper.cs

using MasstransferCommon.Utils;
using MasstransferCommunicate.Process.Model;
using Serilog;

namespace MasstransferCommunicate.Process.Client;

/// <summary>
/// 进程通讯工具类
/// </summary>
public class ProcessHelper
{
    private static readonly Dictionary<string, List<Delegate>> Subscribers = new();

    private static PipeClient? _pipe;


    /// <summary>
    /// 初始化通讯管道
    /// </summary>
    public static void Init()
    {
        if (_pipe != null) return;
        _pipe = new PipeClient();
        _pipe.MessageReceived += HandleMessageReceived;
        _pipe.Start();
    }

    /// <summary>
    /// 订阅消息
    /// </summary>
    /// <param name="topic"></param>
    /// <param name="delegate"></param>
    public static void Subscribe(string topic, Delegate @delegate)
    {
        if (!Subscribers.ContainsKey(topic))
        {
            Subscribers.Add(topic, []);
        }

        Subscribers[topic].Add(@delegate);
    }

    /// <summary>
    /// 发送消息
    /// </summary>
    /// <param name="topic"></param>
    /// <param name="message"></param>
    /// <typeparam name="T"></typeparam>
    public static void Send<T>(string topic, T message)
    {
        var payload = new Message<T>(topic, message);
        _pipe?.SendMessage(JsonUtil.ToJson(payload));
    }

    /// <summary>
    /// 处理接收到的消息
    /// </summary>
    /// <returns></returns>
    private static void HandleMessageReceived(string? message)
    {
        if (message == null) return;

        var dictionary = JsonUtil.ToDictionary(message);
        if (dictionary == null) return;

        var topic = dictionary["Topic"] as string;

        if (!Subscribers.TryGetValue(topic, out var subscribers)) return;

        foreach (var subscriber in subscribers)
        {
            try
            {
                var methodInfo = subscriber.Method;
                var parameters = methodInfo.GetParameters();
                if (parameters.Length != 2) continue;

                var dataType = parameters[1].ParameterType;

                var type = typeof(Message<>).MakeGenericType(dataType);

                var payload = JsonUtil.FromJson(type, message);

                if (payload == null) continue;

                var property = type.GetProperty("Data");

                var data = property?.GetValue(payload);

                // 通知订阅者
                subscriber.DynamicInvoke(topic, data);
            }
            catch (Exception exception)
            {
                Log.Error(exception, "订阅主题 {Topic} 时发生错误", topic);
            }
        }
    }
}