XRLib/Assets/Scripts/UVC/Data/Mqtt/MqttWorker.cs

using Cysharp.Threading.Tasks;
using NUnit.Framework;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using UnityEngine;
using UVC.Data.Core;
using UVC.network;

namespace UVC.Data.Mqtt
{
    /// <summary>
    /// 백그라운드 스레드에서 MQTT 통신, 데이터 처리 및 전파를 모두 담당하는 독립적인 워커 클래스입니다.
    /// 생성 시 bufferDurationSec 값에 따라 두 가지 모드로 동작합니다.
    /// 
    /// 1. 버퍼링 모드 (bufferDurationSec > 0):
    ///    - 수신된 메시지를 일정 주기(propagationIntervalSec)마다 모아서 리스너에게 전달합니다.
    ///    - 오래된 데이터는 자동으로 폐기됩니다.
    /// 
    /// 2. 직접 전파 모드 (bufferDurationSec <= 0):
    ///    - 메시지를 수신하는 즉시 리스너에게 전달합니다. 버퍼링 과정이 생략됩니다.
    /// </summary>
    /// <remarks>
    /// 이 클래스에서 등록된 리스너(콜백)는 Unity의 메인 스레드가 아닌,
    /// 별도의 백그라운드 스레드에서 호출됩니다. 따라서 리스너 내부에서 Unity API를 직접 호출하면 안 됩니다.
    /// </remarks>
    public class MqttWorker
    {
        /// <summary>
        /// 버퍼링 모드에서 사용되는 스레드 안전 큐입니다.
        /// 네트워크 스레드에서 수신된 메시지를 워커의 처리 루프로 안전하게 전달하는 역할을 합니다.
        /// </summary>
        private ConcurrentQueue<MqttDataPacket> IncomingMessageQueue { get; set; }

        private Thread workerThread;
        private volatile bool isRunning = false;

        /// <summary>
        /// 스레드를 안전하게 종료시키기 위한 CancellationTokenSource 입니다.
        /// Stop() 메서드가 호출되면 취소 신호를 보냅니다.
        /// </summary>
        private readonly CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();

        private readonly Dictionary<string, List<MqttDataPacket>> topicBuffers = new Dictionary<string, List<MqttDataPacket>>();
        private readonly Dictionary<string, Action<string, List<MqttDataPacket>>> listeners = new Dictionary<string, Action<string, List<MqttDataPacket>>>();
        private readonly object bufferAndListenerLock = new object();

        /// <summary>
        /// 데이터를 버퍼에 보관할 최대 시간(초)입니다. 이 시간이 지난 데이터는 폐기됩니다.
        /// 이 값이 0보다 크면 버퍼링 모드로 동작합니다.
        /// </summary>
        private float bufferDurationSec = 0f;

        /// <summary>
        /// [버퍼링 모드 전용] 버퍼링된 데이터를 리스너에게 전파하는 주기(초)입니다.
        /// </summary>
        private float propagationIntervalSec = 1f;

        private string domain = "localhost";
        private int port = 1883;

        /// <summary>
        /// MqttWorker의 생성자입니다.
        /// </summary>
        public MqttWorker() { }

        /// <summary>
        /// 현재 구성의 도메인과 포트를 설정합니다.
        /// </summary>
        /// <remarks>이 메서드는 현재 구성에 사용되는 도메인 및 포트 값을 업데이트합니다.
        /// 도메인이 유효한 DNS 이름이고 포트가 유효한 네트워크 포트 범위 내에 있는지 확인합니다.
        ///</remarks>
        /// <param name="domain">접속할 MQTT 브로커의 도메인 주소입니다.</param>
        /// <param name="port">접속할 MQTT 브로커의 포트 번호입니다.</param>
        public void SetDomainPort(string domain, int port)
        {
            this.domain = domain;
            this.port = port;
        }

        /// <summary>
        /// 버퍼 전파의 지속 시간과 간격을 설정합니다.
        /// </summary>
        /// <remarks>이 메서드는 버퍼 작업에 대한 타이밍 매개변수를 구성합니다. 예기치 않은 동작을 방지하려면 두 매개변수 모두
        /// 양수여야 합니다.</remarks>
        /// <param name="durationSec">데이터 버퍼링 시간(초). 0 이하로 설정 시 버퍼링 없이 직접 전파 모드로 동작합니다.</param>
        /// <<param name="intervalSec">버퍼링 모드에서 데이터를 전파할 주기(초)입니다.</param>
        public void SetDurationInterval(float durationSec, float intervalSec)
        {
            this.bufferDurationSec = durationSec;
            this.propagationIntervalSec = intervalSec;
        }


