diff --git a/example/SeqQuery/Program.cs b/example/SeqQuery/Program.cs index c80ac00..9a793a7 100644 --- a/example/SeqQuery/Program.cs +++ b/example/SeqQuery/Program.cs @@ -32,7 +32,7 @@ static void Main(string[] args) var server = arguments[""].ToString(); var query = Normalize(arguments[""]); var apiKey = Normalize(arguments["--apikey"]); - var @from = Normalize(arguments["--from"]); + var from = Normalize(arguments["--from"]); var to = Normalize(arguments["--to"]); await Run(server, apiKey, query, from, to); diff --git a/example/SeqTail/Program.cs b/example/SeqTail/Program.cs index 162fdd1..0d98f85 100644 --- a/example/SeqTail/Program.cs +++ b/example/SeqTail/Program.cs @@ -46,7 +46,7 @@ static void Main(string[] args) Console.WriteLine("Tailing, press Ctrl+C to exit."); Console.CancelKeyPress += (s,a) => cancel.Cancel(); - var run = Task.Run(() => Run(server, apiKey, filter, cancel)); + var run = Task.Run(() => Run(server, apiKey, filter, cancel), cancel.Token); run.GetAwaiter().GetResult(); } @@ -78,8 +78,8 @@ static async Task Run(string server, string apiKey, string filter, CancellationT using (var stream = await connection.Events.StreamAsync(filter: strict)) { var subscription = stream - .Select(jObject => LogEventReader.ReadFromJObject(jObject)) - .Subscribe(evt => Log.Write(evt), () => cancel.Cancel()); + .Select(LogEventReader.ReadFromJObject) + .Subscribe(Log.Write, cancel.Cancel); cancel.Token.WaitHandle.WaitOne(); subscription.Dispose(); diff --git a/seq-api.sln.DotSettings b/seq-api.sln.DotSettings index 0cbd2ce..969cc30 100644 --- a/seq-api.sln.DotSettings +++ b/seq-api.sln.DotSettings @@ -1,7 +1,11 @@ - + AD + True + True True True True + True + True True diff --git a/src/Seq.Api/Client/SeqApiClient.cs b/src/Seq.Api/Client/SeqApiClient.cs index 3afd2c7..e59174f 100644 --- a/src/Seq.Api/Client/SeqApiClient.cs +++ b/src/Seq.Api/Client/SeqApiClient.cs @@ -25,10 +25,9 @@ public class SeqApiClient : IDisposable // Future versions of Seq may not completely support v1 features, however // providing this as an Accept header will ensure what compatibility is available - // can be utilised. + // can be utilized. const string SeqApiV7MediaType = "application/vnd.datalust.seq.v7+json"; - readonly HttpClient _httpClient; readonly CookieContainer _cookies = new CookieContainer(); readonly JsonSerializer _serializer = JsonSerializer.Create( new JsonSerializerSettings @@ -53,12 +52,12 @@ public SeqApiClient(string serverUrl, string apiKey = null, bool useDefaultCrede if (!baseAddress.EndsWith("/")) baseAddress += "/"; - _httpClient = new HttpClient(handler) { BaseAddress = new Uri(baseAddress) }; + HttpClient = new HttpClient(handler) { BaseAddress = new Uri(baseAddress) }; } public string ServerUrl { get; } - public HttpClient HttpClient => _httpClient; + public HttpClient HttpClient { get; } public Task GetRootAsync(CancellationToken cancellationToken = default) { @@ -183,7 +182,7 @@ async Task HttpSendAsync(HttpRequestMessage request, CancellationToken c request.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue(SeqApiV7MediaType)); - var response = await _httpClient.SendAsync(request, cancellationToken).ConfigureAwait(false); + var response = await HttpClient.SendAsync(request, cancellationToken).ConfigureAwait(false); var stream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false); if (response.IsSuccessStatusCode) @@ -239,7 +238,7 @@ static string ResolveLink(ILinked entity, string link, IDictionary> _resourceGroups = new ConcurrentDictionary>(); - readonly Lazy> _root; + readonly object _sync = new object(); + readonly Dictionary> _resourceGroups = new Dictionary>(); + Task _root; public SeqConnection(string serverUrl, string apiKey = null, bool useDefaultCredentials = true) { if (serverUrl == null) throw new ArgumentNullException(nameof(serverUrl)); Client = new SeqApiClient(serverUrl, apiKey, useDefaultCredentials); - - _root = new Lazy>(() => Client.GetRootAsync()); } + + public SeqApiClient Client { get; } public ApiKeysResourceGroup ApiKeys => new ApiKeysResourceGroup(this); @@ -60,27 +62,73 @@ public SeqConnection(string serverUrl, string apiKey = null, bool useDefaultCred public WorkspacesResourceGroup Workspaces => new WorkspacesResourceGroup(this); - public Task LoadResourceGroupAsync(string name, CancellationToken cancellationToken = default) + public async Task EnsureConnected(TimeSpan timeout) { - // Initially, we want to put an incomplete task into the cache so that any concurrent attempts to load the - // same resource group will wait on the same pending call. - var cached = _resourceGroups.GetOrAdd(name, s => ResourceGroupFactory(s, cancellationToken)); - - if (!cached.IsFaulted && !cached.IsCanceled) - return cached; - - // If the cached task failed (ideally a rare situation), clobber it and return a new task (less worried about - // overlapping/concurrent calls on this path). - return _resourceGroups.AddOrUpdate(name, - s => ResourceGroupFactory(s, cancellationToken), - (s, _) => ResourceGroupFactory(s, cancellationToken)); + var started = DateTime.UtcNow; + // Fractional milliseconds are lost here, but that's fine. + var wait = TimeSpan.FromMilliseconds(Math.Min(100, timeout.TotalMilliseconds)); + var deadline = started.Add(timeout); + while (!await ConnectAsync(DateTime.UtcNow > deadline)) + { + await Task.Delay(wait); + } } - async Task ResourceGroupFactory(string name, CancellationToken cancellationToken = default) + async Task ConnectAsync(bool throwOnFailure) { - return await Client.GetAsync(await _root.Value, name + "Resources", cancellationToken: cancellationToken).ConfigureAwait(false); + HttpStatusCode statusCode; + + try + { + statusCode = (await Client.HttpClient.GetAsync("api")).StatusCode; + } + catch + { + if (throwOnFailure) + throw; + + return false; + } + + if (statusCode == HttpStatusCode.OK) + return true; + + if (!throwOnFailure) + return false; + + throw new WebException($"Could not connect to the Seq API endpoint: {(int)statusCode}/{statusCode}."); } - public SeqApiClient Client { get; } + public async Task LoadResourceGroupAsync(string name, CancellationToken cancellationToken = default) + { + Task loadRoot; + lock (_sync) + { + if (_root == null || _root.IsFaulted || _root.IsCanceled) + _root = Client.GetRootAsync(cancellationToken); + + loadRoot = _root; + } + + var rootEntity = await loadRoot.ConfigureAwait(false); + + Task loadGroup; + lock (_sync) + { + // ReSharper disable once InvertIf + if (!_resourceGroups.TryGetValue(name, out loadGroup) || loadGroup.IsFaulted || loadGroup.IsCanceled) + { + loadGroup = GetResourceGroup(rootEntity, name, cancellationToken); + _resourceGroups.Add(name, loadGroup); + } + } + + return await loadGroup.ConfigureAwait(false); + } + + async Task GetResourceGroup(RootEntity root, string name, CancellationToken cancellationToken = default) + { + return await Client.GetAsync(root, name + "Resources", cancellationToken: cancellationToken).ConfigureAwait(false); + } } }