Unity × AWSでつくるリアルタイムマルチプレイ通信【AppSync Events・チュートリアル＆コード例】

AppSyncClient.cs

using UnityEngine;
using System;
using System.Collections;
using NativeWebSocket;
using System.Text;
using System.Threading.Tasks;
using System.Collections.Generic;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;

#region DataModels

namespace AppSync.Messages
{
    
    public class WssHeader
    {
        [JsonProperty("host")]
        public string Host { get; set; }
        [JsonProperty("x-api-key")]
        public string ApiKey { get; set; }
    }

    
    public class ConnectionInitRequest
    {
        [JsonProperty("type")]
        public string Type { get; } = "connection_init";
    }

    
    public class SubscribeRequest
    {
        [JsonProperty("type")]
        public string Type { get; } = "subscribe";
        [JsonProperty("id")]
        public string Id { get; set; }
        [JsonProperty("channel")]
        public string Channel { get; set; }
        [JsonProperty("authorization")]
        public Dictionarystring, string> Authorization { get; set; }
    }

    
    public class PublishRequest
    {
        [JsonProperty("type")]
        public string Type { get; } = "publish";
        [JsonProperty("id")]
        public string Id { get; set; }
        [JsonProperty("channel")]
        public string Channel { get; set; }
        [JsonProperty("events")]
        public Liststring> Events { get; set; }
        [JsonProperty("authorization")]
        public Dictionarystring, string> Authorization { get; set; }
    }

    
    public class EventPayload
    {
        [JsonProperty("message")]
        public string Message { get; set; }
        [JsonProperty("timestamp")]
        public string Timestamp { get; set; }
    }

    
    public class UnsubscribeRequest
    {
        [JsonProperty("type")]
        public string Type { get; } = "unsubscribe";
        [JsonProperty("id")]
        public string Id { get; set; }
    }
}
#endregion

public class AppSyncClient : MonoBehaviour
{
    [Header("AWS AppSync 設定")]
    [Tooltip("例: xxx.appsync-api.us-east-1.amazonaws.com")]
    [SerializeField] private string httpDomain;
    [Tooltip("例: xxx.appsync-realtime-api.us-east-1.amazonaws.com")]
    [SerializeField] private string realtimeDomain;
    [Tooltip("AppSync認可用APIキー")]
    [SerializeField] private string apiKey;
    [Tooltip("publish・subscribeするチャンネル")]
    [SerializeField] private string channel = "/default/test";
    
    private WebSocket _websocket;
    private bool _isConnected = false;
    private string _currentSubscriptionId;

    private string RealtimeEndpoint => $"wss://{realtimeDomain}/event/realtime";

    private string GetEncodedHeaderInfo()
    {
        var header = new AppSync.Messages.WssHeader
        {
            Host = httpDomain,
            ApiKey = apiKey
        };
        string jsonHeader = JsonConvert.SerializeObject(header);
        byte[] headerBytes = Encoding.UTF8.GetBytes(jsonHeader);
        string base64Header = Convert.ToBase64String(headerBytes)
            .Replace('+', '-')
            .Replace("https://zenn.dev/", '_')
            .TrimEnd('=');
        return $"header-{base64Header}";
    }

    async void Start()
    {
        if (string.IsNullOrEmpty(apiKey) || string.IsNullOrEmpty(httpDomain) || string.IsNullOrEmpty(realtimeDomain))
        {
            Debug.LogError("AppSyncの設定が不足しています。インスペクターで設定してください。");
            return;
        }

        await InitializeConnection();
    }

    private async Task InitializeConnection()
    {
        try
        {
            Debug.Log($"AppSyncに接続します: {RealtimeEndpoint}");

            var subprotocols = new Liststring> 
            {  
                "aws-appsync-event-ws", 
                GetEncodedHeaderInfo() 
            };

            Debug.Log($"subprotocolsを使用: [{string.Join(", ", subprotocols)}]");

            _websocket = new WebSocket(RealtimeEndpoint, subprotocols);

            _websocket.OnOpen += () =>
            {
                _isConnected = true;
                Debug.Log("AppSyncへの接続が確立されました。");
                SendConnectionInit();
            };

            _websocket.OnMessage += OnMessageReceived;

            _websocket.OnError += (e) =>
            {
                Debug.LogError($"AppSyncエラー: {e}");
                _isConnected = false;
            };

            _websocket.OnClose += (e) =>
            {
                Debug.Log($"AppSync接続が閉じられました: {e}");
                _isConnected = false;
            };

            await _websocket.Connect();
        }
        catch (Exception e)
        {
            Debug.LogError($"AppSync接続の初期化に失敗: {e.Message}\n{e.StackTrace}");
        }
    }