        /// <summary>
        /// 백그라운드 워커 스레드를 시작합니다.
        /// </summary>
        public void Start()
        {
            if (isRunning) return;
            isRunning = true;
            workerThread = new Thread(Run);
            workerThread.IsBackground = true; // 메인 앱 종료 시 스레드 자동 종료
            workerThread.Start();
        }

        /// <summary>
        /// 백그라운드 워커 스레드를 중지합니다.
        /// </summary>
        public void Stop()
        {
            if (!isRunning) return;
            isRunning = false;
            cancellationTokenSource.Cancel();
            workerThread?.Join(); // 스레드가 완전히 종료될 때까지 대기
        }

        /// <summary>
        /// 워커를 중지하고 관련된 리소스를 모두 해제합니다.
        /// </summary>
        public void Dispose()
        {
            Stop();
            cancellationTokenSource.Dispose();
        }

        /// <summary>
        /// 특정 토픽에 대한 데이터 수신을 시작합니다.
        /// 중요: 여기서 등록된 리스너(콜백)는 백그라운드 스레드에서 호출됩니다.
        /// Unity API(GameObject, Transform 등)에 직접 접근하면 안 됩니다.
        /// </summary>
        public void AddListener(string topic, Action<string, List<MqttDataPacket>> listener)
        {
            lock (bufferAndListenerLock)
            {
                if (!listeners.ContainsKey(topic))
                {
                    listeners[topic] = listener;
                    topicBuffers[topic] = new List<MqttDataPacket>();
                }
                else
                {
                    listeners[topic] += listener;
                }
            }
        }


        /// <summary>
        /// 지정된 토픽에서 리스너를 제거합니다.
        /// </summary>
        /// <remarks>지정된 리스너가 토픽의 마지막 리스너인 경우, 토픽은 내부 컬렉션에서 제거됩니다.
        ///</remarks>
        /// <param name="topic">리스너를 제거할 토픽입니다. null이거나 비어 있을 수 없습니다.</param>
        /// <param name="listener">제거할 리스너로, <see cref="Action{string, List{MqttDataPacket}}"/>으로 표현됩니다. null일 수 없습니다.</param>
        public void RemoveListener(string topic, Action<string, List<MqttDataPacket>> listener)
        {
            lock (bufferAndListenerLock)
            {
                if (listeners.ContainsKey(topic))
                {
                    listeners[topic] -= listener;
                    if (listeners[topic] == null || listeners[topic].GetInvocationList().Length == 0)
                    {
                        listeners.Remove(topic);
                        topicBuffers.Remove(topic);
                    }
                }
            }
        }

        /// <summary>
        /// 백그라운드 스레드에서 실행되는 메인 루프입니다.
        /// </summary>
        private void Run()
        {
            bool isMainThread = PlayerLoopHelper.IsMainThread;
            Debug.Log($"[Worker] 백그라운드 스레드 시작. isMainThread:{isMainThread}");
            // MQTT 서비스 객체를 생성하고 설정합니다.
            MQTTService mqtt = new MQTTService(domain, port);

            // AddListener를 통해 미리 등록된 토픽들을 구독합니다.
            lock (bufferAndListenerLock)
            {
                foreach (var topic in listeners.Keys)
                {
                    mqtt.AddTopicHandler(topic, OnMqttMessageReceived);
                }
            }
            mqtt.Connect();

            // 설정된 bufferDurationSec 값에 따라 동작 모드를 결정합니다.
            if (bufferDurationSec > 0)
            {
                // 버퍼링 모드: 주기적으로 메시지를 처리하고 전파합니다.
                RunBufferingLoop();
            }
            else
            {
                // 직접 전파 모드: 스레드를 대기 상태로 두고, 메시지 수신 시 직접 리스너를 호출합니다.
                RunDirectPropagationLoop();
            }

            // MQTT 연결을 해제하고 관련 리소스를 정리합니다.
            mqtt.Dispose();
            // ------------------------------------
            Debug.Log("[Worker] 백그라운드 스레드 종료.");
        }

        /// <summary>
        /// 버퍼링 모드의 메인 루프입니다.
        /// </summary>
        private void RunBufferingLoop()
        {
            var cancellationToken = cancellationTokenSource.Token;
            var propagationIntervalMs = (int)(propagationIntervalSec * 1000);

            while (!cancellationToken.IsCancellationRequested)
            {
                ProcessIncomingMessages();
                PropagateBufferedData();

                try
                {
                    // 다음 전파 주기까지 대기합니다.
                    if (cancellationToken.WaitHandle.WaitOne(propagationIntervalMs))
                    {
                        break; // Stop() 호출 시 루프 종료
                    }
                }
                catch (ObjectDisposedException) { break; }
            }
        }

        /// <summary>
        /// 직접 전파 모드의 메인 루프입니다. Stop()이 호출될 때까지 스레드를 대기시킵니다.
        /// </summary>
        private void RunDirectPropagationLoop()
        {
            try
            {
                cancellationTokenSource.Token.WaitHandle.WaitOne();
            }
            catch (ObjectDisposedException)
            {
                // 정상적인 종료 과정이므로 예외를 무시합니다.
            }
        }

