using System;
using System.IO;
using System.Net.Sockets;
using System.Text;
using System.Net.Http;
using System.Net.Http.Headers;
using System.Reflection;
using System.Threading;
using System.Threading.Tasks;
using static LaunchDarkly.EventSource.AsyncHelpers;
namespace LaunchDarkly.EventSource
{
internal class EventSourceService
{
#region Private Fields
// Must be >= 4096 to bypass Xamarin.Android.Net.AndroidMessageHandler's
// BufferedStream-4096 small-count code path, which stalls SSE reads for
// ~60 s waiting for the next server keepalive. See SDK-2755.
private const int ReadBufferSize = 8192;
private readonly Configuration _configuration;
private readonly HttpClient _httpClient;
private const string UserAgentProduct = "DotNetClient";
internal static readonly string UserAgentVersion = ((AssemblyInformationalVersionAttribute)typeof(EventSource)
.GetTypeInfo()
.Assembly
.GetCustomAttribute(typeof(AssemblyInformationalVersionAttribute)))
.InformationalVersion;
#endregion
#region Public Events
///
/// Occurs when the connection to the EventSource API has been opened.
///
public event EventHandler ConnectionOpened;
///
/// Occurs when the connection to the EventSource API has been closed.
///
public event EventHandler ConnectionClosed;
#endregion
#region Constructors
internal EventSourceService(Configuration configuration, HttpClient httpClient)
{
_configuration = configuration ?? throw new ArgumentNullException(nameof(configuration));
_httpClient = httpClient ?? throw new ArgumentNullException(nameof(httpClient));
}
#endregion
#region Public Methods
///
/// Initiates the request to the EventSource API and parses Server Sent Events received by the API.
///
/// A A task that represents the work queued to execute in the ThreadPool.
public async Task GetDataAsync(
Action processResponseLineString,
Action processResponseLineUTF8,
string lastEventId,
CancellationToken cancellationToken
)
{
cancellationToken.ThrowIfCancellationRequested();
await ConnectToEventSourceApi(processResponseLineString, processResponseLineUTF8, lastEventId, cancellationToken);
}
#endregion
#region Private Methods
private async Task ConnectToEventSourceApi(
Action processResponseLineString,
Action processResponseLineUTF8,
string lastEventId,
CancellationToken cancellationToken
)
{
var request = CreateHttpRequestMessage(_configuration.Uri, lastEventId);
this._configuration.HttpRequestModifier?.Invoke(request);
using (var response = await _httpClient.SendAsync(request,
HttpCompletionOption.ResponseHeadersRead,
cancellationToken).ConfigureAwait(false))
{
ValidateResponse(response);
using (var stream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false))
{
var encoding = DetectEncoding(response);
if (encoding != Encoding.UTF8)
{
throw new EventSourceServiceCancelledException(
string.Format(Resources.ErrorWrongEncoding, encoding.HeaderName));
}
OnConnectionOpened(response.Headers);
if (_configuration.PreferDataAsUtf8Bytes)
{
await ProcessResponseFromUtf8StreamAsync(processResponseLineUTF8, stream, cancellationToken);
}
else
{
using (var reader = new StreamReader(stream, Encoding.UTF8,
detectEncodingFromByteOrderMarks: true, bufferSize: ReadBufferSize))
{
await ProcessResponseFromReaderAsync(processResponseLineString, reader, cancellationToken);
}
}
}
OnConnectionClosed();
}
}
private Encoding DetectEncoding(HttpResponseMessage response)
{
var charset = response.Content.Headers.ContentType?.CharSet;
if (charset != null)
{
try
{
return Encoding.GetEncoding(charset);
}
catch (ArgumentException) { }
}
return Encoding.UTF8;
}
protected virtual async Task ProcessResponseFromReaderAsync(
Action processResponse,
StreamReader reader,
CancellationToken cancellationToken
)
{
while (!cancellationToken.IsCancellationRequested)
{
var line = await DoWithTimeout(_configuration.ReadTimeout, cancellationToken,
token => AllowCancellation(reader.ReadLineAsync(), token));
if (line == null)
{
// this means the stream is done, i.e. the connection was closed
return;
}
processResponse(line);
}
}
protected async Task ProcessResponseFromUtf8StreamAsync(
Action processResponseLine,
Stream stream,
CancellationToken cancellationToken
)
{
var lineScanner = new ByteArrayLineScanner(ReadBufferSize);
while (!cancellationToken.IsCancellationRequested)
{
// Note that even though Stream.ReadAsync has an overload that takes a CancellationToken, that
// does not actually work for network sockets (https://stackoverflow.com/questions/12421989/networkstream-readasync-with-a-cancellation-token-never-cancels).
// So we must use AsyncHelpers.AllowCancellation to wrap it in a cancellable task.
int bytesRead = await DoWithTimeout(_configuration.ReadTimeout, cancellationToken,
token => AllowCancellation(stream.ReadAsync(lineScanner.Buffer, lineScanner.Count, lineScanner.Available), token));
if (bytesRead == 0)
{
cancellationToken.ThrowIfCancellationRequested();
return;
}
lineScanner.AddedBytes(bytesRead);
while (lineScanner.ScanToEndOfLine(out var lineSpan))
{
processResponseLine(lineSpan);
}
}
}
private HttpRequestMessage CreateHttpRequestMessage(Uri uri, string lastEventId)
{
var request = new HttpRequestMessage(_configuration.Method ?? HttpMethod.Get, uri);
// Add all headers provided in the Configuration Headers. This allows a consumer to provide any request headers to the EventSource API
if (_configuration.RequestHeaders != null)
{
foreach (var item in _configuration.RequestHeaders)
{
try
{
request.Headers.Add(item.Key, item.Value);
}
catch (FormatException)
{
// Avoid showing the Authorization header value in the exception message
if (item.Key.Equals("Authorization", StringComparison.OrdinalIgnoreCase))
{
throw new FormatException("The Authorization header is invalid.");
}
throw;
}
}
}
// Add the request body, if any.
if (_configuration.RequestBodyFactory != null)
{
HttpContent requestBody = _configuration.RequestBodyFactory();
if (requestBody != null)
{
request.Content = requestBody;
}
}
// If the lastEventId was provided, include it as a header to the API request.
if (!string.IsNullOrWhiteSpace(lastEventId))
{
request.Headers.Remove(Constants.LastEventIdHttpHeader);
request.Headers.Add(Constants.LastEventIdHttpHeader, lastEventId);
}
// If we haven't set the LastEventId header and if the EventSource Configuration was provided with a LastEventId,
// include it as a header to the API request.
if (!string.IsNullOrWhiteSpace(_configuration.LastEventId) && !request.Headers.Contains(Constants.LastEventIdHttpHeader))
request.Headers.Add(Constants.LastEventIdHttpHeader, _configuration.LastEventId);
if (request.Headers.UserAgent.Count == 0)
request.Headers.UserAgent.ParseAdd(UserAgentProduct + "/" + UserAgentVersion);
// Add the Accept Header if it wasn't provided in the Configuration
if (!request.Headers.Contains(Constants.AcceptHttpHeader))
request.Headers.Add(Constants.AcceptHttpHeader, Constants.EventStreamContentType);
request.Headers.ExpectContinue = false;
request.Headers.CacheControl = new CacheControlHeaderValue { NoCache = true };
return request;
}
private void ValidateResponse(HttpResponseMessage response)
{
// Any non-2xx response status is an error. A 204 (no content) is also an error.
if (!response.IsSuccessStatusCode || response.StatusCode == System.Net.HttpStatusCode.NoContent)
{
throw new EventSourceServiceUnsuccessfulResponseException((int)response.StatusCode, response.Headers);
}
if (response.Content == null)
{
throw new EventSourceServiceCancelledException(Resources.ErrorEmptyResponse);
}
if (response.Content.Headers.ContentType.MediaType != Constants.EventStreamContentType)
{
throw new EventSourceServiceCancelledException(
string.Format(Resources.ErrorWrongContentType, response.Content.Headers.ContentType));
}
}
private void OnConnectionOpened(HttpHeaders headers)
{
ConnectionOpened?.Invoke(this, new EventSourceOpenedEventArgs(headers));
}
private void OnConnectionClosed()
{
ConnectionClosed?.Invoke(this, EventArgs.Empty);
}
#endregion
}
}