warrentyexpired/ModernUO
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

NATS driven broadcast

Browse source
This commit is contained in:
Leath Cooper 2024-10-19 17:52:53 -04:00
parent c9ae5e3bd2
commit c06c84ca9c
4 changed files with 123 additions and 114 deletions
Download patch file Download diff file Expand all files Collapse all files

83
Projects/Server/Events/MessageBus.cs
View file

@ -1,97 +1,20 @@
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;
public static NatsClient Client => natsClient;
private static NatsJsonContextSerializer<BroadcastMessage> broadcastSerializer =
new (BroadcastJsonContext.Default);
private static INatsJSContext jetstreamContext;
public static INatsJSContext Context => jetstreamContext;
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<T>(string subject, T data, INatsSerializer<T> serializer = null)
{
await natsClient.PublishAsync<T>(subject, data);
}
public static void Kill()
{
cancellationTokenSource.Cancel();
}
}

119
Projects/UOContent/Commands/Broadcast.cs Normal file
View file

@ -0,0 +1,119 @@
using System;
using System.Text.Json.Serialization;
using System.Threading;
using Server.Network;
using NATS.Client;
using NATS.Client.Core;
using NATS.Client.JetStream;
using NATS.Client.JetStream.Models;
using NATS.Net;
namespace Server.Commands;
[JsonSerializable(typeof(BroadcastPayload))]
internal partial class BroadcastJsonContext : JsonSerializerContext;
public record BroadcastPayload
{
[JsonPropertyName("hue")]
public int Hue { get; set; } = 0x482;
[JsonPropertyName("message")]
public string? Message { get; set; }
}
public static class Broadcast
{
private static INatsJSConsumer incomingBroadcastsConsumer;
private static readonly NatsJsonContextSerializer<BroadcastPayload> broadcastSerializer =
new(BroadcastJsonContext.Default);
public static void Configure()
{
CommandSystem.Register("BCast", AccessLevel.GameMaster, BroadcastMessage_OnCommand);
CommandSystem.Register("SMsg", AccessLevel.Counselor, StaffMessage_OnCommand);
}
public static void Initialize()
{
StartBroadcasts();
}
[Usage("BCast <text>")]
[Aliases("B", "BC")]
[Description("Broadcasts a message to everyone online.")]
public static void BroadcastMessage_OnCommand(CommandEventArgs e)
{
EmitBroadcast($"Staff message from {e.Mobile.Name}:", 0x482);
EmitBroadcast($"{e.ArgString}", 0x21);
// BroadcastMessage(AccessLevel.Player, 0x482, $"Staff message from {e.Mobile.Name}:");
// BroadcastMessage(AccessLevel.Player, 0x482, e.ArgString);
}
[Usage("SMsg <text>")]
[Aliases("S", "SM")]
[Description("Broadcasts a message to all online staff.")]
public static void StaffMessage_OnCommand(CommandEventArgs e)
{
BroadcastMessage(AccessLevel.Counselor, e.Mobile.SpeechHue, $"[{e.Mobile.Name}] {e.ArgString}");
}
public static async void EmitBroadcast(string message, int hue)
{
await MessageBus.Client.PublishAsync(
subject: "broadcast.all",
data: new BroadcastPayload()
{
Message = message,
Hue = hue,
},
serializer: broadcastSerializer
);
}
public static void BroadcastMessage(AccessLevel ac, int hue, string message)
{
foreach (var state in NetState.Instances)
{
var m = state.Mobile;
if (m?.AccessLevel >= ac)
{
m.SendMessage(hue, message);
}
}
}
private static async void StartBroadcasts()
{
await MessageBus.Context.CreateStreamAsync(
new StreamConfig(name: "BROADCASTS", subjects: ["broadcast.>"])
);
incomingBroadcastsConsumer = await MessageBus.Context.CreateOrUpdateConsumerAsync(
"BROADCASTS",
new ConsumerConfig("broadcast_incoming_consumer")
);
CancellationTokenSource cancellationTokenSource = new();
await foreach (var msg in incomingBroadcastsConsumer.ConsumeAsync(
serializer: broadcastSerializer,
cancellationToken: cancellationTokenSource.Token
))
{
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
);
}
}
}

34
Projects/UOContent/Commands/Handlers.cs
View file

@ -33,8 +33,6 @@ namespace Server.Commands
Register("Help", AccessLevel.Player, Help_OnCommand);
Register("Move", AccessLevel.GameMaster, Move_OnCommand);
Register("Client", AccessLevel.Counselor, Client_OnCommand);
Register("SMsg", AccessLevel.Counselor, StaffMessage_OnCommand);
Register("BCast", AccessLevel.GameMaster, BroadcastMessage_OnCommand);
Register("Bank", AccessLevel.GameMaster, Bank_OnCommand);
Register("Echo", AccessLevel.Counselor, Echo_OnCommand);
Register("Sound", AccessLevel.GameMaster, Sound_OnCommand);
@ -625,38 +623,6 @@ namespace Server.Commands
}
}
[Usage("SMsg <text>")]
[Aliases("S", "SM")]
[Description("Broadcasts a message to all online staff.")]
public static void StaffMessage_OnCommand(CommandEventArgs e)
{
BroadcastMessage(AccessLevel.Counselor, e.Mobile.SpeechHue, $"[{e.Mobile.Name}] {e.ArgString}");
}
[Usage("BCast <text>")]
[Aliases("B", "BC")]
[Description("Broadcasts a message to everyone online.")]
public static void BroadcastMessage_OnCommand(CommandEventArgs e)
{
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)
{
foreach (var state in NetState.Instances)
{
var m = state.Mobile;
if (m?.AccessLevel >= ac)
{
m.SendMessage(hue, message);
}
}
}
[Usage("AutoPageNotify")]
[Aliases("APN")]
[Description("Toggles your auto-page-notify status.")]

1
Projects/UOContent/UOContent.csproj
View file

@ -34,6 +34,7 @@
<Delete Files="..\..\Distribution\Assemblies\ModernUO.Serialization.Annotations.dll" ContinueOnError="true" />
</Target>
<ItemGroup>
<PackageReference Include="NATS.Net" Version="2.5.1" />
<ProjectReference Include="..\Server\Server.csproj" Private="false" PrivateAssets="All" IncludeAssets="None">
<IncludeInPackage>false</IncludeInPackage>
</ProjectReference>