WebSocket support for streaming events

example/SeqTail/Program.cs

@@ -1,11 +1,13 @@
 using System;
-using System.Collections.Generic;
 using System.Linq;
-using System.Threading;
 using System.Threading.Tasks;
 using DocoptNet;
 using Seq.Api;
-using Seq.Api.Model.Events;
+using Serilog;
+using System.Reactive.Linq;
+using Serilog.Formatting.Compact.Reader;
+using System.Threading;
+using Newtonsoft.Json.Linq;
 
 namespace SeqTail
 {
@@ -27,36 +29,32 @@ class Program
 
         static void Main(string[] args)
         {
-            var cts = new CancellationTokenSource();
+            Log.Logger = new LoggerConfiguration()
+                .MinimumLevel.Verbose()
+                .WriteTo.LiterateConsole()
+                .CreateLogger();
 
-            // ReSharper disable once MethodSupportsCancellation
-            var tail = Task.Run(async () =>
+            try
             {
-                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.2", exit: true);
 
-                    var server = arguments["<server>"].ToString();
-                    var apiKey = Normalize(arguments["--apikey"]);
-                    var filter = Normalize(arguments["--filter"]);
-                    var window = arguments["--window"].AsInt;
+                var server = arguments["<server>"].ToString();
+                var apiKey = Normalize(arguments["--apikey"]);
+                var filter = Normalize(arguments["--filter"]);
 
-                    await Run(server, apiKey, filter, window, cts.Token);
-                }
-                catch (Exception ex)
-                {
-                    Console.ForegroundColor = ConsoleColor.White;
-                    Console.BackgroundColor = ConsoleColor.Red;
-                    Console.WriteLine("seq-tail: {0}", ex);
-                    Console.ResetColor();
-                    Environment.Exit(-1);
-                }
-            });
+                var cancel = new CancellationTokenSource();
+                Console.WriteLine("Tailing, press Ctrl+C to exit.");
+                Console.CancelKeyPress += (s,a) => cancel.Cancel();
 
-            Console.ReadKey(true);
-            cts.Cancel();
-            // ReSharper disable once MethodSupportsCancellation
-            tail.Wait();
+                var run = Task.Run(() => Run(server, apiKey, filter, cancel));
+
+                run.GetAwaiter().GetResult();
+            }
+            catch (Exception ex)
+            {
+                Log.Fatal(ex, "Tailing aborted");
+                Environment.Exit(-1);
+            }
         }
 
         static string Normalize(ValueObject v)
@@ -66,10 +64,8 @@ static string Normalize(ValueObject v)
             return string.IsNullOrWhiteSpace(s) ? null : s;
         }
 
-        static async Task Run(string server, string apiKey, string filter, int window, CancellationToken cancel)
+        static async Task Run(string server, string apiKey, string filter, CancellationTokenSource cancel)
         {
-            var startedAt = DateTime.UtcNow;
-
             var connection = new SeqConnection(server, apiKey);
 
             string strict = null;
@@ -79,68 +75,14 @@ static async Task Run(string server, string apiKey, string filter, int window, C
                 strict = converted.StrictExpression;
             }
 
-            var result = await connection.Events.ListAsync(count: window, render: true, fromDateUtc: startedAt, filter: strict);
-
-            // Since results may come late, we request an overlapping window and exclude
-            // events that have already been printed. If the last seen ID wasn't returned
-            // we assume the count was too small to cover the window.
-            var lastPrintedBatch = new HashSet<string>();
-            string lastReturnedId = null;
-
-            while (!cancel.IsCancellationRequested)
+            using (var stream = await connection.Events.StreamAsync<JObject>(filter: strict))
             {
-                if (result.Count == 0)
-                {
-                    await Task.Delay(TimeSpan.FromSeconds(1));
-                }
-                else
-                {
-                    var noOverlap = result.All(e => e.Id != lastReturnedId);
-
-                    if (noOverlap && lastReturnedId != null)
-                        Console.WriteLine("<window exceeded>");
-
-                    foreach (var eventEntity in ((IEnumerable<EventEntity>)result).Reverse())
-                    {
-                        if (lastPrintedBatch.Contains(eventEntity.Id))
-                        {
-                            continue;
-                        }
-
-                        lastReturnedId = eventEntity.Id;
-
-                        var exception = "";
-                        if (eventEntity.Exception != null)
-                            exception = Environment.NewLine + eventEntity.Exception;
-
-                        var ts = DateTimeOffset.Parse(eventEntity.Timestamp).ToLocalTime();
-
-                        var color = ConsoleColor.White;
-                        switch (eventEntity.Level)
-                        {
-                            case "Verbose":
-                            case "Debug":
-                                color = ConsoleColor.Gray;
-                                break;
-                            case "Warning":
-                                color = ConsoleColor.Yellow;
-                                break;
-                            case "Error":
-                            case "Fatal":
-                                color = ConsoleColor.Red;
-                                break;
-                        }
-
-                        Console.ForegroundColor = color;
-                        Console.WriteLine("{0:G} [{1}] {2}{3}", ts, eventEntity.Level, eventEntity.RenderedMessage, exception);
-                        Console.ResetColor();
-                    }
-
-                    lastPrintedBatch = new HashSet<string>(result.Select(e => e.Id));
-                }
+                var subscription = stream
+                    .Select(jObject => LogEventReader.ReadFromJObject(jObject))
+                    .Subscribe(evt => Log.Write(evt), () => cancel.Cancel());
 
-                var fromDateUtc = lastReturnedId == null ? startedAt : DateTime.UtcNow.AddMinutes(-3);
-                result = await connection.Events.ListAsync(count: window, render: true, fromDateUtc: fromDateUtc, filter: strict);
+                cancel.Token.WaitHandle.WaitOne();
+                subscription.Dispose();
             }
         }
     }

