refactor(net): own live session routing
This commit is contained in:
Erik
parent
707e606e35
commit
961bdd07b7
12 changed files with 1645 additions and 543 deletions
341
src/AcDream.App/Net/LiveSessionEventRouter.cs
Normal file
341
src/AcDream.App/Net/LiveSessionEventRouter.cs
Normal file
|
|
@ -0,0 +1,341 @@
|
|||
using AcDream.Core.Chat;
|
||||
using AcDream.Core.Combat;
|
||||
using AcDream.Core.Items;
|
||||
using AcDream.Core.Net;
|
||||
using AcDream.Core.Net.Messages;
|
||||
using AcDream.Core.Player;
|
||||
using AcDream.Core.Social;
|
||||
using AcDream.Core.Spells;
|
||||
|
||||
namespace AcDream.App.Net;
|
||||
|
||||
internal sealed record LiveEntitySessionSink(
|
||||
Action<WorldSession.EntitySpawn> Spawned,
|
||||
Action<DeleteObject.Parsed> Deleted,
|
||||
Action<PickupEvent.Parsed> PickedUp,
|
||||
Action<WorldSession.EntityMotionUpdate> MotionUpdated,
|
||||
Action<WorldSession.EntityPositionUpdate> PositionUpdated,
|
||||
Action<VectorUpdate.Parsed> VectorUpdated,
|
||||
Action<SetState.Parsed> StateUpdated,
|
||||
Action<ParentEvent.Parsed> ParentUpdated,
|
||||
Action<uint> TeleportStarted,
|
||||
Action<ObjDescEvent.Parsed> AppearanceUpdated,
|
||||
Action<PlayPhysicsScript> PlayPhysicsScript,
|
||||
Action<PlayPhysicsScriptType> PlayPhysicsScriptType);
|
||||
|
||||
internal sealed record LiveEnvironmentSessionSink(
|
||||
Action<uint> EnvironChanged,
|
||||
Action<double> ServerTimeUpdated);
|
||||
|
||||
internal sealed record LiveInventorySessionBindings(
|
||||
ClientObjectTable Objects,
|
||||
LocalPlayerState LocalPlayer,
|
||||
Func<uint> PlayerGuid,
|
||||
Action<IReadOnlyList<ShortcutEntry>>? OnShortcuts,
|
||||
Action<uint>? OnUseDone,
|
||||
ItemManaState? ItemMana,
|
||||
Action<IReadOnlyList<(uint Id, uint Amount)>>? OnDesiredComponents,
|
||||
ExternalContainerState? ExternalContainers);
|
||||
|
||||
internal sealed record LiveCharacterSessionBindings(
|
||||
CombatState Combat,
|
||||
Spellbook Spellbook,
|
||||
Func<uint, IReadOnlyDictionary<uint, uint>, uint>? ResolveSkillFormulaBonus,
|
||||
Action<int, int>? OnSkillsUpdated,
|
||||
Action<GameEvents.CharacterConfirmationRequest>? OnConfirmationRequest,
|
||||
Action<GameEvents.CharacterConfirmationDone>? OnConfirmationDone,
|
||||
Action<uint, uint>? OnCharacterOptions,
|
||||
Func<double>? ClientTime);
|
||||
|
||||
internal sealed record LiveSocialSessionBindings(
|
||||
ChatLog Chat,
|
||||
TurbineChatState TurbineChat,
|
||||
FriendsState? Friends,
|
||||
SquelchState? Squelch);
|
||||
|
||||
/// <summary>
|
||||
/// Owns every inbound subscription for one exact live session. Domain state
|
||||
/// remains in the supplied sinks; this class owns only routing and teardown.
|
||||
/// </summary>
|
||||
internal sealed class LiveSessionEventRouter : IDisposable
|
||||
{
|
||||
private readonly LiveSessionSubscriptionSet _subscriptions = new();
|
||||
private readonly Action<int>? _constructionCheckpoint;
|
||||
private int _constructionStep;
|
||||
private int _accepting = 1;
|
||||
|
||||
public LiveSessionEventRouter(
|
||||
WorldSession session,
|
||||
LiveEntitySessionSink entities,
|
||||
LiveEnvironmentSessionSink environment,
|
||||
LiveInventorySessionBindings inventory,
|
||||
LiveCharacterSessionBindings character,
|
||||
LiveSocialSessionBindings social,
|
||||
Action<int>? constructionCheckpoint = null)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(session);
|
||||
Validate(entities, environment, inventory, character, social);
|
||||
_constructionCheckpoint = constructionCheckpoint;
|
||||
|
||||
try
|
||||
{
|
||||
// Preserve the shipped pre-Connect registration order. Property
|
||||
// state is installed before lifecycle packets can be dispatched.
|
||||
_subscriptions.Add(ObjectTableWiring.Wire(
|
||||
session,
|
||||
inventory.Objects,
|
||||
inventory.PlayerGuid,
|
||||
inventory.LocalPlayer,
|
||||
IsAccepting));
|
||||
ConstructionCheckpoint();
|
||||
_subscriptions.Add(CombatStateWiring.Wire(
|
||||
session,
|
||||
character.Combat,
|
||||
IsAccepting));
|
||||
ConstructionCheckpoint();
|
||||
|
||||
Subscribe(h => session.EntitySpawned += h, h => session.EntitySpawned -= h, entities.Spawned);
|
||||
Subscribe(h => session.EntityDeleted += h, h => session.EntityDeleted -= h, entities.Deleted);
|
||||
Subscribe(h => session.EntityPickedUp += h, h => session.EntityPickedUp -= h, entities.PickedUp);
|
||||
Subscribe(h => session.MotionUpdated += h, h => session.MotionUpdated -= h, entities.MotionUpdated);
|
||||
Subscribe(h => session.PositionUpdated += h, h => session.PositionUpdated -= h, entities.PositionUpdated);
|
||||
Subscribe(h => session.VectorUpdated += h, h => session.VectorUpdated -= h, entities.VectorUpdated);
|
||||
Subscribe(h => session.StateUpdated += h, h => session.StateUpdated -= h, entities.StateUpdated);
|
||||
Subscribe(h => session.ParentUpdated += h, h => session.ParentUpdated -= h, entities.ParentUpdated);
|
||||
Subscribe(h => session.TeleportStarted += h, h => session.TeleportStarted -= h, entities.TeleportStarted);
|
||||
Subscribe(h => session.AppearanceUpdated += h, h => session.AppearanceUpdated -= h, entities.AppearanceUpdated);
|
||||
Subscribe(
|
||||
h => session.PlayPhysicsScriptReceived += h,
|
||||
h => session.PlayPhysicsScriptReceived -= h,
|
||||
entities.PlayPhysicsScript);
|
||||
Subscribe(
|
||||
h => session.PlayPhysicsScriptTypeReceived += h,
|
||||
h => session.PlayPhysicsScriptTypeReceived -= h,
|
||||
entities.PlayPhysicsScriptType);
|
||||
|
||||
Subscribe(h => session.EnvironChanged += h, h => session.EnvironChanged -= h, environment.EnvironChanged);
|
||||
Subscribe(h => session.ServerTimeUpdated += h, h => session.ServerTimeUpdated -= h, environment.ServerTimeUpdated);
|
||||
|
||||
_subscriptions.Add(GameEventWiring.WireAll(
|
||||
session.GameEvents,
|
||||
inventory.Objects,
|
||||
character.Combat,
|
||||
character.Spellbook,
|
||||
social.Chat,
|
||||
inventory.LocalPlayer,
|
||||
social.TurbineChat,
|
||||
onSkillsUpdated: character.OnSkillsUpdated,
|
||||
resolveSkillFormulaBonus: character.ResolveSkillFormulaBonus,
|
||||
onShortcuts: inventory.OnShortcuts,
|
||||
playerGuid: inventory.PlayerGuid,
|
||||
onUseDone: inventory.OnUseDone,
|
||||
itemMana: inventory.ItemMana,
|
||||
onConfirmationRequest: character.OnConfirmationRequest,
|
||||
onConfirmationDone: character.OnConfirmationDone,
|
||||
friends: social.Friends,
|
||||
squelch: social.Squelch,
|
||||
onDesiredComponents: inventory.OnDesiredComponents,
|
||||
onCharacterOptions: character.OnCharacterOptions,
|
||||
clientTime: character.ClientTime,
|
||||
externalContainers: inventory.ExternalContainers,
|
||||
accepting: IsAccepting));
|
||||
ConstructionCheckpoint();
|
||||
_subscriptions.Add(new CombatChatTranslator(
|
||||
character.Combat,
|
||||
social.Chat,
|
||||
IsAccepting));
|
||||
ConstructionCheckpoint();
|
||||
|
||||
Subscribe<HearSpeech.Parsed>(h => session.SpeechHeard += h, h => session.SpeechHeard -= h, speech =>
|
||||
social.Chat.OnLocalSpeech(
|
||||
speech.SenderName,
|
||||
speech.Text,
|
||||
speech.SenderGuid,
|
||||
speech.IsRanged));
|
||||
Subscribe<ServerMessage.Parsed>(
|
||||
h => session.ServerMessageReceived += h,
|
||||
h => session.ServerMessageReceived -= h,
|
||||
message => social.Chat.OnSystemMessage(message.Message, message.ChatType));
|
||||
Subscribe<EmoteText.Parsed>(h => session.EmoteHeard += h, h => session.EmoteHeard -= h, emote =>
|
||||
social.Chat.OnEmote(emote.SenderName, emote.Text, emote.SenderGuid));
|
||||
Subscribe<SoulEmote.Parsed>(h => session.SoulEmoteHeard += h, h => session.SoulEmoteHeard -= h, emote =>
|
||||
social.Chat.OnSoulEmote(emote.SenderName, emote.Text, emote.SenderGuid));
|
||||
Subscribe<PlayerKilled.Parsed>(
|
||||
h => session.PlayerKilledReceived += h,
|
||||
h => session.PlayerKilledReceived -= h,
|
||||
killed => social.Chat.OnPlayerKilled(
|
||||
killed.DeathMessage,
|
||||
killed.VictimGuid,
|
||||
killed.KillerGuid));
|
||||
Subscribe<TurbineChat.Parsed>(
|
||||
h => session.TurbineChatReceived += h,
|
||||
h => session.TurbineChatReceived -= h,
|
||||
parsed => RouteTurbineChat(social.Chat, parsed));
|
||||
Subscribe<PrivateUpdateVital.ParsedFull>(h => session.VitalUpdated += h, h => session.VitalUpdated -= h, vital =>
|
||||
inventory.LocalPlayer.OnVitalUpdate(
|
||||
vital.VitalId,
|
||||
vital.Ranks,
|
||||
vital.Start,
|
||||
vital.Xp,
|
||||
vital.Current));
|
||||
Subscribe<PrivateUpdateVital.ParsedCurrent>(
|
||||
h => session.VitalCurrentUpdated += h,
|
||||
h => session.VitalCurrentUpdated -= h,
|
||||
vital => inventory.LocalPlayer.OnVitalCurrent(vital.VitalId, vital.Current));
|
||||
}
|
||||
catch (Exception constructionError)
|
||||
{
|
||||
Interlocked.Exchange(ref _accepting, 0);
|
||||
try
|
||||
{
|
||||
_subscriptions.Dispose();
|
||||
}
|
||||
catch (Exception cleanupError)
|
||||
{
|
||||
throw new AggregateException(
|
||||
"live-session event routing failed and cleanup also failed",
|
||||
constructionError,
|
||||
cleanupError);
|
||||
}
|
||||
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
public bool Accepting => IsAccepting();
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
Interlocked.Exchange(ref _accepting, 0);
|
||||
_subscriptions.Dispose();
|
||||
}
|
||||
|
||||
private void Subscribe<T>(
|
||||
Action<Action<T>> attach,
|
||||
Action<Action<T>> detach,
|
||||
Action<T> sink)
|
||||
{
|
||||
Action<T> handler = value =>
|
||||
{
|
||||
if (Volatile.Read(ref _accepting) != 0)
|
||||
sink(value);
|
||||
};
|
||||
|
||||
attach(handler);
|
||||
try
|
||||
{
|
||||
_subscriptions.Add(() => detach(handler));
|
||||
}
|
||||
catch
|
||||
{
|
||||
detach(handler);
|
||||
throw;
|
||||
}
|
||||
|
||||
ConstructionCheckpoint();
|
||||
}
|
||||
|
||||
private void ConstructionCheckpoint() =>
|
||||
_constructionCheckpoint?.Invoke(++_constructionStep);
|
||||
|
||||
private bool IsAccepting() => Volatile.Read(ref _accepting) != 0;
|
||||
|
||||
private static void RouteTurbineChat(ChatLog chat, TurbineChat.Parsed parsed)
|
||||
{
|
||||
if (parsed.Body is not TurbineChat.Payload.EventSendToRoom message)
|
||||
return;
|
||||
|
||||
chat.OnChannelBroadcast(
|
||||
message.RoomId,
|
||||
message.SenderName,
|
||||
message.Message,
|
||||
TurbineChatRouting.DisplayName(message.RoomId, message.ChatType));
|
||||
}
|
||||
|
||||
private static void Validate(
|
||||
LiveEntitySessionSink entities,
|
||||
LiveEnvironmentSessionSink environment,
|
||||
LiveInventorySessionBindings inventory,
|
||||
LiveCharacterSessionBindings character,
|
||||
LiveSocialSessionBindings social)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(entities);
|
||||
ArgumentNullException.ThrowIfNull(environment);
|
||||
ArgumentNullException.ThrowIfNull(inventory);
|
||||
ArgumentNullException.ThrowIfNull(character);
|
||||
ArgumentNullException.ThrowIfNull(social);
|
||||
ArgumentNullException.ThrowIfNull(entities.Spawned);
|
||||
ArgumentNullException.ThrowIfNull(entities.Deleted);
|
||||
ArgumentNullException.ThrowIfNull(entities.PickedUp);
|
||||
ArgumentNullException.ThrowIfNull(entities.MotionUpdated);
|
||||
ArgumentNullException.ThrowIfNull(entities.PositionUpdated);
|
||||
ArgumentNullException.ThrowIfNull(entities.VectorUpdated);
|
||||
ArgumentNullException.ThrowIfNull(entities.StateUpdated);
|
||||
ArgumentNullException.ThrowIfNull(entities.ParentUpdated);
|
||||
ArgumentNullException.ThrowIfNull(entities.TeleportStarted);
|
||||
ArgumentNullException.ThrowIfNull(entities.AppearanceUpdated);
|
||||
ArgumentNullException.ThrowIfNull(entities.PlayPhysicsScript);
|
||||
ArgumentNullException.ThrowIfNull(entities.PlayPhysicsScriptType);
|
||||
ArgumentNullException.ThrowIfNull(environment.EnvironChanged);
|
||||
ArgumentNullException.ThrowIfNull(environment.ServerTimeUpdated);
|
||||
ArgumentNullException.ThrowIfNull(inventory.Objects);
|
||||
ArgumentNullException.ThrowIfNull(inventory.LocalPlayer);
|
||||
ArgumentNullException.ThrowIfNull(inventory.PlayerGuid);
|
||||
ArgumentNullException.ThrowIfNull(character.Combat);
|
||||
ArgumentNullException.ThrowIfNull(character.Spellbook);
|
||||
ArgumentNullException.ThrowIfNull(social.Chat);
|
||||
ArgumentNullException.ThrowIfNull(social.TurbineChat);
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed class LiveSessionSubscriptionSet : IDisposable
|
||||
{
|
||||
private List<IDisposable>? _subscriptions = [];
|
||||
|
||||
public void Add(IDisposable subscription)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(subscription);
|
||||
List<IDisposable>? subscriptions = _subscriptions;
|
||||
if (subscriptions is null)
|
||||
{
|
||||
subscription.Dispose();
|
||||
throw new ObjectDisposedException(nameof(LiveSessionSubscriptionSet));
|
||||
}
|
||||
|
||||
subscriptions.Add(subscription);
|
||||
}
|
||||
|
||||
public void Add(Action unsubscribe) => Add(new ActionSubscription(unsubscribe));
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
List<IDisposable>? subscriptions = Interlocked.Exchange(ref _subscriptions, null);
|
||||
if (subscriptions is null)
|
||||
return;
|
||||
|
||||
List<Exception>? errors = null;
|
||||
for (int index = subscriptions.Count - 1; index >= 0; index--)
|
||||
{
|
||||
try
|
||||
{
|
||||
subscriptions[index].Dispose();
|
||||
}
|
||||
catch (Exception error)
|
||||
{
|
||||
(errors ??= []).Add(error);
|
||||
}
|
||||
}
|
||||
|
||||
if (errors is not null)
|
||||
throw new AggregateException(
|
||||
"one or more live-session subscriptions failed to detach",
|
||||
errors);
|
||||
}
|
||||
|
||||
private sealed class ActionSubscription(Action unsubscribe) : IDisposable
|
||||
{
|
||||
private Action? _unsubscribe = unsubscribe;
|
||||
|
||||
public void Dispose() => Interlocked.Exchange(ref _unsubscribe, null)?.Invoke();
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue