Plain text ingestion WIP: read frames of input with trailing lines

src/SeqCli/Csv/CsvTokenizer.cs

@@ -8,7 +8,7 @@ namespace SeqCli.Csv
 {
     class CsvTokenizer : Tokenizer<CsvToken>
     {
-        static readonly TextParser<TextSpan> Content = Span.While(ch => ch != '"');
+        static readonly TextParser<TextSpan> Content = Span.WithoutAny(ch => ch == '"');
 
         protected override IEnumerable<Result<CsvToken>> Tokenize(TextSpan span)
         {
@@ -26,9 +26,9 @@ protected override IEnumerable<Result<CsvToken>> Tokenize(TextSpan span)
                     if (!next.HasValue) yield break;
 
                     var text = Content(next.Location);
-                    while (text.HasValue)
+                    while (text.HasValue || !text.Remainder.IsAtEnd)
                     {
-                        if (text.Value.Length > 0)
+                        if (text.HasValue)
                         {
                             if (TryMatchSpecialContent(text.Value, out var specialTokenType) &&
                                 !IsEscapedDoubleQuote(text.Remainder))

src/SeqCli/PlainText/Frame.cs

@@ -0,0 +1,23 @@
+// Copyright 2018 Datalust Pty Ltd
+//
+// 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.
+
+namespace SeqCli.PlainText
+{
+    struct Frame
+    {
+        public bool HasValue { get; set; }
+        public bool IsOrphan { get; set; }
+        public string Value { get; set; }
+    }
+}

src/SeqCli/PlainText/FrameReader.cs

@@ -0,0 +1,126 @@
+// Copyright 2018 Datalust Pty Ltd
+//
+// 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.IO;
+using System.Text;
+using System.Threading.Tasks;
+using Superpower;
+using Superpower.Model;
+
+namespace SeqCli.PlainText
+{
+    class FrameReader
+    {
+        readonly TextReader _source;
+        readonly TimeSpan _trailingLineArrivalDeadline;
+        readonly TextParser<TextSpan> _frameStart;
+
+        string _unconsumedFirstLine;
+        Task<string> _unawaitedNextLine;
+
+        public FrameReader(TextReader source, TextParser<TextSpan> frameStart, TimeSpan trailingLineArrivalDeadline)
+        {
+            _source = source ?? throw new ArgumentNullException(nameof(source));
+            _frameStart = frameStart ?? throw new ArgumentNullException(nameof(frameStart));
+            _trailingLineArrivalDeadline = trailingLineArrivalDeadline;
+        }
+
+        public async Task<Frame> TryReadAsync()
+        {
+            var valueBuilder = new StringBuilder();
+            var hasValue = false;
+
+            if (_unconsumedFirstLine != null)
+            {
+                valueBuilder.AppendLine(_unconsumedFirstLine);
+                _unconsumedFirstLine = null;
+                hasValue = true;
+            }
+            else if (_unawaitedNextLine != null)
+            {
+                var line = await _unawaitedNextLine;
+                _unawaitedNextLine = null;
+                if (line == null)
+                    return new Frame();
+
+                valueBuilder.AppendLine(line);
+                hasValue = true;
+
+                if (!IsFrameStart(line))
+                    return new Frame {HasValue = true, IsOrphan = true, Value = valueBuilder.ToString()};
+            }
+
+            Task<string> readLine = null;
+            while (true)
+            {
+                readLine = readLine ?? Task.Run(_source.ReadLineAsync);                
+                var index = Task.WaitAny(new Task[] {readLine}, _trailingLineArrivalDeadline);
+                if (index == -1) // Timeout
+                {
+                    if (hasValue)
+                    {
+                        _unawaitedNextLine = readLine;
+                        return new Frame {HasValue = true, Value = valueBuilder.ToString()};
+                    }
+
+                    // else, around we go!
+                }
+                else
+                {
+                    var line = await readLine;
+                    readLine = null;
+                    if (line == null)
+                    {
+                        if (hasValue)
+                        {
+                            return new Frame {HasValue = true, Value = valueBuilder.ToString()};
+                        }
+
+                        return new Frame();
+                    }
+
+                    if (IsFrameStart(line))
+                    {
+                        if (hasValue)
+                        {
+                            _unconsumedFirstLine = line;
+                            return new Frame {HasValue = true, Value = valueBuilder.ToString()};
+                        }
+
+                        valueBuilder.AppendLine(line);
+                        hasValue = true;
+                    }
+                    else
+                    {
+                        if (!hasValue)
+                        {
+                            valueBuilder.AppendLine(line);
+                            return new Frame {HasValue = true, Value = valueBuilder.ToString(), IsOrphan = true};
+                        }
+
+                        valueBuilder.AppendLine(line);
+                    }
+                }
+            }
+
+            bool IsFrameStart(string line)
+            {
+                if (line == null) throw new ArgumentNullException(nameof(line));
+                var result = _frameStart(new TextSpan(line));
+                return result.HasValue && result.Value.Length > 0;
+            }
+        }
+    }
+}

src/SeqCli/SeqCli.csproj

@@ -25,7 +25,7 @@
     <PackageReference Include="Serilog.Formatting.Compact.Reader" Version="1.0.2" />
     <PackageReference Include="Serilog.Sinks.Console" Version="3.0.1" />
     <PackageReference Include="Autofac" Version="4.0.0" />
-    <PackageReference Include="Superpower" Version="1.1.0" />
+    <PackageReference Include="Superpower" Version="2.0.0-*" />
     <PackageReference Include="System.Reactive" Version="3.1.1" />
     <PackageReference Include="System.Security.Cryptography.ProtectedData" Version="4.4.0" />
     <PackageReference Include="Seq.Api" Version="4.2.2" />

test/SeqCli.Tests/PlainText/FrameReaderTests.cs

@@ -0,0 +1,104 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Text;
+using System.Threading.Tasks;
+using SeqCli.PlainText;
+using Superpower;
+using Superpower.Model;
+using Superpower.Parsers;
+using Xunit;
+
+namespace SeqCli.Tests.PlainText
+{
+    public class FrameReaderTests
+    {
+        static TextParser<TextSpan> SpanMatchedBy<T>(TextParser<T> parser)
+        {
+            return i =>
+            {
+                var result = parser(i);
+
+                if (!result.HasValue)
+                    return Result.CastEmpty<T, TextSpan>(result);
+
+                return Result.Value(
+                    i.Until(result.Remainder),
+                    i,
+                    result.Remainder);
+            };
+        }
+
+        [Fact]
+        public async Task SplitsLinesIntoFrames()
+        {
+            var source = new StringBuilder();
+            source.AppendLine("first");
+            source.AppendLine("second");
+
+            var reader = new FrameReader(
+                new StringReader(source.ToString()),
+                SpanMatchedBy(Character.Letter),
+                TimeSpan.FromMilliseconds(1));
+
+            var first = await reader.TryReadAsync();
+            Assert.True(first.HasValue);
+            Assert.Equal("first" + Environment.NewLine, first.Value);
+
+            var second = await reader.TryReadAsync();
+            Assert.True(second.HasValue);
+            Assert.Equal("second" + Environment.NewLine, second.Value);
+
+            var empty = await reader.TryReadAsync();
+            Assert.False(empty.HasValue);
+        }
+
+        [Fact]
+        public async Task TerminatesWhenNoLinesArePresent()
+        {
+            var reader = new FrameReader(
+                new StringReader(""),
+                SpanMatchedBy(Character.Letter),
+                TimeSpan.FromMilliseconds(1));
+
+            var none = await reader.TryReadAsync();
+            Assert.False(none.HasValue);
+        }
+
+        [Fact]
+        public async Task CollectsTrailingLines()
+        {
+            var source = new StringBuilder();
+            source.AppendLine("first");
+            source.AppendLine(" some more");
+            source.AppendLine(" and more");
+            source.AppendLine("second");
+            source.AppendLine("third");
+            source.AppendLine(" and yet more");
+
+            var frames = await ReadAllFrames(source.ToString(), SpanMatchedBy(Character.Letter));
+            Assert.Equal(3, frames.Length);
+            Assert.StartsWith("first", frames[0].Value);
+            Assert.EndsWith("and more" + Environment.NewLine, frames[0].Value);
+        }
+
+        static async Task<Frame[]> ReadAllFrames(string source, TextParser<TextSpan> frameStart)
+        {
+            var reader = new FrameReader(
+                new StringReader(source),
+                frameStart,
+                TimeSpan.FromMilliseconds(1));
+
+            var result = new List<Frame>();
+
+            var frame = await reader.TryReadAsync();
+            while (frame.HasValue)
+            {
+                result.Add(frame);
+                frame = await reader.TryReadAsync();
+            }
+
+            return result.ToArray();
+        }
+    }
+}