example/SeqTail/Properties/launchSettings.json

@@ -0,0 +1,8 @@
+{
+  "profiles": {
+    "seq-tail": {
+      "executablePath": "C:\\Development\\nblumhardt\\seq-api\\example\\SeqTail\\bin\\Debug\\net46\\win7-x64\\seq-tail.exe",
+      "commandLineArgs": "http://localhost:5341 --apikey=fSCcBOyOfttZ0kiZFgq"
+    }
+  }
+}

example/SeqTail/project.json

@@ -6,8 +6,13 @@
   },
 
   "dependencies": {
+    "System.Threading.Tasks": "4.0.11",
     "Seq.Api": { "target": "project" },
-    "docopt.net": "0.6.1.9"
+    "docopt.net": "0.6.1.9",
+    "Serilog.Formatting.Compact.Reader": "1.0.0-dev-00008",
+    "Serilog.Sinks.Literate": "2.0.0",
+    "System.Reactive": "3.0.0",
+    "Newtonsoft.Json": "9.0.1"
   },
 
   "frameworks": {

src/Seq.Api/Api/Client/SeqApiClient.cs

@@ -13,6 +13,9 @@
 using Seq.Api.Model.Root;
 using Seq.Api.Serialization;
 using Tavis.UriTemplates;
+using System.Threading;
+using Seq.Api.Streams;
+using System.Net.WebSockets;
 
 namespace Seq.Api.Client
 {
@@ -26,6 +29,7 @@ public class SeqApiClient : IDisposable
         const string SeqApiV3MediaType = "application/vnd.continuousit.seq.v3+json";
 
         readonly HttpClient _httpClient;
+        readonly CookieContainer _cookies = new CookieContainer();
         readonly JsonSerializer _serializer = JsonSerializer.Create(
             new JsonSerializerSettings
             {
@@ -41,7 +45,7 @@ public SeqApiClient(string serverUrl, string apiKey = null)
             if (!string.IsNullOrEmpty(apiKey))
                 _apiKey = apiKey;
 
-            var handler = new HttpClientHandler { CookieContainer = new CookieContainer() };
+            var handler = new HttpClientHandler { CookieContainer = _cookies };
 
             var baseAddress = serverUrl;
             if (!baseAddress.EndsWith("/"))
@@ -115,6 +119,30 @@ public async Task DeleteAsync<TEntity>(ILinked entity, string link, TEntity cont
             new StreamReader(stream).ReadToEnd();
         }
 
+        public async Task<ObservableStream<TEntity>> StreamAsync<TEntity>(ILinked entity, string link, IDictionary<string, object> parameters = null)
+        {
+            return await WebSocketStreamAsync(entity, link, parameters, reader => _serializer.Deserialize<TEntity>(new JsonTextReader(reader)));
+        }
+
+        public async Task<ObservableStream<string>> StreamTextAsync(ILinked entity, string link, IDictionary<string, object> parameters = null)
+        {
+            return await WebSocketStreamAsync(entity, link, parameters, reader => reader.ReadToEnd());
+        }
+
+        async Task<ObservableStream<T>> WebSocketStreamAsync<T>(ILinked entity, string link, IDictionary<string, object> parameters, Func<TextReader, T> deserialize)
+        {
+            var linkUri = ResolveLink(entity, link, parameters);
+
+            var socket = new ClientWebSocket();
+            socket.Options.Cookies = _cookies;
+            if (_apiKey != null)
+                socket.Options.SetRequestHeader("X-Seq-ApiKey", _apiKey);
+
+            await socket.ConnectAsync(new Uri(linkUri), CancellationToken.None);
+
+            return new ObservableStream<T>(socket, deserialize);
+        }
+
         async Task<T> HttpGetAsync<T>(string url)
         {
             var request = new HttpRequestMessage(HttpMethod.Get, url);

src/Seq.Api/Api/ResourceGroups/EventsResourceGroup.cs

@@ -5,6 +5,7 @@
 using Seq.Api.Model.Events;
 using Seq.Api.Model.Shared;
 using Seq.Api.Model.Signals;
+using Seq.Api.Streams;
 
 namespace Seq.Api.ResourceGroups
 {
@@ -156,5 +157,47 @@ public async Task<ResultSetPart> DeleteInSignalAsync(
             var body = signal ?? new SignalEntity();
             return await GroupPostAsync<SignalEntity, ResultSetPart>("DeleteInSignal", body, parameters).ConfigureAwait(false);
         }
+
+        /// <summary>
+        /// Connect to the live event stream. Dispose the resulting stream to disconnect.
+        /// </summary>
+        /// <typeparam name="T">The type into which events should be deserialized.</typeparam>
+        /// <param name="intersectIds">If provided, a list of signal ids whose intersection will be filtered for the result.</param>
+        /// <param name="filter">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().</param>
+        /// <returns>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.</returns>
+        public async Task<ObservableStream<T>> StreamAsync<T>(
+            string[] intersectIds = null,
+            string filter = null)
+        {
+            var parameters = new Dictionary<string, object>();
+            if (intersectIds != null && intersectIds.Length > 0) { parameters.Add("intersectIds", string.Join(",", intersectIds)); }
+            if (filter != null) { parameters.Add("filter", filter); }
+
+            var group = await LoadGroupAsync().ConfigureAwait(false);
+            return await Client.StreamAsync<T>(group, "Stream", parameters).ConfigureAwait(false);
+        }
+
+        /// <summary>
+        /// Retrieve a list of events that match a set of conditions. The complete result is buffered into memory,
+        /// so if a large result set is expected, use InSignalAsync() and lastReadEventId to page the results.
+        /// </summary>
+        /// <param name="intersectIds">If provided, a list of signal ids whose intersection will be filtered for the result.</param>
+        /// <param name="filter">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().</param>
+        /// <returns>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.</returns>
+        public async Task<ObservableStream<string>> StreamDocumentsAsync(
+            string[] intersectIds = null,
+            string filter = null)
+        {
+            var parameters = new Dictionary<string, object>();
+            if (intersectIds != null && intersectIds.Length > 0) { parameters.Add("intersectIds", string.Join(",", intersectIds)); }
+            if (filter != null) { parameters.Add("filter", filter); }
+
+            var group = await LoadGroupAsync().ConfigureAwait(false);
+            return await Client.StreamTextAsync(group, "Stream", parameters).ConfigureAwait(false);
+        }
     }
 }

src/Seq.Api/Api/Streams/ObservableStream.Unsubscriber.cs

@@ -0,0 +1,40 @@
+// Copyright 2016 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<T>
+    {
+        sealed class Unsubscriber : IDisposable
+        {
+            readonly ObservableStream<T> _stream;
+            readonly IObserver<T> _observer;
+
+            public Unsubscriber(ObservableStream<T> sink, IObserver<T> observer)
+            {
+                if (sink == null) throw new ArgumentNullException(nameof(sink));
+                if (observer == null) throw new ArgumentNullException(nameof(observer));
+                _stream = sink;
+                _observer = observer;
+            }
+
+            public void Dispose()
+            {
+                _stream.Unsubscribe(_observer);
+            }
+        }
+    }
+}