519 lines
13 KiB
C#
519 lines
13 KiB
C#
using System;
|
|
using System.Collections.Generic;
|
|
using System.Diagnostics;
|
|
using System.IO;
|
|
using System.Net.Http;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using WxOrder.Framework.Helpers;
|
|
|
|
namespace LaunchDarkly.EventSource
|
|
{
|
|
/// <summary>
|
|
/// Provides an EventSource client for consuming Server-Sent Events. Additional details on the Server-Sent Events spec
|
|
/// can be found at https://html.spec.whatwg.org/multipage/server-sent-events.html
|
|
/// </summary>
|
|
public class EventSource : IEventSource, IDisposable
|
|
{
|
|
#region Private Fields
|
|
|
|
private readonly Configuration _configuration;
|
|
private readonly HttpClient _httpClient;
|
|
private readonly ILogger _logger;
|
|
|
|
private List<string> _eventDataStringBuffer;
|
|
private MemoryStream _eventDataUtf8ByteBuffer;
|
|
private string _eventName;
|
|
private string _lastEventId;
|
|
private readonly ExponentialBackoffWithDecorrelation _backOff;
|
|
private readonly TimeSpan _configuredInitialRetryDelay;
|
|
private readonly TimeSpan _configuredMaxRetryDelay;
|
|
private CancellationTokenSource _currentRequestToken;
|
|
private readonly CancellationTokenSource _shutdownTokenSource = new CancellationTokenSource();
|
|
private static readonly TimeSpan MaxSleepTime = TimeSpan.FromMilliseconds(int.MaxValue);
|
|
private readonly Stopwatch _connectionTimer = new Stopwatch();
|
|
private ReadyState _readyState;
|
|
|
|
#endregion
|
|
|
|
#region Public Events
|
|
|
|
/// <inheritdoc/>
|
|
public event EventHandler<StateChangedEventArgs> Opened;
|
|
|
|
/// <inheritdoc/>
|
|
public event EventHandler<StateChangedEventArgs> Closed;
|
|
|
|
/// <inheritdoc/>
|
|
public event EventHandler<MessageReceivedEventArgs> MessageReceived;
|
|
|
|
/// <inheritdoc/>
|
|
public event EventHandler<CommentReceivedEventArgs> CommentReceived;
|
|
|
|
/// <inheritdoc/>
|
|
public event EventHandler<ExceptionEventArgs> Error;
|
|
|
|
#endregion Public Events
|
|
|
|
#region Public Properties
|
|
|
|
/// <inheritdoc/>
|
|
public ReadyState ReadyState
|
|
{
|
|
get
|
|
{
|
|
lock (this)
|
|
{
|
|
return _readyState;
|
|
}
|
|
}
|
|
private set
|
|
{
|
|
lock (this)
|
|
{
|
|
_readyState = value;
|
|
}
|
|
}
|
|
}
|
|
|
|
internal TimeSpan BackOffDelay
|
|
{
|
|
get;
|
|
private set;
|
|
}
|
|
|
|
// Exposed for tests, so that retry-bounds behavior can be asserted directly
|
|
internal ExponentialBackoffWithDecorrelation BackOff => _backOff;
|
|
|
|
#endregion
|
|
|
|
#region Public Constructors
|
|
|
|
/// <summary>
|
|
/// Initializes a new instance of the <see cref="EventSource" /> class.
|
|
/// </summary>
|
|
/// <param name="configuration">the configuration</param>
|
|
public EventSource(Configuration configuration)
|
|
{
|
|
_readyState = ReadyState.Raw;
|
|
|
|
_configuration = configuration ?? throw new ArgumentNullException(nameof(configuration));
|
|
|
|
_logger = _configuration.Logger;
|
|
|
|
_configuredInitialRetryDelay = _configuration.InitialRetryDelay;
|
|
_configuredMaxRetryDelay = _configuration.MaxRetryDelay;
|
|
|
|
_backOff = new ExponentialBackoffWithDecorrelation(_configuredInitialRetryDelay,
|
|
_configuredMaxRetryDelay);
|
|
|
|
_httpClient = _configuration.HttpClient ?? CreateHttpClient();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Shortcut for initializing an <see cref="EventSource"/> with only a stream URI
|
|
/// and no custom properties.
|
|
/// </summary>
|
|
/// <param name="uri">the stream URI</param>
|
|
/// <exception cref="ArgumentNullException">if the URI is null</exception>
|
|
public EventSource(Uri uri) : this(Configuration.Builder(uri).Build()) { }
|
|
|
|
#endregion
|
|
|
|
#region Public Methods
|
|
|
|
/// <inheritdoc/>
|
|
public async Task StartAsync()
|
|
{
|
|
bool firstTime = true;
|
|
while (ReadyState != ReadyState.Shutdown)
|
|
{
|
|
if (!firstTime)
|
|
{
|
|
if (_connectionTimer.IsRunning)
|
|
{
|
|
if (_connectionTimer.Elapsed >= _configuration.BackoffResetThreshold)
|
|
{
|
|
// Sustained healthy operation reverts any temporary bounds and resets
|
|
// n. Reverting takes precedence over bounds that were
|
|
// set during the preceding fault window.
|
|
ClearTemporaryRetryDelayBounds();
|
|
_backOff.ResetBackoffN();
|
|
}
|
|
_connectionTimer.Reset();
|
|
}
|
|
await MaybeWaitWithBackOff();
|
|
}
|
|
firstTime = false;
|
|
|
|
CancellationTokenSource newRequestTokenSource = null;
|
|
CancellationToken cancellationToken;
|
|
lock (this)
|
|
{
|
|
if (_readyState == ReadyState.Shutdown)
|
|
{
|
|
// in case Close() was called in between the previous ReadyState check and the creation of the new token
|
|
return;
|
|
}
|
|
newRequestTokenSource = new CancellationTokenSource();
|
|
_currentRequestToken?.Dispose();
|
|
_currentRequestToken = newRequestTokenSource;
|
|
}
|
|
|
|
if (ReadyState == ReadyState.Connecting || ReadyState == ReadyState.Open)
|
|
{
|
|
throw new InvalidOperationException(string.Format(Resources.ErrorAlreadyStarted, ReadyState));
|
|
}
|
|
|
|
SetReadyState(ReadyState.Connecting);
|
|
cancellationToken = newRequestTokenSource.Token;
|
|
|
|
try
|
|
{
|
|
await ConnectToEventSourceAsync(cancellationToken);
|
|
|
|
// ConnectToEventSourceAsync normally doesn't return, unless it detects that the request has been cancelled.
|
|
Close(ReadyState.Closed);
|
|
}
|
|
catch (Exception e)
|
|
{
|
|
_logger.Error(e.Message);
|
|
CancelCurrentRequest();
|
|
|
|
// If the user called Close(), ReadyState = Shutdown, so errors are irrelevant.
|
|
if (ReadyState != ReadyState.Shutdown)
|
|
{
|
|
Exception realException = e;
|
|
if (realException is AggregateException ae && ae.InnerExceptions.Count == 1)
|
|
{
|
|
realException = ae.InnerException;
|
|
}
|
|
if (realException is OperationCanceledException oe)
|
|
{
|
|
// This exception could either be the result of us explicitly cancelling a request, in which case we don't
|
|
// need to do anything else, or it could be that the request timed out.
|
|
if (oe.CancellationToken.IsCancellationRequested)
|
|
{
|
|
realException = null;
|
|
}
|
|
else
|
|
{
|
|
realException = new TimeoutException();
|
|
}
|
|
}
|
|
if (realException != null)
|
|
{
|
|
OnError(new ExceptionEventArgs(realException));
|
|
}
|
|
Close(ReadyState.Closed);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task MaybeWaitWithBackOff()
|
|
{
|
|
TimeSpan sleepTime = _backOff.GetNextBackOff();
|
|
if (sleepTime > MaxSleepTime)
|
|
{
|
|
// Task.Delay throws for anything above the platform timer ceiling, and this method
|
|
// runs outside the reconnect loop's exception handling, so an unclamped value
|
|
// would fault StartAsync and stop the stream permanently.
|
|
sleepTime = MaxSleepTime;
|
|
}
|
|
if (sleepTime > TimeSpan.Zero)
|
|
{
|
|
_logger.Info(string.Format("Waiting {0} milliseconds before reconnecting...", sleepTime.TotalMilliseconds));
|
|
BackOffDelay = sleepTime;
|
|
try
|
|
{
|
|
await Task.Delay(sleepTime, _shutdownTokenSource.Token);
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
// Cancellation happened during the wait, likely intentional close
|
|
_logger.Debug("Backoff wait interrupted by shutdown");
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public void Restart(bool resetBackoffDelay)
|
|
{
|
|
lock (this)
|
|
{
|
|
if (_readyState != ReadyState.Open)
|
|
{
|
|
return;
|
|
}
|
|
if (resetBackoffDelay)
|
|
{
|
|
_backOff.ResetBackoffN();
|
|
}
|
|
}
|
|
CancelCurrentRequest();
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public void SetTemporaryRetryDelayBounds(TimeSpan initialDelay, TimeSpan maxDelay)
|
|
{
|
|
_backOff.SetBounds(initialDelay, maxDelay);
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public void ClearTemporaryRetryDelayBounds()
|
|
{
|
|
_backOff.SetBounds(_configuredInitialRetryDelay, _configuredMaxRetryDelay);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Closes the connection to the SSE server. The <c>EventSource</c> cannot be reopened after this.
|
|
/// </summary>
|
|
public void Close()
|
|
{
|
|
if (ReadyState != ReadyState.Shutdown)
|
|
{
|
|
Close(ReadyState.Shutdown);
|
|
}
|
|
CancelCurrentRequest();
|
|
_shutdownTokenSource.Cancel();
|
|
|
|
// do not dispose httpClient if it is user provided
|
|
if (_configuration.HttpClient == null)
|
|
{
|
|
_httpClient.Dispose();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Equivalent to calling <see cref="Close()"/>.
|
|
/// </summary>
|
|
public void Dispose()
|
|
{
|
|
Dispose(true);
|
|
}
|
|
|
|
private void Dispose(bool disposing)
|
|
{
|
|
if (disposing)
|
|
{
|
|
Close();
|
|
}
|
|
}
|
|
|
|
#endregion
|
|
|
|
#region Private Methods
|
|
|
|
private HttpClient CreateHttpClient()
|
|
{
|
|
var client = _configuration.HttpMessageHandler is null ?
|
|
new HttpClient() :
|
|
new HttpClient(_configuration.HttpMessageHandler, false);
|
|
client.Timeout = _configuration.ResponseStartTimeout;
|
|
return client;
|
|
}
|
|
|
|
private void CancelCurrentRequest()
|
|
{
|
|
CancellationTokenSource requestTokenSource = null;
|
|
lock (this)
|
|
{
|
|
requestTokenSource = _currentRequestToken;
|
|
_currentRequestToken = null;
|
|
}
|
|
if (requestTokenSource != null)
|
|
{
|
|
_logger.Debug("Cancelling current request");
|
|
requestTokenSource.Cancel();
|
|
requestTokenSource.Dispose();
|
|
}
|
|
}
|
|
|
|
internal virtual EventSourceService GetEventSourceService(Configuration configuration)
|
|
{
|
|
return new EventSourceService(configuration, _httpClient, _logger);
|
|
}
|
|
|
|
private async Task ConnectToEventSourceAsync(CancellationToken cancellationToken)
|
|
{
|
|
_eventDataStringBuffer = null;
|
|
_eventDataUtf8ByteBuffer = null;
|
|
|
|
var svc = GetEventSourceService(_configuration);
|
|
|
|
svc.ConnectionOpened += (o, e) =>
|
|
{
|
|
_connectionTimer.Restart();
|
|
SetReadyState(ReadyState.Open, OnOpened, e.Headers);
|
|
};
|
|
svc.ConnectionClosed += (o, e) => { SetReadyState(ReadyState.Closed, OnClosed); };
|
|
|
|
await svc.GetDataAsync(
|
|
ProcessResponseLineString,
|
|
ProcessResponseLineUtf8,
|
|
_lastEventId,
|
|
cancellationToken
|
|
);
|
|
}
|
|
|
|
private void Close(ReadyState state)
|
|
{
|
|
_logger.Debug(string.Format("Close({0}) - state was {1}", state, ReadyState));
|
|
SetReadyState(state, OnClosed);
|
|
}
|
|
|
|
private void ProcessResponseLineString(string content)
|
|
{
|
|
if (content == null)
|
|
{
|
|
// StreamReader may emit a null if the stream has been closed; there's nothing to
|
|
// be done at this level in that case
|
|
return;
|
|
}
|
|
if (content.Length == 0)
|
|
{
|
|
DispatchEvent();
|
|
}
|
|
else
|
|
{
|
|
HandleParsedLine(EventParser.ParseLineString(content));
|
|
}
|
|
}
|
|
|
|
private void ProcessResponseLineUtf8(Utf8ByteSpan content)
|
|
{
|
|
if (content.Length == 0)
|
|
{
|
|
DispatchEvent();
|
|
}
|
|
else
|
|
{
|
|
HandleParsedLine(EventParser.ParseLineUtf8Bytes(content));
|
|
}
|
|
}
|
|
|
|
|
|
private void SetReadyState(ReadyState state, Action<StateChangedEventArgs> action = null,
|
|
IEnumerable<KeyValuePair<string, IEnumerable<string>>> headers = null)
|
|
{
|
|
lock (this)
|
|
{
|
|
if (_readyState == state || _readyState == ReadyState.Shutdown)
|
|
{
|
|
return;
|
|
}
|
|
|
|
_readyState = state;
|
|
}
|
|
|
|
if (action != null)
|
|
{
|
|
action(new StateChangedEventArgs(state, headers));
|
|
}
|
|
}
|
|
|
|
private void HandleParsedLine(EventParser.Result result)
|
|
{
|
|
if (result.IsComment)
|
|
{
|
|
OnCommentReceived(new CommentReceivedEventArgs(result.GetValueAsString()));
|
|
}
|
|
else if (result.IsDataField)
|
|
{
|
|
if (result.ValueString != null)
|
|
{
|
|
if (_eventDataStringBuffer is null)
|
|
{
|
|
_eventDataStringBuffer = new List<string>(2);
|
|
}
|
|
_eventDataStringBuffer.Add(result.ValueString);
|
|
_eventDataStringBuffer.Add("\n");
|
|
}
|
|
else
|
|
{
|
|
if (_eventDataUtf8ByteBuffer is null)
|
|
{
|
|
_eventDataUtf8ByteBuffer = new MemoryStream(result.ValueBytes.Length + 1);
|
|
}
|
|
_eventDataUtf8ByteBuffer.Write(result.ValueBytes.Data, result.ValueBytes.Offset, result.ValueBytes.Length);
|
|
_eventDataUtf8ByteBuffer.WriteByte((byte)'\n');
|
|
}
|
|
}
|
|
else if (result.IsEventField)
|
|
{
|
|
_eventName = result.GetValueAsString();
|
|
}
|
|
else if (result.IsIdField)
|
|
{
|
|
_lastEventId = result.GetValueAsString();
|
|
}
|
|
else if (result.IsRetryField)
|
|
{
|
|
if (long.TryParse(result.GetValueAsString(), out var retry))
|
|
{
|
|
// Clamp while the value is still an integer.
|
|
var millis = Math.Max(0,
|
|
Math.Min(retry, Constants.MaxServerDirectedRetryDelayMillis));
|
|
_backOff.SetServerDirectedMinDelay(TimeSpan.FromMilliseconds(millis));
|
|
}
|
|
}
|
|
}
|
|
|
|
private void DispatchEvent()
|
|
{
|
|
var name = _eventName ?? Constants.MessageField;
|
|
_eventName = null;
|
|
MessageEvent message;
|
|
|
|
if (_eventDataStringBuffer != null)
|
|
{
|
|
if (_eventDataStringBuffer.Count == 0)
|
|
{
|
|
return;
|
|
}
|
|
// remove last item which is always a trailing newline
|
|
_eventDataStringBuffer.RemoveAt(_eventDataStringBuffer.Count - 1);
|
|
var dataString = string.Concat(_eventDataStringBuffer);
|
|
message = new MessageEvent(name, dataString, _lastEventId, _configuration.Uri);
|
|
|
|
_eventDataStringBuffer.Clear();
|
|
}
|
|
else
|
|
{
|
|
if (_eventDataUtf8ByteBuffer is null || _eventDataUtf8ByteBuffer.Length == 0)
|
|
{
|
|
return;
|
|
}
|
|
var dataSpan = new Utf8ByteSpan(_eventDataUtf8ByteBuffer.GetBuffer(), 0,
|
|
(int)_eventDataUtf8ByteBuffer.Length - 1); // remove trailing newline
|
|
message = new MessageEvent(name, dataSpan, _lastEventId, _configuration.Uri);
|
|
|
|
// We've now taken ownership of the original buffer; null out the previous
|
|
// reference to it so a new one will be created next time
|
|
_eventDataUtf8ByteBuffer = null;
|
|
}
|
|
|
|
_logger.Debug(string.Format("Received event \"{0}\"", name));
|
|
OnMessageReceived(new MessageReceivedEventArgs(message));
|
|
}
|
|
|
|
private void OnOpened(StateChangedEventArgs e) =>
|
|
Opened?.Invoke(this, e);
|
|
|
|
private void OnClosed(StateChangedEventArgs e) =>
|
|
Closed?.Invoke(this, e);
|
|
|
|
private void OnMessageReceived(MessageReceivedEventArgs e) =>
|
|
MessageReceived?.Invoke(this, e);
|
|
|
|
private void OnCommentReceived(CommentReceivedEventArgs e) =>
|
|
CommentReceived?.Invoke(this, e);
|
|
|
|
private void OnError(ExceptionEventArgs e) =>
|
|
Error?.Invoke(this, e);
|
|
|
|
#endregion
|
|
}
|
|
}
|