diff --git a/example/SeqEnableAAD/SeqEnableAAD.csproj b/example/SeqEnableAAD/SeqEnableAAD.csproj index 77cb863..c7dddb1 100644 --- a/example/SeqEnableAAD/SeqEnableAAD.csproj +++ b/example/SeqEnableAAD/SeqEnableAAD.csproj @@ -2,7 +2,7 @@ Exe - net6.0 + net9.0 seq-enable-aad enable diff --git a/example/SeqQuery/SeqQuery.csproj b/example/SeqQuery/SeqQuery.csproj index a63bcac..0b8eecc 100644 --- a/example/SeqQuery/SeqQuery.csproj +++ b/example/SeqQuery/SeqQuery.csproj @@ -1,7 +1,7 @@  - net6.0 + net9.0 Exe seq-query enable diff --git a/example/SeqTail/Program.cs b/example/SeqTail/Program.cs index 0122366..7aff26b 100644 --- a/example/SeqTail/Program.cs +++ b/example/SeqTail/Program.cs @@ -3,10 +3,10 @@ using DocoptNet; using Seq.Api; using Serilog; -using System.Reactive.Linq; using Serilog.Formatting.Compact.Reader; using System.Threading; -using Newtonsoft.Json.Linq; + +// ReSharper disable AccessToDisposedClosure const string usage = @"seq-tail: watch a Seq query from your console. @@ -29,17 +29,17 @@ try { - var arguments = new Docopt().Apply(usage, args, version: "Seq Tail 0.2", exit: true)!; + var arguments = new Docopt().Apply(usage, args, version: "Seq Tail 0.3", exit: true)!; var server = arguments[""].ToString(); var apiKey = Normalize(arguments["--apikey"]); var filter = Normalize(arguments["--filter"]); - var cancel = new CancellationTokenSource(); + using var cts = new CancellationTokenSource(); Console.WriteLine("Tailing, press Ctrl+C to exit."); - Console.CancelKeyPress += (_,_) => cancel.Cancel(); + Console.CancelKeyPress += (_,_) => cts.Cancel(); - var run = Task.Run(() => Run(server, apiKey, filter, cancel), cancel.Token); + var run = Task.Run(() => Run(server, apiKey, filter, cts.Token), cts.Token); run.GetAwaiter().GetResult(); } @@ -56,22 +56,20 @@ return string.IsNullOrWhiteSpace(s) ? null : s; } -static async Task Run(string server, string? apiKey, string? filter, CancellationTokenSource cancel) +static async Task Run(string server, string? apiKey, string? filter, CancellationToken cancel) { var connection = new SeqConnection(server, apiKey); - string? strict = null; + string? strictFilter = null; if (filter != null) { var converted = await connection.Expressions.ToStrictAsync(filter); - strict = converted.StrictExpression; + strictFilter = converted.StrictExpression; } - using var stream = await connection.Events.StreamAsync(filter: strict); - var subscription = stream - .Select(LogEventReader.ReadFromJObject) - .Subscribe(Log.Write, cancel.Cancel); - - cancel.Token.WaitHandle.WaitOne(); - subscription.Dispose(); + await foreach (var evt in connection.Events.StreamDocumentsAsync(filter: strictFilter, clef: true, cancellationToken: cancel)) + { + var logEvent = LogEventReader.ReadFromString(evt); + Log.Write(logEvent); + } } diff --git a/example/SeqTail/SeqTail.csproj b/example/SeqTail/SeqTail.csproj index 9bf2168..d84543d 100644 --- a/example/SeqTail/SeqTail.csproj +++ b/example/SeqTail/SeqTail.csproj @@ -1,7 +1,7 @@  - net6.0 + net9.0 seq-tail Exe enable diff --git a/example/SignalCopy/SignalCopy.csproj b/example/SignalCopy/SignalCopy.csproj index 019cb75..0f30840 100644 --- a/example/SignalCopy/SignalCopy.csproj +++ b/example/SignalCopy/SignalCopy.csproj @@ -1,7 +1,7 @@  - net6.0 + net9.0 signal-copy Exe enable diff --git a/src/Seq.Api/Client/SeqApiClient.cs b/src/Seq.Api/Client/SeqApiClient.cs index b2c4d1b..ac9a718 100644 --- a/src/Seq.Api/Client/SeqApiClient.cs +++ b/src/Seq.Api/Client/SeqApiClient.cs @@ -14,8 +14,8 @@ using System; using System.Collections.Generic; -using System.ComponentModel; using System.IO; +using System.Linq; using System.Net; using System.Net.Http; using System.Net.Http.Headers; @@ -27,9 +27,10 @@ using Seq.Api.Model.Root; using Seq.Api.Serialization; using System.Threading; -using Seq.Api.Streams; using System.Net.WebSockets; +using System.Runtime.CompilerServices; using Seq.Api.Model.Shared; +using Seq.Api.Streams; namespace Seq.Api.Client { @@ -38,13 +39,13 @@ namespace Seq.Api.Client /// public sealed class SeqApiClient : IDisposable { - readonly string _apiKey; - // Future versions of Seq may not completely support vN-1 features, however // providing this as an Accept header will ensure what compatibility is available // can be utilized. const string SeqApiV11MediaType = "application/vnd.datalust.seq.v11+json"; + // ReSharper disable once NotAccessedField.Local + readonly bool _defaultMessageHandler; readonly CookieContainer _cookies = new(); readonly JsonSerializer _serializer = JsonSerializer.Create( new JsonSerializerSettings @@ -54,36 +55,6 @@ public sealed class SeqApiClient : IDisposable FloatParseHandling = FloatParseHandling.Decimal }); - /// - /// Construct a . - /// - /// The base URL of the Seq server. - /// An API key to use when making requests to the server, if required. - /// Whether default credentials will be sent with HTTP requests; the default is true. - [Obsolete("Prefer `SeqApiClient(serverUrl, apiKey, createHttpMessageHandler)` instead."), EditorBrowsable(EditorBrowsableState.Never)] - public SeqApiClient(string serverUrl, string apiKey, bool useDefaultCredentials) - : this(serverUrl, apiKey, handler => handler.UseDefaultCredentials = useDefaultCredentials) - { - } - - /// - /// Construct a . - /// - /// The base URL of the Seq server. - /// An API key to use when making requests to the server, if required. - /// An optional callback to configure the used when making HTTP requests - /// to the Seq API. - [Obsolete("Prefer `SeqApiClient(serverUrl, apiKey, createHttpMessageHandler)` instead."), EditorBrowsable(EditorBrowsableState.Never)] - public SeqApiClient(string serverUrl, string apiKey, Action configureHttpClientHandler) - : this(serverUrl, apiKey, cookies => - { - var handler = new HttpClientHandler { CookieContainer = cookies }; - configureHttpClientHandler?.Invoke(handler); - return handler; - }) - { - } - /// /// Construct a . /// @@ -93,6 +64,8 @@ public SeqApiClient(string serverUrl, string apiKey, Action c /// to the Seq API. The callback receives a that is shared with WebSocket requests made by the client. public SeqApiClient(string serverUrl, string apiKey = null, Func createHttpMessageHandler = null) { + _defaultMessageHandler = createHttpMessageHandler == null; + // This is required for compatibility with the obsolete constructor, which we can remove sometime in 2024. var httpMessageHandler = createHttpMessageHandler?.Invoke(_cookies) ?? #if SOCKETS_HTTP_HANDLER @@ -103,9 +76,6 @@ public SeqApiClient(string serverUrl, string apiKey = null, Func @@ -323,9 +293,27 @@ public async Task DeleteAsync(ILinked entity, str /// Named parameters to substitute into the link template, if required. /// A supporting cancellation. /// A stream of values from the websocket. - public async Task> StreamAsync(ILinked entity, string link, IDictionary parameters = null, CancellationToken cancellationToken = default) + public IAsyncEnumerable StreamAsync(ILinked entity, string link, IDictionary parameters = null, CancellationToken cancellationToken = default) + { + return WebSocketStreamAsync(entity, link, default, parameters, delegate { }, reader => _serializer.Deserialize(new JsonTextReader(reader)), cancellationToken); + } + + /// + /// Connect to a websocket at the address specified by following from . + /// When the WebSocket opens, a single message is sent, and then messages received on + /// socket are returned. + /// + /// The type of the values received over the websocket. + /// The type of message to send. + /// An entity previously retrieved from the API. + /// The name of the outbound link template present in 's collection. + /// The message to send at establishment of the WebSocket connection. + /// Named parameters to substitute into the link template, if required. + /// A supporting cancellation. + /// A stream of values from the websocket. + public IAsyncEnumerable StreamSendAsync(ILinked entity, string link, TMessage message, IDictionary parameters = null, CancellationToken cancellationToken = default) { - return await WebSocketStreamAsync(entity, link, parameters, reader => _serializer.Deserialize(new JsonTextReader(reader)), cancellationToken); + return WebSocketStreamAsync(entity, link, message, parameters, (writer, m) => _serializer.Serialize(writer, m), reader => _serializer.Deserialize(new JsonTextReader(reader)), cancellationToken); } /// @@ -336,23 +324,106 @@ public async Task> StreamAsync(ILinked entity /// Named parameters to substitute into the link template, if required. /// A supporting cancellation. /// A stream of raw messages from the websocket. - public async Task> StreamTextAsync(ILinked entity, string link, IDictionary parameters = null, CancellationToken cancellationToken = default) + public IAsyncEnumerable StreamTextAsync(ILinked entity, string link, IDictionary parameters = null, CancellationToken cancellationToken = default) + { + return WebSocketStreamAsync(entity, link, default, parameters, delegate { }, reader => reader.ReadToEnd(), cancellationToken); + } + + /// + /// Connect to a websocket at the address specified by following from . + /// When the WebSocket opens, a single message is sent, and then messages received on + /// socket are returned. + /// + /// The type of message to send. + /// An entity previously retrieved from the API. + /// The name of the outbound link template present in 's collection. + /// The message to send at establishment of the WebSocket connection. + /// Named parameters to substitute into the link template, if required. + /// A supporting cancellation. + /// A stream of raw messages from the websocket. + public IAsyncEnumerable StreamTextSendAsync(ILinked entity, string link, TMessage message, IDictionary parameters = null, CancellationToken cancellationToken = default) { - return await WebSocketStreamAsync(entity, link, parameters, reader => reader.ReadToEnd(), cancellationToken); + return WebSocketStreamAsync(entity, link, message, parameters, delegate { }, reader => reader.ReadToEnd(), cancellationToken); } - async Task> WebSocketStreamAsync(ILinked entity, string link, IDictionary parameters, Func deserialize, CancellationToken cancellationToken = default) + readonly struct NoMessage + { + // Type marker only. + } + + async IAsyncEnumerable WebSocketStreamAsync(ILinked entity, string link, TMessage message, IDictionary parameters, Action serialize, Func deserialize, [EnumeratorCancellation] CancellationToken cancellationToken = default) { var linkUri = ResolveLink(entity, link, parameters); - var socket = new ClientWebSocket(); - socket.Options.Cookies = _cookies; - if (_apiKey != null) - socket.Options.SetRequestHeader("X-Seq-ApiKey", _apiKey); + using var socket = new ClientWebSocket(); + +#if WEBSOCKET_USE_HTTPCLIENT + if (_defaultMessageHandler) + { + await socket.ConnectAsync(new Uri(linkUri), HttpClient, cancellationToken).WithApiExceptions().ConfigureAwait(false); + } + else +#endif + { + socket.Options.Cookies = _cookies; - await socket.ConnectAsync(new Uri(linkUri), cancellationToken); + foreach (var header in HttpClient.DefaultRequestHeaders) + { + socket.Options.SetRequestHeader(header.Key, header.Value.FirstOrDefault()); + } - return new ObservableStream(socket, deserialize); + await socket.ConnectAsync(new Uri(linkUri), cancellationToken).WithApiExceptions() + .ConfigureAwait(false); + } + + var buffer = new byte[16 * 1024]; + var current = new MemoryStream(); + var encoding = new UTF8Encoding(false); + var reader = new StreamReader(current, encoding); + + if (message is not NoMessage) + { + var w = new StreamWriter(current, encoding); + serialize(w, message); + // ReSharper disable once MethodHasAsyncOverload + w.Flush(); + await socket.SendAsync(new ArraySegment(current.GetBuffer(), 0, (int)current.Length), + WebSocketMessageType.Text, true, cancellationToken).WithApiExceptions().ConfigureAwait(false); + current.Position = 0; + current.SetLength(0); + } + + while (socket.State == WebSocketState.Open) + { + cancellationToken.ThrowIfCancellationRequested(); + + var received = await socket.ReceiveAsync(new ArraySegment(buffer), cancellationToken).WithApiExceptions().ConfigureAwait(false); + if (received.MessageType == WebSocketMessageType.Close) + { + if (socket.State != WebSocketState.Closed) + await socket.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "", cancellationToken).WithApiExceptions().ConfigureAwait(false); + + if (received.CloseStatus != WebSocketCloseStatus.NormalClosure) + { + throw new SeqApiException(received.CloseStatusDescription ?? received.CloseStatus.ToString()!); + } + } + else + { + current.Write(buffer, 0, received.Count); + + if (received.EndOfMessage) + { + current.Position = 0; + var value = deserialize(reader); + + current.SetLength(0); + reader.DiscardBufferedData(); + + yield return value; + } + } + } } async Task HttpGetAsync(string url, CancellationToken cancellationToken = default) diff --git a/src/Seq.Api/Client/SeqApiException.cs b/src/Seq.Api/Client/SeqApiException.cs index 853a1b3..12842b9 100644 --- a/src/Seq.Api/Client/SeqApiException.cs +++ b/src/Seq.Api/Client/SeqApiException.cs @@ -15,27 +15,47 @@ using System; using System.Net; -namespace Seq.Api.Client +#nullable enable + +namespace Seq.Api.Client; + +/// +/// Thrown when an action cannot be performed. +/// +public class SeqApiException : Exception { /// - /// Thrown when an action cannot be performed. + /// Construct a with the given message and status code. /// - public class SeqApiException : Exception + /// A message describing the error. + public SeqApiException(string message) + : base(message) { - /// - /// Construct a with the given message and status code. - /// - /// A message describing the error. - /// The corresponding status code returned from Seq, if available. - public SeqApiException(string message, HttpStatusCode? statusCode) - : base(message) - { - StatusCode = statusCode; - } - - /// - /// The status code returned from Seq, if available. - /// - public HttpStatusCode? StatusCode { get; } } + + /// + /// Construct a with the given message and status code. + /// + /// A message describing the error. + /// The corresponding status code returned from Seq, if available. + public SeqApiException(string message, HttpStatusCode? statusCode) + : this(message) + { + StatusCode = statusCode; + } + + /// + /// Construct a with the given message and status code. + /// + /// A message describing the error. + /// The cause of the API failure. + public SeqApiException(string message, Exception innerException) + : base(message, innerException) + { + } + + /// + /// The status code returned from Seq, if available. + /// + public HttpStatusCode? StatusCode { get; } } \ No newline at end of file diff --git a/src/Seq.Api/Model/Cluster/ClusterNodeEntity.cs b/src/Seq.Api/Model/Cluster/ClusterNodeEntity.cs index c31e8d5..4604dd3 100644 --- a/src/Seq.Api/Model/Cluster/ClusterNodeEntity.cs +++ b/src/Seq.Api/Model/Cluster/ClusterNodeEntity.cs @@ -65,14 +65,13 @@ public class ClusterNodeEntity : Entity public int? RemainingActiveOps { get; set; } /// - /// An informational description of the node's current state, or null if no additional - /// information about the node is available. + /// Whether the node is currently leading the cluster. /// - public string StateDescription { get; set; } + public bool IsLeading { get; set; } /// - /// Whether the node is currently leading the cluster. + /// Whether the node has connected to the cluster network. /// - public bool IsLeading { get; set; } + public bool IsConnected { get; set; } } } diff --git a/src/Seq.Api/ResourceGroups/DataResourceGroup.cs b/src/Seq.Api/ResourceGroups/DataResourceGroup.cs index a2a7a70..df10b64 100644 --- a/src/Seq.Api/ResourceGroups/DataResourceGroup.cs +++ b/src/Seq.Api/ResourceGroups/DataResourceGroup.cs @@ -43,6 +43,7 @@ internal DataResourceGroup(ILoadResourceGroup connection) /// created but not saved, a signal from another server, or the modified representation of an entity already persisted. /// The query timeout; if not specified, the query will run until completion. /// Values for any free variables that appear in . + /// Enable detailed (server-side) query tracing. /// Token through which the operation can be cancelled. /// A structured result set. public async Task QueryAsync( @@ -53,9 +54,10 @@ public async Task QueryAsync( SignalEntity unsavedSignal = null, TimeSpan? timeout = null, Dictionary variables = null, + bool trace = false, CancellationToken cancellationToken = default) { - MakeParameters(query, rangeStartUtc, rangeEndUtc, signal, unsavedSignal, timeout, variables, out var body, out var parameters); + MakeParameters(query, rangeStartUtc, rangeEndUtc, signal, unsavedSignal, timeout, variables, trace, out var body, out var parameters); return await GroupPostAsync("Query", body, parameters, cancellationToken).ConfigureAwait(false); } @@ -71,6 +73,7 @@ public async Task QueryAsync( /// The query timeout; if not specified, the query will run until completion. /// Values for any free variables that appear in . /// Token through which the operation can be cancelled. + /// Enable detailed (server-side) query tracing. /// A CSV result set. public async Task QueryCsvAsync( string query, @@ -80,9 +83,10 @@ public async Task QueryCsvAsync( SignalEntity unsavedSignal = null, TimeSpan? timeout = null, Dictionary variables = null, + bool trace = false, CancellationToken cancellationToken = default) { - MakeParameters(query, rangeStartUtc, rangeEndUtc, signal, unsavedSignal, timeout, variables, out var body, out var parameters); + MakeParameters(query, rangeStartUtc, rangeEndUtc, signal, unsavedSignal, timeout, variables, trace, out var body, out var parameters); parameters.Add("format", "text/csv"); return await GroupPostReadStringAsync("Query", body, parameters, cancellationToken).ConfigureAwait(false); } @@ -95,6 +99,7 @@ static void MakeParameters( SignalEntity unsavedSignal, TimeSpan? timeout, Dictionary variables, + bool trace, out EvaluationContextPart body, out Dictionary parameters) { @@ -114,6 +119,9 @@ static void MakeParameters( if (timeout != null) parameters.Add("timeoutMS", timeout.Value.TotalMilliseconds.ToString("0")); + + if (trace) + parameters.Add("trace", true); body = new EvaluationContextPart { Signal = unsavedSignal, Variables = variables }; } diff --git a/src/Seq.Api/ResourceGroups/EventsResourceGroup.cs b/src/Seq.Api/ResourceGroups/EventsResourceGroup.cs index 4f51004..514e552 100644 --- a/src/Seq.Api/ResourceGroups/EventsResourceGroup.cs +++ b/src/Seq.Api/ResourceGroups/EventsResourceGroup.cs @@ -20,7 +20,6 @@ using Seq.Api.Model.Events; using Seq.Api.Model.Shared; using Seq.Api.Model.Signals; -using Seq.Api.Streams; namespace Seq.Api.ResourceGroups { @@ -42,6 +41,7 @@ internal EventsResourceGroup(ILoadResourceGroup connection) /// with versions prior to 5.0, which did not mark permalinks explicitly. /// Run the search at a lower priority, using a lower proportion of available server /// resources (may take longer to complete). + /// Enable detailed (server-side) query tracing. /// Token through which the operation can be cancelled. /// The event. public async Task FindAsync( @@ -49,6 +49,7 @@ public async Task FindAsync( bool render = false, string permalinkId = null, bool background = false, + bool trace = false, CancellationToken cancellationToken = default) { if (id == null) throw new ArgumentNullException(nameof(id)); @@ -57,6 +58,7 @@ public async Task FindAsync( if (render) parameters.Add("render", true); if (permalinkId != null) parameters.Add("permalinkId", permalinkId); if (background) parameters.Add("background", true); + if (trace) parameters.Add("trace", true); return await GroupGetAsync("Item", parameters, cancellationToken).ConfigureAwait(false); } @@ -75,6 +77,77 @@ public async Task FindAsync( /// If specified, the event's message template and properties will be rendered into its RenderedMessage property. /// Earliest (inclusive) date/time from which to search. /// Latest (exclusive) date/time from which to search. + /// If the request is for a permalinked event, specifying the id of the permalink here will + /// allow events that have otherwise been deleted to be found. The special value `"unknown"` provides backwards compatibility + /// with versions prior to 5.0, which did not mark permalinks explicitly. + /// Values for any free variables that appear in . + /// Run the search at a lower priority, using a lower proportion of available server + /// resources (may take longer to complete). + /// Token through which the operation can be cancelled. + /// Enable detailed (server-side) query tracing. + /// The complete list of events, ordered from least to most recent. + public async IAsyncEnumerable EnumerateAsync( + SignalEntity unsavedSignal = null, + SignalExpressionPart signal = null, + string filter = null, + int count = 30, + string startAtId = null, + string afterId = null, + bool render = false, + DateTime? fromDateUtc = null, + DateTime? toDateUtc = null, + string permalinkId = null, + Dictionary variables = null, + bool background = false, + bool trace = false, + [EnumeratorCancellation] + CancellationToken cancellationToken = default) + { + var parameters = new Dictionary{{ "count", count }}; + if (signal != null) { parameters.Add("signal", signal.ToString()); } + if (filter != null) { parameters.Add("filter", filter); } + if (startAtId != null) { parameters.Add("startAtId", startAtId); } + if (afterId != null) { parameters.Add("afterId", afterId); } + if (render) { parameters.Add("render", true); } + if (fromDateUtc != null) { parameters.Add("fromDateUtc", fromDateUtc.Value); } + if (toDateUtc != null) { parameters.Add("toDateUtc", toDateUtc.Value); } + if (permalinkId != null) { parameters.Add("permalinkId", permalinkId); } + if (background) parameters.Add("background", true); + if (trace) parameters.Add("trace", true); + + var group = await LoadGroupAsync(cancellationToken).ConfigureAwait(false); + if (variables == null && unsavedSignal == null) + { + await foreach (var evt in Client.StreamAsync(group, "Scan", parameters, cancellationToken)) + { + yield return evt; + } + yield break; + } + + var body = new EvaluationContextPart { Signal = unsavedSignal, Variables = variables }; + parameters.Add("wait", true); + await foreach (var evt in Client.StreamSendAsync(group, "Scan", body, parameters, cancellationToken)) + { + yield return evt; + } + } + + /// + /// Retrieve all events that match a set of conditions, using the paged search API rather than WebSockets. Note that this method is less + /// efficient and requires more server resources than the . + /// + /// A constructed signal that may not appear on the server, for example, a that has been + /// created but not saved, a signal from another server, or the modified representation of an entity already persisted. + /// If provided, a signal expression describing the set of events that will be filtered for the result. + /// A strict Seq filter expression to match (text expressions must be in double quotes). To + /// convert a "fuzzy" filter into a strict one the way the Seq UI does, use connection.Expressions.ToStrictAsync(). + /// The number of events to retrieve. If not specified will default to 30. + /// An event id from which to start searching (inclusively). + /// An event id to search after (exclusively). + /// If specified, the event's message template and properties will be rendered into its RenderedMessage property. + /// Earliest (inclusive) date/time from which to search. + /// Latest (exclusive) date/time from which to search. /// If specified, the number of events after the first match to keep searching before a partial /// result set is returned. Used to improve responsiveness when the result is displayed in a user interface, not typically used in /// batch processing scenarios. @@ -84,9 +157,10 @@ public async Task FindAsync( /// Values for any free variables that appear in . /// Run the search at a lower priority, using a lower proportion of available server /// resources (may take longer to complete). + /// Enable detailed (server-side) query tracing. /// Token through which the operation can be cancelled. /// The complete list of events, ordered from least to most recent. - public async IAsyncEnumerable EnumerateAsync( + public async IAsyncEnumerable PagedEnumerateAsync( SignalEntity unsavedSignal = null, SignalExpressionPart signal = null, string filter = null, @@ -100,6 +174,7 @@ public async IAsyncEnumerable EnumerateAsync( string permalinkId = null, Dictionary variables = null, bool background = false, + bool trace = false, [EnumeratorCancellation] CancellationToken cancellationToken = default) { @@ -114,7 +189,7 @@ public async IAsyncEnumerable EnumerateAsync( while (true) { var resultSet = await PageAsync(unsavedSignal, signal, filter, nextCount, startAtId, nextAfterId, render, - fromDateUtc, toDateUtc, shortCircuitAfter, permalinkId, variables, background, cancellationToken).ConfigureAwait(false); + fromDateUtc, toDateUtc, shortCircuitAfter, permalinkId, variables, background, trace, cancellationToken).ConfigureAwait(false); foreach (var evt in resultSet.Events) { @@ -135,7 +210,6 @@ public async IAsyncEnumerable EnumerateAsync( nextCount = Math.Min(remaining, pageSize); } } - /// /// Retrieve all events that match a set of conditions. The complete result is buffered into memory, @@ -153,15 +227,13 @@ public async IAsyncEnumerable EnumerateAsync( /// If specified, the event's message template and properties will be rendered into its RenderedMessage property. /// Earliest (inclusive) date/time from which to search. /// Latest (exclusive) date/time from which to search. - /// If specified, the number of events after the first match to keep searching before a partial - /// result set is returned. Used to improve responsiveness when the result is displayed in a user interface, not typically used in - /// batch processing scenarios. /// If the request is for a permalinked event, specifying the id of the permalink here will /// allow events that have otherwise been deleted to be found. The special value `"unknown"` provides backwards compatibility /// with versions prior to 5.0, which did not mark permalinks explicitly. /// Values for any free variables that appear in . /// Run the search at a lower priority, using a lower proportion of available server /// resources (may take longer to complete). + /// Enable detailed (server-side) query tracing. /// Token through which the operation can be cancelled. /// The result set with a page of events. public async Task> ListAsync( @@ -174,16 +246,15 @@ public async Task> ListAsync( bool render = false, DateTime? fromDateUtc = null, DateTime? toDateUtc = null, - int? shortCircuitAfter = null, string permalinkId = null, Dictionary variables = null, bool background = false, + bool trace = false, CancellationToken cancellationToken = default) { var results = new List(); await foreach (var item in EnumerateAsync(unsavedSignal, signal, filter, count, startAtId, afterId, render, - fromDateUtc, toDateUtc, shortCircuitAfter, permalinkId, variables, background, cancellationToken) - .WithCancellation(cancellationToken) + fromDateUtc, toDateUtc, permalinkId, variables, background, trace, cancellationToken) .ConfigureAwait(false)) results.Add(item); return results; @@ -213,6 +284,7 @@ public async Task> ListAsync( /// Run the search at a lower priority, using a lower proportion of available server /// resources (may take longer to complete). /// Token through which the operation can be cancelled. + /// Enable detailed (server-side) query tracing. /// The result set with a page of events. public async Task PageAsync( SignalEntity unsavedSignal = null, @@ -228,6 +300,7 @@ public async Task PageAsync( string permalinkId = null, Dictionary variables = null, bool background = false, + bool trace = false, CancellationToken cancellationToken = default) { var parameters = new Dictionary{{ "count", count }}; @@ -241,6 +314,7 @@ public async Task PageAsync( if (shortCircuitAfter != null) { parameters.Add("shortCircuitAfter", shortCircuitAfter.Value); } if (permalinkId != null) { parameters.Add("permalinkId", permalinkId); } if (background) parameters.Add("background", true); + if (trace) parameters.Add("trace", true); var body = new EvaluationContextPart { Signal = unsavedSignal, Variables = variables }; return await GroupPostAsync("InSignal", body, parameters, cancellationToken).ConfigureAwait(false); @@ -279,52 +353,96 @@ public async Task DeleteAsync( } /// - /// Connect to the live event stream, read as strongly-typed objects. Dispose the resulting stream to disconnect. + /// Connect to the live event stream. Cancel or dispose the resulting iterator to disconnect. /// - /// The type into which events should be deserialized. + /// A constructed signal that may not appear on the server, for example, a that has been + /// created but not saved, a signal from another server, or the modified representation of an entity already persisted. /// If provided, a signal expression describing the set of events that will be filtered for the result. /// A strict Seq filter expression to match (text expressions must be in double quotes). To /// convert a "fuzzy" filter into a strict one the way the Seq UI does, use connection.Expressions.ToStrictAsync(). + /// If specified, the event's message template and properties will be rendered into its property. + /// Values for any free variables that appear in . /// Token through which the operation can be cancelled. - /// An observable that will stream events from the server to subscribers. Events will be buffered server-side until the first - /// subscriber connects, ensure at least one subscription is made in order to avoid event loss. /// See the Seq ingestion /// docs for event schema information. - public async Task> StreamAsync( + public async IAsyncEnumerable StreamAsync( + SignalEntity unsavedSignal = null, SignalExpressionPart signal = null, string filter = null, - CancellationToken cancellationToken = default) + Dictionary variables = null, + bool render = false, + [EnumeratorCancellation] CancellationToken cancellationToken = default) { var parameters = new Dictionary(); if (signal != null) { parameters.Add("signal", signal.ToString()); } if (filter != null) { parameters.Add("filter", filter); } + if (render) { parameters.Add("render", true);} var group = await LoadGroupAsync(cancellationToken).ConfigureAwait(false); - return await Client.StreamAsync(group, "Stream", parameters, cancellationToken).ConfigureAwait(false); + + if (unsavedSignal != null) + { + parameters.Add("wait", true); + var body = new EvaluationContextPart { Signal = unsavedSignal, Variables = variables }; + await foreach (var evt in Client.StreamSendAsync(group, "Stream", body, parameters, + cancellationToken)) + { + yield return evt; + } + } + + await foreach (var evt in Client.StreamAsync(group, "Stream", parameters, cancellationToken)) + { + yield return evt; + } } /// - /// Connect to the live event stream, read as raw JSON documents. Dispose the resulting stream to disconnect. + /// Connect to the live event stream, returning events as raw JSON documents. Cancel or dispose the + /// resulting iterator to disconnect. /// + /// A constructed signal that may not appear on the server, for example, a that has been + /// created but not saved, a signal from another server, or the modified representation of an entity already persisted. /// If provided, a signal expression describing the set of events that will be filtered for the result. /// A strict Seq filter expression to match (text expressions must be in double quotes). To /// convert a "fuzzy" filter into a strict one the way the Seq UI does, use connection.Expressions.ToStrictAsync(). + /// If specified, the event's message template and properties will be rendered into its property. + /// If specified, compact JSON format will be requested instead of the default format. + /// Values for any free variables that appear in . /// Token through which the operation can be cancelled. - /// An observable that will stream events from the server to subscribers. Events will be buffered server-side until the first - /// subscriber connects, ensure at least one subscription is made in order to avoid event loss. /// See the Seq ingestion /// docs for event schema information. - public async Task> StreamDocumentsAsync( + public async IAsyncEnumerable StreamDocumentsAsync( + SignalEntity unsavedSignal = null, SignalExpressionPart signal = null, string filter = null, - CancellationToken cancellationToken = default) + Dictionary variables = null, + bool render = false, + bool clef = false, + [EnumeratorCancellation] CancellationToken cancellationToken = default) { var parameters = new Dictionary(); if (signal != null) { parameters.Add("signal", signal.ToString()); } if (filter != null) { parameters.Add("filter", filter); } + if (render) { parameters.Add("render", true);} + if (clef) { parameters.Add("clef", true);} var group = await LoadGroupAsync(cancellationToken).ConfigureAwait(false); - return await Client.StreamTextAsync(group, "Stream", parameters, cancellationToken).ConfigureAwait(false); + + if (unsavedSignal != null) + { + parameters.Add("wait", true); + var body = new EvaluationContextPart { Signal = unsavedSignal, Variables = variables }; + await foreach (var evt in Client.StreamTextSendAsync(group, "Stream", body, parameters, cancellationToken)) + { + yield return evt; + } + } + + await foreach (var evt in Client.StreamTextAsync(group, "Stream", parameters, cancellationToken)) + { + yield return evt; + } } } } diff --git a/src/Seq.Api/ResourceGroups/UsersResourceGroup.cs b/src/Seq.Api/ResourceGroups/UsersResourceGroup.cs index 99b83a1..f555f3a 100644 --- a/src/Seq.Api/ResourceGroups/UsersResourceGroup.cs +++ b/src/Seq.Api/ResourceGroups/UsersResourceGroup.cs @@ -169,7 +169,7 @@ public async Task LoginWindowsIntegratedAsync(CancellationToken canc var providers = await GroupGetAsync("AuthenticationProviders", cancellationToken: cancellationToken).ConfigureAwait(false); var provider = providers.Providers.SingleOrDefault(p => p.Name == "Integrated Windows Authentication"); if (provider == null) - throw new SeqApiException("The Integrated Windows Authentication provider is not available.", null); + throw new SeqApiException("The Integrated Windows Authentication provider is not available."); var response = await Client.HttpClient.GetAsync(provider.Url, cancellationToken).ConfigureAwait(false); response.EnsureSuccessStatusCode(); return await FindCurrentAsync(cancellationToken).ConfigureAwait(false); diff --git a/src/Seq.Api/Seq.Api.csproj b/src/Seq.Api/Seq.Api.csproj index a0a4021..2fca167 100644 --- a/src/Seq.Api/Seq.Api.csproj +++ b/src/Seq.Api/Seq.Api.csproj @@ -3,7 +3,7 @@ Client library for the Seq HTTP API. 2025.1.0 Datalust;Contributors - netstandard2.0;net6.0 + netstandard2.0;net6.0;net8.0 true true seq @@ -11,20 +11,24 @@ seq-api-icon.png https://github.com/datalust/seq-api Apache-2.0 - 9 + latest - + $(DefineConstants);SOCKETS_HTTP_HANDLER - + + + $(DefineConstants);SOCKETS_HTTP_HANDLER;WEBSOCKET_USE_HTTPCLIENT + + - + runtime; build; native; contentfiles; analyzers; buildtransitive diff --git a/src/Seq.Api/SeqConnection.cs b/src/Seq.Api/SeqConnection.cs index de64a6a..9fc90f7 100644 --- a/src/Seq.Api/SeqConnection.cs +++ b/src/Seq.Api/SeqConnection.cs @@ -14,7 +14,6 @@ using System; using System.Collections.Generic; -using System.ComponentModel; using System.Net; using System.Net.Http; using System.Threading; @@ -37,33 +36,6 @@ public class SeqConnection : ILoadResourceGroup, IDisposable readonly Dictionary> _resourceGroups = new(); Task _root; - /// - /// Construct a . - /// - /// The base URL of the Seq server. - /// An API key to use when making requests to the server, if required. - /// Whether default credentials will be sent with HTTP requests; the default is true. - [Obsolete("Prefer `SeqConnection(serverUrl, apiKey, createHttpMessageHandler)` instead."), EditorBrowsable(EditorBrowsableState.Never)] - public SeqConnection(string serverUrl, string apiKey, bool useDefaultCredentials) - { - if (serverUrl == null) throw new ArgumentNullException(nameof(serverUrl)); - Client = new SeqApiClient(serverUrl, apiKey, useDefaultCredentials); - } - - /// - /// Construct a . - /// - /// The base URL of the Seq server. - /// An API key to use when making requests to the server, if required. - /// An optional callback to configure the used when making HTTP requests - /// to the Seq API. - [Obsolete("Prefer `SeqConnection(serverUrl, apiKey, createHttpMessageHandler)` instead."), EditorBrowsable(EditorBrowsableState.Never)] - public SeqConnection(string serverUrl, string apiKey, Action configureHttpClientHandler) - { - if (serverUrl == null) throw new ArgumentNullException(nameof(serverUrl)); - Client = new SeqApiClient(serverUrl, apiKey, configureHttpClientHandler); - } - /// /// Construct a . /// diff --git a/src/Seq.Api/Streams/ObservableStream.Unsubscriber.cs b/src/Seq.Api/Streams/ObservableStream.Unsubscriber.cs deleted file mode 100644 index ffded68..0000000 --- a/src/Seq.Api/Streams/ObservableStream.Unsubscriber.cs +++ /dev/null @@ -1,39 +0,0 @@ -// Copyright © Datalust; based on code from -// Serilog.Sinks.Observable, Copyright 2013-2016 Serilog Contributors -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -using System; - -namespace Seq.Api.Streams -{ - public partial class ObservableStream - { - sealed class Unsubscriber : IDisposable - { - readonly ObservableStream _stream; - readonly IObserver _observer; - - public Unsubscriber(ObservableStream sink, IObserver observer) - { - _stream = sink ?? throw new ArgumentNullException(nameof(sink)); - _observer = observer ?? throw new ArgumentNullException(nameof(observer)); - } - - public void Dispose() - { - _stream.Unsubscribe(_observer); - } - } - } -} diff --git a/src/Seq.Api/Streams/ObservableStream.cs b/src/Seq.Api/Streams/ObservableStream.cs deleted file mode 100644 index 00a735e..0000000 --- a/src/Seq.Api/Streams/ObservableStream.cs +++ /dev/null @@ -1,230 +0,0 @@ -// Copyright © Datalust; based on code from -// Serilog.Sinks.Observable, Copyright 2013-2016 Serilog Contributors -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -using System; -using System.Collections.Generic; -using System.Threading; -using System.Net.WebSockets; -using System.Threading.Tasks; -using System.IO; -using System.Text; -using System.Linq; - -namespace Seq.Api.Streams -{ - /// - /// A stream of values read from a . - /// - /// The type of the values. - public sealed partial class ObservableStream : IObservable, IDisposable - { - readonly object _syncRoot = new object(); - IList> _observers = new List>(); - bool _ended, _disposed; - Task _run; - - readonly ClientWebSocket _socket; - readonly Func _deserialize; - - internal ObservableStream(ClientWebSocket socket, Func deserialize) - { - _deserialize = deserialize ?? throw new ArgumentNullException(nameof(deserialize)); - _socket = socket ?? throw new ArgumentNullException(nameof(socket)); - } - - /// - public IDisposable Subscribe(IObserver observer) - { - if (observer == null) throw new ArgumentNullException(nameof(observer)); - - lock (_syncRoot) - { - if (_disposed) - throw new ObjectDisposedException("The observable stream is disposed."); - - if (_ended) - { - observer.OnCompleted(); - return new Unsubscriber(this, observer); - } - - _observers = _observers.Concat(new[] { observer }).ToList(); - - if (_run == null) - _run = Task.Run(Receive); - } - - return new Unsubscriber(this, observer); - } - - void Unsubscribe(IObserver observer) - { - if (observer == null) throw new ArgumentNullException(nameof(observer)); - - lock (_syncRoot) - { - if (_disposed) - throw new ObjectDisposedException("The observable stream is disposed."); - - _observers = _observers.Except(new[] { observer }).ToList(); - } - } - - async Task Receive() - { - var buffer = new byte[16 * 1024]; - var current = new MemoryStream(); - var reader = new StreamReader(current, new UTF8Encoding(false)); - - while (_socket.State == WebSocketState.Open) - { - var received = await _socket.ReceiveAsync(new ArraySegment(buffer), CancellationToken.None); - if (received.MessageType == WebSocketMessageType.Close) - { - // Managed web sockets are self-closing - if (_socket.State != WebSocketState.Closed) - { - await _socket.CloseAsync(WebSocketCloseStatus.NormalClosure, "", CancellationToken.None); - } - } - else - { - current.Write(buffer, 0, received.Count); - - if (received.EndOfMessage) - { - current.Position = 0; - var value = _deserialize(reader); - - current.SetLength(0); - reader.DiscardBufferedData(); - - Emit(value); - } - } - } - - End(); - } - - void Emit(T value) - { - if (value == null) throw new ArgumentNullException(nameof(value)); - - IList> observers; - lock (_syncRoot) - { - if (_ended || _disposed) - return; - - observers = _observers; - } - - IList exceptions = null; - foreach (var observer in observers) - { - try - { - observer.OnNext(value); - } - catch (Exception ex) - { - if (exceptions == null) - exceptions = new List(); - exceptions.Add(ex); - } - } - - if (exceptions != null) - OnError(exceptions); - } - - void End() - { - IList> observers; - lock (_syncRoot) - { - if (_ended) - return; - - observers = _observers; - _observers = new List>(); - _ended = true; - } - - IList exceptions = null; - foreach (var observer in observers) - { - try - { - observer.OnCompleted(); - } - catch (Exception ex) - { - if (exceptions == null) - exceptions = new List(); - exceptions.Add(ex); - } - } - - if (exceptions != null) - OnError(exceptions); - } - - static void OnError(IEnumerable exceptions) - { - // This will hit TaskScheduler.UnobservedTaskException - throw new AggregateException("At least one observer failed to accept the event", exceptions); - } - - /// - public void Dispose() - { - lock (_syncRoot) - { - if (_disposed) return; - _disposed = true; - } - - try - { - if (_socket.State == WebSocketState.Open) - { - using var timeout = new CancellationTokenSource(); - timeout.CancelAfter(30000); - _socket.CloseAsync(WebSocketCloseStatus.NormalClosure, "Close requested", timeout.Token) - .ConfigureAwait(false).GetAwaiter().GetResult(); - } - } - catch - { - // Don't mask exceptions by throwing here during unwinding - } - - if (_run != null) - { - try - { - _run.ConfigureAwait(false).GetAwaiter().GetResult(); - } - catch (TaskCanceledException) - { - } - } - - _socket.Dispose(); - } - } -} diff --git a/src/Seq.Api/Streams/WebSocketTaskExtensions.cs b/src/Seq.Api/Streams/WebSocketTaskExtensions.cs new file mode 100644 index 0000000..605e415 --- /dev/null +++ b/src/Seq.Api/Streams/WebSocketTaskExtensions.cs @@ -0,0 +1,33 @@ +using System; +using System.Net.WebSockets; +using System.Threading.Tasks; +using Seq.Api.Client; + +namespace Seq.Api.Streams; + +// Async enumerators and try/catch blocks don't play nicely together. These extensions help maintain the +// previous exception contract around API calls. +static class WebSocketTaskExtensions +{ + public static async Task WithApiExceptions(this Task task) + { + await ((Task)task).WithApiExceptions(); + return task.Result; + } + + public static async Task WithApiExceptions(this Task task) + { + try + { + await task; + } + catch (WebSocketException ex) + { + throw new SeqApiException($"The WebSocket API call failed ({ex.ErrorCode}/{ex.WebSocketErrorCode}).", ex); + } + catch (Exception ex) + { + throw new SeqApiException("The API call failed.", ex); + } + } +} \ No newline at end of file diff --git a/test/Seq.Api.Tests/SeqConnectionTests.cs b/test/Seq.Api.Tests/SeqConnectionTests.cs index cbaa9bf..5e42597 100644 --- a/test/Seq.Api.Tests/SeqConnectionTests.cs +++ b/test/Seq.Api.Tests/SeqConnectionTests.cs @@ -1,4 +1,5 @@ -using Xunit; +using System.Net.Http; +using Xunit; namespace Seq.Api.Tests { @@ -9,12 +10,14 @@ public void WhenConstructedTheHandlerConfigurationCallbackIsCalled() { var callCount = 0; -#pragma warning disable CS0618 // Type or member is obsolete - using var _ = new SeqConnection("https://test.example.com", null, handler => { - Assert.NotNull(handler); - ++callCount; - }); -#pragma warning restore CS0618 // Type or member is obsolete + using var _ = new SeqConnection( + "https://test.example.com", + apiKey: null, + createHttpMessageHandler: cookies => + { + ++callCount; + return new HttpClientHandler { CookieContainer = cookies }; + }); Assert.Equal(1, callCount); }