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

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using UnityEngine;
using UVC.network;

namespace UVC.Data
{
    /// <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<List<MqttDataPacket>>> listeners = new Dictionary<string, Action<List<MqttDataPacket>>>();
        private readonly object bufferAndListenerLock = new object();

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

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

        private string domain;
        private int port;

        /// <summary>
        /// MqttWorker의 생성자입니다.
        /// </summary>
        /// <param name="domain">접속할 MQTT 브로커의 도메인 주소입니다.</param>
        /// <param name="port">접속할 MQTT 브로커의 포트 번호입니다.</param>
        /// <param name="bufferDurationSec">데이터 버퍼링 시간(초). 0 이하로 설정 시 버퍼링 없이 직접 전파 모드로 동작합니다.</param>
        /// <param name="propagationIntervalSec">버퍼링 모드에서 데이터를 전파할 주기(초)입니다.</param>
        public MqttWorker(string domain = "localhost", int port = 1883, float bufferDurationSec = 0f, float propagationIntervalSec = 1f)
        {
            this.domain = domain;
            this.port = port;
            this.bufferDurationSec = bufferDurationSec;
            this.propagationIntervalSec = propagationIntervalSec;
            this.IncomingMessageQueue = new ConcurrentQueue<MqttDataPacket>();
        }

        /// <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<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{List{MqttDataPacket}}"/>으로 표현됩니다. null일 수 없습니다.</param>
        public void RemoveListener(string topic, Action<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()
        {
            Debug.Log("[Worker] 백그라운드 스레드 시작.");
            // 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)
        {
            if (bufferDurationSec > 0)
            {
                // 버퍼링 모드: 메시지를 큐에 넣어 워커 스레드로 전달합니다.
                IncomingMessageQueue.Enqueue(new MqttDataPacket(topic, message));
            }
            else
            {
                // 직접 전파 모드: 즉시 리스너를 호출합니다.
                lock (bufferAndListenerLock)
                {
                    listeners[topic]?.Invoke(new List<MqttDataPacket>() { new MqttDataPacket(topic, 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];
                    // 오래된 데이터 제거
                    buffer.RemoveAll(p => p.Timestamp < cutoffTime);

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

                    if (newPackets.Count > 0)
                    {
                        try
                        {
                            // 중요: 이 콜백은 Worker 스레드에서 직접 호출됩니다.
                            listeners[topic]?.Invoke(newPackets);
                        }
                        catch (Exception ex)
                        {
                            Debug.LogError($"[Worker] 리스너 실행 중 오류 발생 (Topic: {topic}): {ex.Message}");
                        }

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