ReactiveNode.cs

using System.Reactive;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using CodeCasa.AutomationPipelines.Lights.Utils;
using CodeCasa.Lights;
using Microsoft.Extensions.Logging;

namespace CodeCasa.AutomationPipelines.Lights.ReactiveNode;

/// <summary>
/// A pipeline node that dynamically switches between different child nodes based on an observable source.
/// The active node can change at runtime, allowing for reactive behavior switching.
/// </summary>
public class ReactiveNode : PipelineNode<LightTransition>
{
    private readonly string? _name;
    private readonly ILogger<ReactiveNode>? _logger;
    private readonly IEqualityComparer<LightTransition>? _equalityComparer;
    private readonly Subject<Unit> _nodeChangedSubject = new();
    private readonly Subject<Action> _stateQueue = new();
    private readonly IDisposable _queueSubscription;
    private IDisposable? _nodeObservableSubscription;
    private IDisposable? _activeNodeSubscription;

    /// <summary>
    /// Initializes a new instance of the <see cref="ReactiveNode"/> class.
    /// </summary>
    /// <param name="nodeObservable">An observable that emits the pipeline nodes to activate. Null values deactivate the current node.</param>
    /// <param name="equalityComparer">Optional equality comparer used to determine whether the output has changed. When <see langword="null"/>, the output is always set when a new value is received.</param>
    public ReactiveNode(IObservable<IPipelineNode<LightTransition>?> nodeObservable, IEqualityComparer<LightTransition>? equalityComparer = null) :
        this(null, nodeObservable, null!, equalityComparer)
    {
    }

    /// <summary>
    /// Initializes a new instance of the <see cref="ReactiveNode"/> class.
    /// </summary>
    /// <param name="name">Optional name for the reactive node, used for logging purposes.</param>
    /// <param name="nodeObservable">An observable that emits the pipeline nodes to activate. Null values deactivate the current node.</param>
    /// <param name="logger">Optional logger for diagnostic information.</param>
    /// <param name="equalityComparer">Optional equality comparer used to determine whether the output has changed. When <see langword="null"/>, the output is always set when a new value is received.</param>
    public ReactiveNode(string? name, IObservable<IPipelineNode<LightTransition>?> nodeObservable, ILogger<ReactiveNode> logger, IEqualityComparer<LightTransition>? equalityComparer = null)
    {
        _name = name;
        _logger = logger;
        _equalityComparer = equalityComparer;
        PassThrough = true;

        _queueSubscription = _stateQueue
            .Synchronize()
            .Subscribe(action => action());

        _nodeObservableSubscription = nodeObservable
            .Subscribe(n =>
            {
                _stateQueue.OnNext(() =>
                {
                    if (n == null)
                    {
                        DeactivateActiveNode();
                        PassThrough = true;
                        _logger?.LogTrace($"{LogPrefix}No active node. Passing through data.");
                    }
                    else
                    {
                        ActivateNode(n);
                    }

                    _nodeChangedSubject.OnNext(Unit.Default);
                });
            });
    }

    /// <summary>
    /// Gets the currently active pipeline node, or <see langword="null"/> if no node is active.
    /// </summary>
    public IPipelineNode<LightTransition>? ActiveNode { get; private set; }
    private string LogPrefix => _name == null ? "" : $"{_name}: ";

    /// <summary>
    /// Gets an observable that emits whenever the active node changes.
    /// </summary>
    public IObservable<Unit> NodeChanged => _nodeChangedSubject.AsObservable();

    /// <inheritdoc />
    protected override void InputReceived(LightTransition? input)
    {
        _stateQueue.OnNext(() =>
        {
            if (ActiveNode != null)
            {
                ActiveNode.Input = input;
            }
        });
    }

    private void DeactivateActiveNode()
    {
        _activeNodeSubscription?.Dispose();
        if (ActiveNode != null)
        {
            ActiveNode.Input = null;
            ActiveNode.DisposeOrDisposeAsync().GetAwaiter().GetResult();
        }

        ActiveNode = null;
        _activeNodeSubscription = null;
    }

    private void ActivateNode(IPipelineNode<LightTransition> node)
    {
        DeactivateActiveNode();
        ActiveNode = node;
        _logger?.LogTrace($"{LogPrefix}Activating {node}.");
        ActiveNode.Input = Input;
        if (_equalityComparer == null || !_equalityComparer.Equals(Output, ActiveNode.Output))
        {
            Output = ActiveNode.Output;
        }
        _activeNodeSubscription = ActiveNode.OnNewOutput.Subscribe(output =>
        {
            _stateQueue.OnNext(() => UpdateOutput(output));
        });
        PassThrough = false;
    }

    private void UpdateOutput(LightTransition? newOutput)
    {
        if (_equalityComparer == null || !_equalityComparer.Equals(Output, newOutput))
        {
            Output = newOutput;
        }
    }

    /// <inheritdoc />
    public override string ToString() => _name ?? base.ToString();

    /// <inheritdoc />
    public override async ValueTask DisposeAsync()
    {
        _nodeObservableSubscription?.Dispose();
        _nodeObservableSubscription = null;

        if (ActiveNode != null)
        {
            await ActiveNode.DisposeOrDisposeAsync();
        }

        _queueSubscription.Dispose();
        _stateQueue.Dispose();
        _nodeChangedSubject.Dispose();

        await base.DisposeAsync();
    }
}