386 lines
16 KiB
C#
386 lines
16 KiB
C#
using Cysharp.Threading.Tasks;
|
|
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;
|
|
public bool IsRunning => isRunning;
|
|
|
|
/// <summary>
|
|
/// 스레드를 안전하게 종료시키기 위한 CancellationTokenSource 입니다.
|
|
/// Exit() 메서드가 호출되면 취소 신호를 보냅니다.
|
|
/// </summary>
|
|
private 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;
|
|
if (cancellationTokenSource.IsCancellationRequested)
|
|
{
|
|
cancellationTokenSource.Dispose();
|
|
cancellationTokenSource = new CancellationTokenSource();
|
|
}
|
|
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; // Exit() 호출 시 루프 종료
|
|
}
|
|
}
|
|
catch (ObjectDisposedException) { break; }
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// 직접 전파 모드의 메인 루프입니다. Exit()이 호출될 때까지 스레드를 대기시킵니다.
|
|
/// </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;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|