2025.1 API changes to support WebSockets, query tracing, node connection state

Author: nblumhardt

example/SeqEnableAAD/SeqEnableAAD.csproj

@@ -2,7 +2,7 @@
 
   <PropertyGroup>
     <OutputType>Exe</OutputType>
-    <TargetFramework>net6.0</TargetFramework>
+    <TargetFramework>net9.0</TargetFramework>
     <AssemblyName>seq-enable-aad</AssemblyName>
     <Nullable>enable</Nullable>
   </PropertyGroup>

example/SeqQuery/SeqQuery.csproj

@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
-    <TargetFramework>net6.0</TargetFramework>
+    <TargetFramework>net9.0</TargetFramework>
     <OutputType>Exe</OutputType>
     <AssemblyName>seq-query</AssemblyName>
     <Nullable>enable</Nullable>

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["<server>"].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<JObject>(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);
+    }
 }

example/SeqTail/SeqTail.csproj

@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
-    <TargetFramework>net6.0</TargetFramework>
+    <TargetFramework>net9.0</TargetFramework>
     <AssemblyName>seq-tail</AssemblyName>
     <OutputType>Exe</OutputType>
     <Nullable>enable</Nullable>

example/SignalCopy/SignalCopy.csproj

@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
-    <TargetFramework>net6.0</TargetFramework>
+    <TargetFramework>net9.0</TargetFramework>
     <AssemblyName>signal-copy</AssemblyName>
     <OutputType>Exe</OutputType>
     <Nullable>enable</Nullable>

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
     /// </summary>
     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
             });
 
-        /// <summary>
-        /// Construct a <see cref="SeqApiClient"/>.
-        /// </summary>
-        /// <param name="serverUrl">The base URL of the Seq server.</param>
-        /// <param name="apiKey">An API key to use when making requests to the server, if required.</param>
-        /// <param name="useDefaultCredentials">Whether default credentials will be sent with HTTP requests; the default is <c>true</c>.</param>
-        [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)
-        {
-        }
-
-        /// <summary>
-        /// Construct a <see cref="SeqApiClient"/>.
-        /// </summary>
-        /// <param name="serverUrl">The base URL of the Seq server.</param>
-        /// <param name="apiKey">An API key to use when making requests to the server, if required.</param>
-        /// <param name="configureHttpClientHandler">An optional callback to configure the <see cref="HttpClientHandler"/> used when making HTTP requests
-        /// to the Seq API.</param>
-        [Obsolete("Prefer `SeqApiClient(serverUrl, apiKey, createHttpMessageHandler)` instead."), EditorBrowsable(EditorBrowsableState.Never)]
-        public SeqApiClient(string serverUrl, string apiKey, Action<HttpClientHandler> configureHttpClientHandler)
-            : this(serverUrl, apiKey, cookies =>
-            {
-                var handler = new HttpClientHandler { CookieContainer = cookies };
-                configureHttpClientHandler?.Invoke(handler);
-                return handler;
-            })
-        {
-        }
-
         /// <summary>
         /// Construct a <see cref="SeqApiClient"/>.
         /// </summary>