        /// <summary>
        /// MQTT 메시지가 수신될 때마다 MQTTService에 의해 호출되는 콜백 메서드입니다.
        /// 이 메서드는 워커 스레드에서 실행됩니다.
        /// </summary>
        /// <param name="topic">메시지가 수신된 토픽</param>
        /// <param name="message">수신된 메시지 내용</param>
        private void OnMqttMessageReceived(string topic, string message)
        {
            bool isMainThread = PlayerLoopHelper.IsMainThread;
            //Debug.Log($"OnMqttMessageReceived isMainThread: {isMainThread}, topic: {topic}");
            if (bufferDurationSec > 0)
            {
                // 버퍼링 모드: 메시지를 큐에 넣어 워커 스레드로 전달합니다.
                try
                {
                    var dataArray = DataArrayPool.Get().FromJsonString(message);
                    foreach (var dataObject in dataArray)
                    {
                        IncomingMessageQueue.Enqueue(MqttDataPacketPool.Get().FromDataObject(topic, dataObject));
                    }
                    dataArray.ReturnToPool();
                }
                catch (Exception ex)
                {
                    Debug.LogError($"[Worker] 주제 {topic}에 대한 수신 JSON 배열을 구문 분석하지 못했습니다. 오류: {ex.Message}");
                }
            }
            else
            {
                // 직접 전파 모드: 즉시 리스너를 호출합니다.
                lock (bufferAndListenerLock)
                {
                    if (listeners[topic] != null)
                    {
                        try
                        {
                            var list = new List<MqttDataPacket>();
                            var dataArray = DataArrayPool.Get().FromJsonString(message);
                            foreach (var dataObject in dataArray)
                            {
                                list.Add(new MqttDataPacket(topic, dataObject));
                            }
                            // 직접 전파 모드에서도 타임스탬프 순으로 정렬
                            list.Sort((p1, p2) => p1.Timestamp.CompareTo(p2.Timestamp));
                            listeners[topic].Invoke(topic, list);
                            dataArray.ReturnToPool();
                        }
                        catch (Exception ex)
                        {
                            Debug.LogError($"[Worker] 주제 {topic}에 대한 수신 JSON 배열을 구문 분석하지 못했습니다. 오류: {ex.Message}");
                        }
                    }
                }
            }
        }

        /// <summary>
        /// [버퍼링 모드 전용] 큐의 메시지를 토픽별 버퍼로 옮깁니다.
        /// </summary>
        private void ProcessIncomingMessages()
        {
            // 큐에서 메시지를 가져와 처리
            while (IncomingMessageQueue.TryDequeue(out MqttDataPacket packet))
            {
                lock (bufferAndListenerLock)
                {
                    if (topicBuffers.ContainsKey(packet.Topic))
                    {
                        topicBuffers[packet.Topic].Add(packet);
                    }
                }
            }
        }

        /// <summary>
        /// [버퍼링 모드 전용] 버퍼링된 데이터를 리스너에게 전파합니다.
        /// </summary>
        private void PropagateBufferedData()
        {
            DateTime cutoffTime = DateTime.UtcNow.AddSeconds(-bufferDurationSec);

            lock (bufferAndListenerLock)
            {
                foreach (var topic in listeners.Keys.ToList())
                {
                    if (!topicBuffers.ContainsKey(topic)) continue;

                    var buffer = topicBuffers[topic];

                    // 오래된 데이터 풀에 반환 및 버퍼에서 제거
                    var outdatedPackets = buffer.Where(p => p.Timestamp < cutoffTime).ToList();
                    if (outdatedPackets.Count > 0)
                    {
                        foreach (var packet in outdatedPackets)
                        {
                            packet.ReturnToPool();
                        }
                        buffer.RemoveAll(p => p.Timestamp < cutoffTime);
                    }

                    var newPackets = buffer.Where(p => !p.IsPropagated).ToList();

                    

                    if (newPackets.Count > 0)
                    {
                        try
                        {
                            //타임스탬프 순으로 정렬
                            newPackets.Sort((p1, p2) => p1.Timestamp.CompareTo(p2.Timestamp));
                            // 중요: 이 콜백은 Worker 스레드에서 직접 호출됩니다.
                            
                            listeners[topic]?.Invoke(topic, newPackets);
                        }
                        catch (Exception ex)
                        {
                            Debug.LogError($"[Worker] 리스너 실행 중 오류 발생 (Topic: {topic}): {ex.Message}");
                        }

                        foreach (var packet in newPackets)
                        {
                            packet.IsPropagated = true;
                        }
                    }
                }
            }
        }
    }
}