4
\$\begingroup\$

I'm trying to build a nicely encapsulated mechanism for providing timeouts that wait for a signalling channel to go quiet for a duration. As a fun study of System.Reactive, I decided to base this mechanism on the Observable.Throttle extension method.

Where possible, I hope to avoid using Subjects in my reactive code but I am unfamiliar with which other toys may help me generate this throttled sequence. I often read that Subjects should only be used for example code because they introduce issues where mutable state is exposed and they are a 'lazy', 'non-Reactive' way of generating sequences. I believe I avoid this in an acceptable way. I am unsure if:

  1. this is appropriate or generally acceptable use of subjects
  2. the lifetime management is done correctly
  3. there is a more concise way of achieving all of this

using System;
using System.Reactive;
using System.Reactive.Linq;
using System.Reactive.Subjects;

/// <summary>
/// Provides a timeout mechanism that will not timeout if it is signalled often enough
/// </summary>
internal class TrafficTimeout
{
    private readonly Action onTimeout;
    private object signalLock = new object();
    private IObserver<Unit> signals;

    /// <summary>
    /// Initialises a new instance of the <see cref="TrafficTimeout"/> class.
    /// </summary>
    /// <param name="timeout">The duration to wait after receiving signals before timing out.</param>
    /// <param name="onTimeout">The <see cref="Action"/> to perform when the the timeout duration expires.</param>
    public TrafficTimeout(TimeSpan timeout, Action onTimeout)
    {
        // Subscribe to a throttled observable to trigger the expirey
        var messageQueue = new BehaviorSubject<Unit>(Unit.Default);
        IDisposable subscription = null;
        subscription = messageQueue.Throttle(timeout).Subscribe(
        p =>
        {
            messageQueue.OnCompleted();
            messageQueue.Dispose();
            onTimeout();
        });

        this.signals = messageQueue.AsObserver();
        this.onTimeout = onTimeout;
    }

    /// <summary>
    /// Signals that traffic has been received.
    /// </summary>
    public void Signal()
    {
        lock (this.signalLock)
        {
            this.signals.OnNext(Unit.Default);
        }
    }
}

Since creating this question, I've found a bug with this code where calling OnCompleted from within the subcribe callback results in two Unit values being emmited. I am assuming that this is the same signal.

\$\endgroup\$
  • \$\begingroup\$ I really want to avoid using Subjects in my reactive code - is there any particular reason for this? I'm just curious what's wrong with them, not going to suggest it. \$\endgroup\$ – t3chb0t Nov 17 '17 at 21:21
  • 1
    \$\begingroup\$ @t3chb0t I've made some edits that should answer your question. \$\endgroup\$ – Gusdor Nov 17 '17 at 21:30

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.