From c9ae5e3bd2db78cf5180cc2a013aedf10acd1da5 Mon Sep 17 00:00:00 2001 From: Leath Cooper Date: Sat, 19 Oct 2024 16:10:11 -0400 Subject: [PATCH] NATS --- Projects/Server/Events/EventSink.cs | 3 + Projects/Server/Events/MessageBus.cs | 97 +++++++++++++++++++++++++ Projects/Server/Items/Item.cs | 6 ++ Projects/Server/Main.cs | 21 ++++-- Projects/Server/Server.csproj | 1 + Projects/UOContent/Commands/Handlers.cs | 6 +- Projects/UOContent/Misc/Broadcasts.cs | 9 +++ 7 files changed, 135 insertions(+), 8 deletions(-) create mode 100644 Projects/Server/Events/MessageBus.cs diff --git a/Projects/Server/Events/EventSink.cs b/Projects/Server/Events/EventSink.cs index 6a72185f4..8b9f83533 100644 --- a/Projects/Server/Events/EventSink.cs +++ b/Projects/Server/Events/EventSink.cs @@ -41,4 +41,7 @@ public static partial class EventSink public static event Action ServerStarted; public static void InvokeServerStarted() => ServerStarted?.Invoke(); + + public static event Action IncomingMessage; + public static void InvokeIncomingMessage(string message, int hue) => IncomingMessage?.Invoke(message, hue); } diff --git a/Projects/Server/Events/MessageBus.cs b/Projects/Server/Events/MessageBus.cs new file mode 100644 index 000000000..5a30d1b49 --- /dev/null +++ b/Projects/Server/Events/MessageBus.cs @@ -0,0 +1,97 @@ +using System; +using System.Text.Json.Serialization; +using System.Threading; +using NATS.Client; +using NATS.Client.Core; +using NATS.Client.JetStream; +using NATS.Client.JetStream.Models; +using NATS.Net; + +namespace Server; + +[JsonSerializable(typeof(BroadcastMessage))] +internal partial class BroadcastJsonContext : JsonSerializerContext; + +public record BroadcastMessage +{ + [JsonPropertyName("hue")] + public int Hue { get; set; } = 0x482; + + [JsonPropertyName("message")] + public string? Message { get; set; } +} + +public static class MessageBus +{ + private static NatsClient natsClient; + private static INatsJSContext jetstreamContext; + private static INatsJSConsumer incomingBroadcastsConsumer; + private static CancellationTokenSource cancellationTokenSource; + + private static NatsJsonContextSerializer broadcastSerializer = + new (BroadcastJsonContext.Default); + + public static void Start() + { + natsClient = new NatsClient(); + jetstreamContext = natsClient.CreateJetStreamContext(); + + StartBroadcasts(); + } + + private static async void StartBroadcasts() + { + await jetstreamContext.CreateStreamAsync( + new StreamConfig(name: "INCOMING_NOTIFICATIONS", subjects: ["incoming.notifications.>"]) + ); + + incomingBroadcastsConsumer = await jetstreamContext.CreateOrUpdateConsumerAsync( + "INCOMING_NOTIFICATIONS", + new ConsumerConfig("incoming_broadcasts_consumer") + { + FilterSubject = "incoming.notifications.broadcast" + } + ); + + cancellationTokenSource = new CancellationTokenSource(); + + await foreach (var msg in incomingBroadcastsConsumer.ConsumeAsync( + cancellationToken: cancellationTokenSource.Token, + serializer: broadcastSerializer + )) + { + if (msg.Data == null) + { + Console.WriteLine($"Message Received w/ null Data"); + continue; + } + + Console.WriteLine($"Message Received {msg.Data}"); + EventSink.InvokeIncomingMessage(msg.Data.Message, msg.Data.Hue); + await msg.AckAsync(cancellationToken: cancellationTokenSource.Token); + } + } + + public static async void SendBroadcast(string message, int hue) + { + await natsClient.PublishAsync( + subject: "incoming.notifications.broadcast", + data: new BroadcastMessage() + { + Message = message, + Hue = hue, + }, + serializer: broadcastSerializer + ); + } + + public static async void Publish(string subject, T data, INatsSerializer serializer = null) + { + await natsClient.PublishAsync(subject, data); + } + + public static void Kill() + { + cancellationTokenSource.Cancel(); + } +} diff --git a/Projects/Server/Items/Item.cs b/Projects/Server/Items/Item.cs index aeab60811..df8048955 100644 --- a/Projects/Server/Items/Item.cs +++ b/Projects/Server/Items/Item.cs @@ -233,6 +233,8 @@ public class Item : IHued, IComparable, ISpawnable, IObjectPropertyListEnt SetLastMoved(); World.AddEntity(this); + + MessageBus.Publish("items.created", $"{Serial}:{itemID}"); } public Item(Serial serial) => Serial = serial; @@ -1072,6 +1074,8 @@ public class Item : IHued, IComparable, ISpawnable, IObjectPropertyListEnt { writer.WriteEncodedInt(info.m_SavedFlags); } + + MessageBus.Publish("items.serialize", $"{Serial}:{ItemID}"); } public void MoveToWorld(WorldLocation worldLocation) @@ -3036,6 +3040,8 @@ public class Item : IHued, IComparable, ISpawnable, IObjectPropertyListEnt // if (version < 9) VerifyCompactInfo(); + + MessageBus.Publish("items.deserialize", $"{Serial}:{ItemID}"); } private void FixHolding_Sandbox() diff --git a/Projects/Server/Main.cs b/Projects/Server/Main.cs index 13577440d..ab19073ff 100644 --- a/Projects/Server/Main.cs +++ b/Projects/Server/Main.cs @@ -189,9 +189,11 @@ public static class Core if (fi.Directory != null && Directory.Exists(fi.Directory.FullName)) { fullPath = fi.Directory.EnumerateFiles( - fi.Name, - new EnumerationOptions { MatchCasing = MatchCasing.CaseInsensitive } - ).FirstOrDefault()?.FullName; + fi.Name, + new EnumerationOptions { MatchCasing = MatchCasing.CaseInsensitive } + ) + .FirstOrDefault() + ?.FullName; } } @@ -317,6 +319,7 @@ public static class Core process.Start(); } + logger.Information("Restart done"); } catch (Exception e) @@ -380,13 +383,15 @@ public static class Core Utility.PopColor(); Utility.PushColor(ConsoleColor.DarkGray); - Console.WriteLine(@"Copyright 2019-2023 ModernUO Development Team + Console.WriteLine( + @"Copyright 2019-2023 ModernUO Development Team This program comes with ABSOLUTELY NO WARRANTY; This is free software, and you are welcome to redistribute it under certain conditions. You should have received a copy of the GNU General Public License along with this program. If not, see . - ".TrimMultiline()); + ".TrimMultiline() + ); Utility.PopColor(); Console.CancelKeyPress += Console_CancelKeyPressed; @@ -426,6 +431,8 @@ public static class Core TileMatrixLoader.LoadTileMatrix(); + MessageBus.Start(); + RegionJsonSerializer.LoadRegions(); World.Load(); @@ -433,6 +440,7 @@ public static class Core TcpServer.Start(); PingServer.Start(); + EventSink.InvokeServerStarted(); RunEventLoop(); } @@ -561,7 +569,8 @@ public static class Core if (World.DirtyTrackingEnabled) { var manualDirtyCheckingAttribute = type.GetCustomAttribute(false); - var codeGennedAttribute = type.GetCustomAttribute(false); + var codeGennedAttribute = + type.GetCustomAttribute(false); if (manualDirtyCheckingAttribute == null && codeGennedAttribute == null) { diff --git a/Projects/Server/Server.csproj b/Projects/Server/Server.csproj index a47b3fb1a..f05fa2864 100644 --- a/Projects/Server/Server.csproj +++ b/Projects/Server/Server.csproj @@ -35,6 +35,7 @@ + diff --git a/Projects/UOContent/Commands/Handlers.cs b/Projects/UOContent/Commands/Handlers.cs index de1e831e2..f3e6cd86a 100644 --- a/Projects/UOContent/Commands/Handlers.cs +++ b/Projects/UOContent/Commands/Handlers.cs @@ -638,8 +638,10 @@ namespace Server.Commands [Description("Broadcasts a message to everyone online.")] public static void BroadcastMessage_OnCommand(CommandEventArgs e) { - BroadcastMessage(AccessLevel.Player, 0x482, $"Staff message from {e.Mobile.Name}:"); - BroadcastMessage(AccessLevel.Player, 0x482, e.ArgString); + MessageBus.SendBroadcast($"Staff message from {e.Mobile.Name}:", 0x482); + MessageBus.SendBroadcast($"{e.ArgString}", 0x21); + // BroadcastMessage(AccessLevel.Player, 0x482, $"Staff message from {e.Mobile.Name}:"); + // BroadcastMessage(AccessLevel.Player, 0x482, e.ArgString); } public static void BroadcastMessage(AccessLevel ac, int hue, string message) diff --git a/Projects/UOContent/Misc/Broadcasts.cs b/Projects/UOContent/Misc/Broadcasts.cs index 74eb2ad4f..90e0c2cbc 100644 --- a/Projects/UOContent/Misc/Broadcasts.cs +++ b/Projects/UOContent/Misc/Broadcasts.cs @@ -1,3 +1,5 @@ +using System; + namespace Server.Misc { public static class Broadcasts @@ -6,6 +8,13 @@ namespace Server.Misc { EventSink.ServerCrashed += EventSink_Crashed; EventSink.Shutdown += EventSink_Shutdown; + EventSink.IncomingMessage += EventSink_IncomingMessage; + } + + public static void EventSink_IncomingMessage(string message, int hue) + { + Console.WriteLine("Incoming Message"); + World.Broadcast(hue, true, message); } public static void EventSink_Crashed(ServerCrashedEventArgs e)