    private void OnMessageReceived(byte[] bytes)
    {
        var message = Encoding.UTF8.GetString(bytes);
        Debug.Log($"メッセージ受信: {message}");
        
        try
        {
            var json = JObject.Parse(message);
            string type = (string)json["type"];

            switch (type)
            {
                case "connection_ack":
                    Debug.Log("接続が確認されました。チャンネルをsubscribeします...");
                    SubscribeToChannel();
                    break;
                case "subscribe_success":
                    Debug.Log("チャンネルをsubscribeしました。");
                    break;
                case "publish_success":
                    Debug.Log("イベントをpublishしました。");
                    break;
                case "data":
                    HandleDataMessage(json);
                    break;
                case "ka": 
                    Debug.Log("Keep-aliveメッセージを受信しました。");
                    break;
                case "unsubscribe_success":
                    Debug.Log("チャンネルをunsubscribeしました。");
                    break;
                default:
                    Debug.LogWarning($"未処理のメッセージタイプ '{type}': {json.ToString()}");
                    break;
            }
        }
        catch (JsonException e)
        {
            Debug.LogError($"JSONメッセージの解析に失敗: {e.Message}");
        }
    }

    private void HandleDataMessage(JObject json)
    {
        if (json["event"] == null) return;
        
        string eventStr = (string)json["event"];
        try
        {
            var eventObj = JObject.Parse(eventStr);
            Debug.Log($"イベントデータ受信: {eventObj.ToString()}");
            
            if (eventObj["message"] != null)
            {
                string message = (string)eventObj["message"];
                Debug.Log($"イベントメッセージ: {message}");
            }
        }
        catch (JsonException)
        {
             Debug.Log($"イベントデータがJSONオブジェクトではありません: {eventStr}");
        }
    }
    
    private async void SendConnectionInit()
    {
        try
        {
            var request = new AppSync.Messages.ConnectionInitRequest();
            string json = JsonConvert.SerializeObject(request);
            await _websocket.SendText(json);
            Debug.Log("connection_init メッセージを送信しました。");
        }
        catch (Exception e)
        {
            Debug.LogError($"connection_init の送信に失敗: {e.Message}\n{e.StackTrace}");
        }
    }

    private async void SubscribeToChannel()
    {
        try
        {
            _currentSubscriptionId = Guid.NewGuid().ToString();
            var request = new AppSync.Messages.SubscribeRequest
            {
                Id = _currentSubscriptionId,
                Channel = channel,
                Authorization = new Dictionarystring, string>
                {
                    { "host", httpDomain },
                    { "x-api-key", apiKey }
                }
            };
            string json = JsonConvert.SerializeObject(request);
            await _websocket.SendText(json);
            Debug.Log($"subscribe リクエストを送信しました (ID: {_currentSubscriptionId})");
        }
        catch (Exception e)
        {
            Debug.LogError($"subscribeに失敗: {e.Message}\n{e.StackTrace}");
        }
    }

    public async void PublishEvent(string message)
    {
        if (!_isConnected)
        {
            Debug.LogWarning("接続されていません。イベントを発行できません。");
            return;
        }

        try
        {
            var payload = new AppSync.Messages.EventPayload
            {
                Message = message,
                Timestamp = DateTime.UtcNow.ToString("o")
            };
            string innerJson = JsonConvert.SerializeObject(payload);

            var request = new AppSync.Messages.PublishRequest
            {
                Id = Guid.NewGuid().ToString(),
                Channel = channel,
                Events = new Liststring> { innerJson },
                Authorization = new Dictionarystring, string> { { "x-api-key", apiKey } }
            };
            
            string json = JsonConvert.SerializeObject(request);
            await _websocket.SendText(json);
            Debug.Log($"イベントを発行しました: {json}");
        }
        catch (Exception e)
        {
            Debug.LogError($"イベントの発行に失敗: {e.Message}\n{e.StackTrace}");
        }
    }

    public async Task Unsubscribe()
    {
        if (!_isConnected)
        {
            Debug.LogWarning("接続されていません。subscribe解除できません。");
            return;
        }

        if (string.IsNullOrEmpty(_currentSubscriptionId))
        {
            Debug.LogWarning("unsubscribeするアクティブなsubscriptionがありません。");
            return;
        }

        try
        {
            var request = new AppSync.Messages.UnsubscribeRequest { Id = _currentSubscriptionId };
            string json = JsonConvert.SerializeObject(request);
            await _websocket.SendText(json);
            Debug.Log($"unsubscribe リクエストを送信しました (ID: {_currentSubscriptionId})");
            _currentSubscriptionId = null;
        }
        catch (Exception e)
        {
            Debug.LogError($"unsubscribeに失敗: {e.Message}\n{e.StackTrace}");
        }
    }

    async void Update()
    {
        #if !UNITY_WEBGL || UNITY_EDITOR
            _websocket?.DispatchMessageQueue();
        #endif

        
        if (Input.GetKeyDown(KeyCode.P))
        {
            PublishEvent("Hello from Unity!");
        }
        if (Input.GetKeyDown(KeyCode.U))
        {
            await Unsubscribe();
        }
    }

}