@@ -93,6 +64,8 @@ public SeqApiClient(string serverUrl, string apiKey, Action<HttpClientHandler> c
         /// to the Seq API. The callback receives a <see cref="CookieContainer"/> that is shared with WebSocket requests made by the client.</param>
         public SeqApiClient(string serverUrl, string apiKey = null, Func<CookieContainer, HttpMessageHandler> 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<CookieContainer
 
             ServerUrl = serverUrl ?? throw new ArgumentNullException(nameof(serverUrl));
 
-            if (!string.IsNullOrEmpty(apiKey))
-                _apiKey = apiKey;
-            
             var baseAddress = serverUrl;
             if (!baseAddress.EndsWith("/", StringComparison.Ordinal))
                 baseAddress += "/";
@@ -114,8 +84,8 @@ public SeqApiClient(string serverUrl, string apiKey = null, Func<CookieContainer
             HttpClient.BaseAddress = new Uri(baseAddress);
             HttpClient.DefaultRequestHeaders.Accept.Add(new MediaTypeWithQualityHeaderValue(SeqApiV11MediaType));
 
-            if (_apiKey != null)
-                HttpClient.DefaultRequestHeaders.Add("X-Seq-ApiKey", _apiKey);
+            if (!string.IsNullOrEmpty(apiKey))
+                HttpClient.DefaultRequestHeaders.Add("X-Seq-ApiKey", apiKey);
         }
 
         /// <summary>
@@ -323,9 +293,27 @@ public async Task<TResponse> DeleteAsync<TEntity, TResponse>(ILinked entity, str
         /// <param name="parameters">Named parameters to substitute into the link template, if required.</param>
         /// <param name="cancellationToken">A <see cref="CancellationToken"/> supporting cancellation.</param>
         /// <returns>A stream of values from the websocket.</returns>
-        public async Task<ObservableStream<TEntity>> StreamAsync<TEntity>(ILinked entity, string link, IDictionary<string, object> parameters = null, CancellationToken cancellationToken = default)
+        public IAsyncEnumerable<TEntity> StreamAsync<TEntity>(ILinked entity, string link, IDictionary<string, object> parameters = null, CancellationToken cancellationToken = default)
+        {
+            return WebSocketStreamAsync<NoMessage, TEntity>(entity, link, default, parameters, delegate { }, reader => _serializer.Deserialize<TEntity>(new JsonTextReader(reader)), cancellationToken);
+        }
+
+        /// <summary>
+        /// Connect to a websocket at the address specified by following <paramref name="link"/> from <paramref name="entity"/>.
+        /// When the WebSocket opens, a single message <paramref name="message"/> is sent, and then messages received on
+        /// socket are returned.
+        /// </summary>
+        /// <typeparam name="TEntity">The type of the values received over the websocket.</typeparam>
+        /// <typeparam name="TMessage">The type of message to send.</typeparam>
+        /// <param name="entity">An entity previously retrieved from the API.</param>
+        /// <param name="link">The name of the outbound link template present in <paramref name="entity"/>'s <see cref="ILinked.Links"/> collection.</param>
+        /// <param name="message">The message to send at establishment of the WebSocket connection.</param>
+        /// <param name="parameters">Named parameters to substitute into the link template, if required.</param>
+        /// <param name="cancellationToken">A <see cref="CancellationToken"/> supporting cancellation.</param>
+        /// <returns>A stream of values from the websocket.</returns>
+        public IAsyncEnumerable<TEntity> StreamSendAsync<TMessage, TEntity>(ILinked entity, string link, TMessage message, IDictionary<string, object> parameters = null, CancellationToken cancellationToken = default)
         {
-            return await WebSocketStreamAsync(entity, link, parameters, reader => _serializer.Deserialize<TEntity>(new JsonTextReader(reader)), cancellationToken);
+            return WebSocketStreamAsync(entity, link, message, parameters, (writer, m) => _serializer.Serialize(writer, m), reader => _serializer.Deserialize<TEntity>(new JsonTextReader(reader)), cancellationToken);
         }
 
         /// <summary>
@@ -336,23 +324,106 @@ public async Task<ObservableStream<TEntity>> StreamAsync<TEntity>(ILinked entity
         /// <param name="parameters">Named parameters to substitute into the link template, if required.</param>
         /// <param name="cancellationToken">A <see cref="CancellationToken"/> supporting cancellation.</param>
         /// <returns>A stream of raw messages from the websocket.</returns>
-        public async Task<ObservableStream<string>> StreamTextAsync(ILinked entity, string link, IDictionary<string, object> parameters = null, CancellationToken cancellationToken = default)
+        public IAsyncEnumerable<string> StreamTextAsync(ILinked entity, string link, IDictionary<string, object> parameters = null, CancellationToken cancellationToken = default)
+        {
+            return WebSocketStreamAsync<NoMessage, string>(entity, link, default, parameters, delegate { }, reader => reader.ReadToEnd(), cancellationToken);
+        }
+        
+        /// <summary>
+        /// Connect to a websocket at the address specified by following <paramref name="link"/> from <paramref name="entity"/>.
+        /// When the WebSocket opens, a single message <paramref name="message"/> is sent, and then messages received on
+        /// socket are returned.
+        /// </summary>
+        /// <typeparam name="TMessage">The type of message to send.</typeparam>
+        /// <param name="entity">An entity previously retrieved from the API.</param>
+        /// <param name="link">The name of the outbound link template present in <paramref name="entity"/>'s <see cref="ILinked.Links"/> collection.</param>
+        /// <param name="message">The message to send at establishment of the WebSocket connection.</param>
+        /// <param name="parameters">Named parameters to substitute into the link template, if required.</param>
+        /// <param name="cancellationToken">A <see cref="CancellationToken"/> supporting cancellation.</param>
+        /// <returns>A stream of raw messages from the websocket.</returns>
+        public IAsyncEnumerable<string> StreamTextSendAsync<TMessage>(ILinked entity, string link, TMessage message, IDictionary<string, object> 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<ObservableStream<T>> WebSocketStreamAsync<T>(ILinked entity, string link, IDictionary<string, object> parameters, Func<TextReader, T> deserialize, CancellationToken cancellationToken = default)
+        readonly struct NoMessage
+        {
+            // Type marker only.
+        }
+
+        async IAsyncEnumerable<T> WebSocketStreamAsync<TMessage, T>(ILinked entity, string link, TMessage message, IDictionary<string, object> parameters, Action<TextWriter, TMessage> serialize, Func<TextReader, T> 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<T>(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<byte>(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<byte>(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<T> HttpGetAsync<T>(string url, CancellationToken cancellationToken = default)