simapi-net/Synapse/EventServer.cs

using System;
using System.Linq;
using System.Text.Json;
using System.Text.RegularExpressions;
using System.Threading.Tasks;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using MQTTnet;
using SimApi.Helpers;

namespace SimApi;

public partial class Synapse
{
    
    private string EventServerTopicPrefix => $"{Options.SysName}/event/";

    private void RunEventServer()
    {
        Client!.ApplicationMessageReceivedAsync += e =>
        {
            if (!e.ApplicationMessage.Topic.StartsWith(EventServerTopicPrefix)) return Task.CompletedTask;
            var reqBody = e.ApplicationMessage.ConvertPayloadToString();
            var eventName = e.ApplicationMessage.Topic.Replace(EventServerTopicPrefix, string.Empty);
            logger.LogDebug("Synapse Event Receive: {EventName}\n{Body}", eventName, reqBody);
            var methods = EventRegistry
                .Where(x => Regex.IsMatch(eventName,
                    "^" + Regex.Escape(x.Key!).Replace("\\+", "[^/]+").Replace("\\#", ".*") + "$"))
                .ToArray();
            foreach (var method in methods)
            {
                var callClass = sp.CreateScope().ServiceProvider.GetRequiredService(method.Class!);
                var mt = callClass.GetType().GetMethod(method.Method!);
                try
                {
                    if (mt!.GetParameters().Length == 2)
                    {
                        var pt = mt.GetParameters()[1].ParameterType;
                        mt.Invoke(callClass, pt == typeof(string)
                            ? [eventName, reqBody]
                            : [eventName, JsonSerializer.Deserialize(reqBody, pt, SimApiUtil.JsonOption)]);
                    }
                    else
                    {
                        mt.Invoke(callClass, [eventName]);
                    }
                }
                catch (Exception ex)
                {
                    logger.LogError("Synapse Event Processor Error: {Err}\n{Stack}", ex.Message,ex.StackTrace);
                }
            }
            return Task.CompletedTask;
        };
        SubEventServerTopic();
    }

    private void SubEventServerTopic()
    {
        foreach (var ev in EventRegistry)
        {
            var topic = $"{EventServerTopicPrefix}{ev.Key}";
            if (Options.EventLoadBalancing)
            {
                topic = "$queue/" + topic;
            }
            var evSubOpts = MqttFactory.CreateSubscribeOptionsBuilder()
                .WithTopicFilter(o => o.WithTopic(topic)).Build();
            Client!.SubscribeAsync(evSubOpts).Wait();
            logger.LogDebug("Synapse Event Register Event Success: {EventName}\nFull Topic: {Topic}", ev.Key, topic);
        }
    }
}