Fix bugs with processing partial messages.

Actually use MessageTemplate config item.

Author: tetious

src/SyslogProxy/Messages/JsonSyslogMessage.cs

@@ -1,5 +1,6 @@
 namespace SyslogProxy.Messages
 {
+    using System;
     using System.Linq;
     using System.Text.RegularExpressions;
 
@@ -9,23 +10,46 @@ public class JsonSyslogMessage
     {
         private static readonly Regex DateStampRegex = new Regex(@"\w{3} \w{3}.{4}\d{2}:\d{2}:\d{2}.\d{3}");
 
-        public JsonSyslogMessage(string syslogLine)
+        public JsonSyslogMessage(string rawMessage)
         {
-            var splitLine = syslogLine.Split(' ');
+            this.RawMessage = rawMessage;
 
-            var priority = int.Parse(splitLine[0]);
-            var facility = priority / 8;
-            var severity = priority % 8;
+            var splitLine = rawMessage.Split(' ');
+            if (splitLine.Length < 4)
+            {
+                this.Invalid = true;
+                return;
+            }
+
+            try
+            {
+                var priority = int.Parse(splitLine[0]);
+                var facility = priority / 8;
+                var severity = priority % 8;
 
-            this.Facility = ((Facility) facility).ToString();
-            this.Level = ((Severity) severity).ToString();
+                this.Facility = ((Facility)facility).ToString();
+                this.Level = ((Severity)severity).ToString();
+            }
+            catch (Exception)
+            {
+                Console.WriteLine("Could not parse priority. [{0}]", rawMessage);
+                this.Invalid = true;
+                return;
+            }
+
+            DateTime notUsed;
+            this.Invalid = !DateTime.TryParse(splitLine[1], out notUsed);
 
             this.Timestamp = splitLine[1];
             this.Hostname = splitLine[2].Trim();
             this.ApplicationName = splitLine[3].Trim();
             this.Message = DateStampRegex.Replace(string.Join(" ", splitLine.Skip(4)).Trim(), string.Empty).Trim();
         }
 
+        public bool Invalid { get; private set; }
+
+        public string RawMessage { get; private set; }
+
         public string Timestamp { get; set; }
 
         public string Level { get; set; }
@@ -44,7 +68,7 @@ public override string ToString()
             {
                 Level = this.Level,
                 Timestamp = this.Timestamp,
-                MessageTemplate = "{Hostname}:{ApplicationName} {Message}",
+                MessageTemplate = Configuration.MessageTemplate,
                 Properties = new { this.Facility, this.Hostname, this.ApplicationName, this.Message }
             });
         }

src/SyslogProxy/Proxy.cs

@@ -1,6 +1,7 @@
 namespace SyslogProxy
 {
     using System;
+    using System.Linq;
     using System.Net;
     using System.Net.Sockets;
     using System.Text;
@@ -12,6 +13,8 @@ public class Proxy
     {
         private readonly Action<string> messageHandler;
 
+        private const int BufferSize = 2048;
+
         public Proxy(Action<string> messageHandler)
         {
             this.messageHandler = messageHandler;
@@ -35,11 +38,13 @@ private async Task EchoAsync(TcpClient client)
             Console.WriteLine("New client connected.");
             using (client)
             {
-                var buf = new byte[4096];
                 var stream = client.GetStream();
+                var buf = new byte[BufferSize];
+                var accumulator = new StringBuilder();
                 while (true)
                 {
                     var timeoutTask = Task.Delay(TimeSpan.FromSeconds(Configuration.TcpConnectionTimeout));
+                    Array.Clear(buf, 0, BufferSize);
                     var amountReadTask = stream.ReadAsync(buf, 0, buf.Length);
                     var completedTask = await Task.WhenAny(timeoutTask, amountReadTask)
                                                   .ConfigureAwait(false);
@@ -54,8 +59,14 @@ private async Task EchoAsync(TcpClient client)
                     {
                         break;
                     }
-
-                    this.messageHandler(Encoding.UTF8.GetString(buf).TrimEnd('\0'));
+                    accumulator.Append(Encoding.UTF8.GetString(buf).TrimEnd('\0'));
+                    if (accumulator.ToString().Contains("\n"))
+                    {
+                        var splitMessage = accumulator.ToString().Split('\n').ToList();
+                        accumulator = new StringBuilder(splitMessage.Last());
+                        splitMessage.RemoveAt(splitMessage.Count - 1);
+                        splitMessage.ForEach(this.messageHandler);
+                    }
                 }
             }
             Console.WriteLine("Client disconnected");

src/SyslogProxy/SeqWriter.cs

@@ -14,6 +14,12 @@ public class SeqWriter
 
         public async Task WriteToSeq(JsonSyslogMessage message, int delay = 0)
         {
+            if (message.Invalid)
+            {
+                Console.WriteLine("Skipping incomplete/invalid message. [{0}]", message.RawMessage);
+                return;
+            }
+
             await Task.Delay(Math.Min(delay, 60000));
 
             ExceptionDispatchInfo capturedException = null;
@@ -30,7 +36,7 @@ public async Task WriteToSeq(JsonSyslogMessage message, int delay = 0)
             {
                 Console.Write("Couldn't write to SEQ. Exception: [{0}]", capturedException.SourceException.Message);
                 this.retryCount++;
-                await this.WriteToSeq(message, 100 ^ this.retryCount);
+                await this.WriteToSeq(message, (int)Math.Pow(100, this.retryCount));
             }
         }