Removed some cleanup

UdpListener shutdown simplification

README.md

@@ -25,7 +25,8 @@ https://groups.google.com/forum/#!forum/timberwinr
 
 Latest Build:
 
-![alt tag](https://ci.appveyor.com/api/projects/status/github/Cimpress-MCP/TimberWinR)
+![alt tag](https://ci.appveyor.com/api/projects/status/qi5x2lg153h1tim6/branch/master?svg=true)
+
 
 ## Inputs
 The current supported Input format sources are:
@@ -36,7 +37,8 @@ The current supported Input format sources are:
  5. [Stdin](https://github.com/Cimpress-MCP/TimberWinR/blob/master/TimberWinR/mdocs/StdinInput.md) (Standard Input for Debugging)
  6. [W3C](https://github.com/Cimpress-MCP/TimberWinR/blob/master/TimberWinR/mdocs/W3CInput.md)(Internet Information Services W3C Advanced/Custom Format)
  7. [Udp](https://github.com/Cimpress-MCP/TimberWinR/blob/master/TimberWinR/mdocs/UdpInput.md) (listens for UDP on port for JSON messages)
- 8. [TailFiles](https://github.com/Cimpress-MCP/TimberWinR/blob/master/TimberWinR/mdocs/TailFiles.md) (Tails log files efficiently *New*)
+ 8. [TailFiles](https://github.com/Cimpress-MCP/TimberWinR/blob/master/TimberWinR/mdocs/TailFiles.md) (Tails log files efficiently)
+ 8. [Generator](https://github.com/Cimpress-MCP/TimberWinR/blob/master/TimberWinR/mdocs/Generator.md) (Generate logs for testing *New*)
 
 ## Codecs
 The current list of supported codecs are:

TimberWinR.ServiceHost/Program.cs

@@ -28,9 +28,6 @@ namespace TimberWinR.ServiceHost
         
         private static void Main(string[] args)
         {
-
-           
-
             Arguments arguments = new Arguments();
            
             HostFactory.Run(hostConfigurator =>

TimberWinR.ServiceHost/Properties/AssemblyInfo.cs

@@ -32,5 +32,5 @@ using System.Runtime.InteropServices;
 // You can specify all the values or you can default the Build and Revision Numbers 
 // by using the '*' as shown below:
 // [assembly: AssemblyVersion("1.0.*")]
-[assembly: AssemblyVersion("1.3.21.0")]
+[assembly: AssemblyVersion("1.3.24.0")]
-[assembly: AssemblyFileVersion("1.3.21.0")]
+[assembly: AssemblyFileVersion("1.3.24.0")]

TimberWinR.TestGenerator/CommandLineOptions.cs

@@ -15,6 +15,9 @@ namespace TimberWinR.TestGenerator
         [Option("timberWinRConfig", DefaultValue = "default.json", HelpText = "Config file/directory to use")]
         public string TimberWinRConfigFile { get; set; }
 
+        [Option("start", HelpText = "Start an instance of TimberWinR")]
+        public bool StartTimberWinR { get; set; }
+
         [Option("testDir", DefaultValue = ".", HelpText = "Test directory to use (created if necessary)")]
         public string TestDir { get; set; }
 

TimberWinR.TestGenerator/JsonLogFileGenerator.cs

@@ -63,8 +63,7 @@ namespace TimberWinR.TestGenerator
             // This text is always added, making the file longer over time 
             // if it is not deleted. 
             using (StreamWriter sw = File.AppendText(logFilePath))
-            {
+            {               
-                sw.AutoFlush = true;
                 for (int i = 0; i < parms.NumMessages; i++)
                 {
                     JObject o = new JObject

TimberWinR.TestGenerator/LogFileGenerator.cs

@@ -64,8 +64,7 @@ namespace TimberWinR.TestGenerator
             // This text is always added, making the file longer over time 
             // if it is not deleted. 
             using (StreamWriter sw = File.AppendText(logFilePath))
-            {
+            {              
-                sw.AutoFlush = true;
                 for (int i = 0; i < parms.NumMessages; i++)
                 {
                     JObject o = new JObject

TimberWinR.TestGenerator/Program.cs

@@ -4,6 +4,7 @@ using System.Diagnostics;
 using System.IO;
 using System.Linq;
 using System.Linq.Expressions;
+using System.Net;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
@@ -55,7 +56,7 @@ namespace TimberWinR.TestGenerator
 
             ramCounter.CategoryName = "Memory";
             ramCounter.CounterName = "% Committed Bytes In Use";
-           
+
             Options = new CommandLineOptions();
 
             if (CommandLine.Parser.Default.ParseArguments(args, Options))
@@ -86,7 +87,8 @@ namespace TimberWinR.TestGenerator
                 var sw = Stopwatch.StartNew();
 
                 // Startup TimberWinR
-                StartTimberWinR(Options.TimberWinRConfigFile, Options.LogLevel, ".", false);
+                if (Options.StartTimberWinR)
+                    StartTimberWinR(Options.TimberWinRConfigFile, Options.LogLevel, ".", false);
 
                 // Run the Generators
                 var arrayOfTasks = RunGenerators(Options);
@@ -114,7 +116,16 @@ namespace TimberWinR.TestGenerator
                 sw.Start();
 
                 // Get all the stats
-                var jsonTimberWinr = ShutdownTimberWinR();
+                JObject jsonTimberWinr;
+
+                if (Options.StartTimberWinR)
+                    jsonTimberWinr = ShutdownTimberWinR();
+                else
+                {
+                    jsonTimberWinr = GetDiagnosticsOutput();
+                    if (jsonTimberWinr == null)
+                        return 3;
+                }
 
                 LogManager.GetCurrentClassLogger().Info("Finished Shutdown: " + sw.Elapsed);
                 sw.Stop();
@@ -131,6 +142,30 @@ namespace TimberWinR.TestGenerator
             return 1;
         }
 
+        public static string GET(string url)
+        {
+            try
+            {
+                HttpWebRequest request = (HttpWebRequest)WebRequest.Create(url);
+                HttpWebResponse response = (HttpWebResponse)request.GetResponse();
+                Stream stream = response.GetResponseStream();
+                StreamReader reader = new StreamReader(stream);
+
+                string data = reader.ReadToEnd();
+
+                reader.Close();
+                stream.Close();
+
+                return data;
+            }
+            catch (Exception e)
+            {
+                LogManager.GetCurrentClassLogger().ErrorException("Error in GET", e);
+            }
+
+            return null;
+        }
+
         private static void CopySourceFile(string fileName, string outputDir)
         {
             FileInfo fi = new FileInfo(fileName);
@@ -199,16 +234,16 @@ namespace TimberWinR.TestGenerator
                 switch (inputProp.Name)
                 {
                     case "udp":
-                        if (VerifyConditions(json, new string[] {"udp"}, inputProp, jresult) != 0)
+                        if (VerifyConditions(json, new string[] { "udp" }, inputProp, jresult) != 0)
                             return 1;
                         break;
                     case "tcp":
-                        if (VerifyConditions(json, new string[] {"tcp"}, inputProp, jresult) != 0)
+                        if (VerifyConditions(json, new string[] { "tcp" }, inputProp, jresult) != 0)
                             return 1;
                         break;
                     case "log":
                     case "taillog":
-                        if (VerifyConditions(json, new string[] {"log", "taillog"}, inputProp, jresult) != 0)
+                        if (VerifyConditions(json, new string[] { "log", "taillog" }, inputProp, jresult) != 0)
                             return 1;
                         break;
                 }
@@ -269,13 +304,32 @@ namespace TimberWinR.TestGenerator
             return 0;
         }
 
+
+        private static JObject GetDiagnosticsOutput()
+        {
+            if (Diagnostics != null)
+                return Diagnostics.DiagnosticsOutput();
+            else
+            {
+                var jsonDiag = GET("http://localhost:5141");
+                if (jsonDiag == null)
+                {
+                    LogManager.GetCurrentClassLogger().Error("TimberWinR diagnostics port not responding.");
+                    return null;
+                }
+                return JObject.Parse(jsonDiag);
+            }        
+        }
+
         // Wait till all output has been transmitted.
         private static void WaitForOutputTransmission()
         {
             bool completed = false;
             do
             {
-                var json = Diagnostics.DiagnosticsOutput();
+                var json = GetDiagnosticsOutput();
+                if (json == null)
+                    return;
 
                 //Console.WriteLine(json.ToString(Formatting.Indented));
 
@@ -337,21 +391,27 @@ namespace TimberWinR.TestGenerator
 
         private static JObject ShutdownTimberWinR()
         {
-            _timberWinR.Shutdown();
+            if (_timberWinR != null)
+            {
+                // Cancel any/all other threads
+                _cancellationTokenSource.Cancel();
 
-            // Cancel any/all other threads
+                _timberWinR.Shutdown();
-            _cancellationTokenSource.Cancel();
 
-            var json = Diagnostics.DiagnosticsOutput();
+                var json = Diagnostics.DiagnosticsOutput();
 
-            LogManager.GetCurrentClassLogger()
+                LogManager.GetCurrentClassLogger()
-                .Info("Average CPU Usage: {0}%, Average RAM Usage: {1}MB, Max CPU: {2}%, Max Mem: {3}MB", _avgCpuUsage, _avgMemUsage, _maxCpuUsage, _maxMemUsage);
+                    .Info("Average CPU Usage: {0}%, Average RAM Usage: {1}MB, Max CPU: {2}%, Max Mem: {3}MB",
+                        _avgCpuUsage, _avgMemUsage, _maxCpuUsage, _maxMemUsage);
 
-            LogManager.GetCurrentClassLogger().Info(json.ToString());
+                LogManager.GetCurrentClassLogger().Info(json.ToString());
 
-            Diagnostics.Shutdown();
+                Diagnostics.Shutdown();
 
-            return json;
+                return json;
+            }
+
+            return new JObject();
         }
 
         static void StartTimberWinR(string configFile, string logLevel, string logFileDir, bool enableLiveMonitor)
@@ -363,7 +423,7 @@ namespace TimberWinR.TestGenerator
         }
 
         private static void TimberWinROnOnConfigurationProcessed(Configuration configuration)
-        {         
+        {
             if (!string.IsNullOrEmpty(Options.RedisHost) && configuration.RedisOutputs != null && configuration.RedisOutputs.Count() > 0)
             {
                 foreach (var ro in configuration.RedisOutputs)

TimberWinR.TestGenerator/TimberWinR.TestGenerator.csproj

@@ -108,6 +108,15 @@
     <Content Include="results3.json">
       <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
     </Content>
+    <Content Include="test4-tw.json">
+      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+    </Content>
+    <Content Include="results4.json">
+      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+    </Content>
+    <Content Include="test4.json">
+      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+    </Content>
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\TimberWinR\TimberWinR.csproj">

TimberWinR.TestGenerator/results1.json

@@ -5,14 +5,14 @@
                 "taillog": {
                     "test1: message sent count": "[messages] == 7404",
                     "test2: average cpu": "[avgCpuUsage] <= 30",
-                    "test3: maximum memory": "[maxMemUsage] <= 20"                
+                    "test3: maximum memory": "[maxMemUsage] <= 30"                
                 }
             },
             {
                 "udp": {
                     "test1: message sent count": "[messages] == 1234",
                     "test2: average cpu": "[avgCpuUsage] <= 30",
-                    "test3: maximum memory": "[maxMemUsage] <= 20"              
+                    "test3: maximum memory": "[maxMemUsage] <= 30"              
                 }
             }
         ]

TimberWinR.TestGenerator/results2.json

@@ -5,14 +5,14 @@
                 "taillog": {
                     "test1: message sent count": "[messages] == 7404",
                     "test2: average cpu": "[avgCpuUsage] <= 30",
-                    "test3: maximum memory": "[maxMemUsage] <= 15"
+                    "test3: maximum memory": "[maxMemUsage] <= 20"
                 }
             },
             {
                 "udp": {
                     "test1: message sent count": "[messages] == 1234",
                     "test2: average cpu": "[avgCpuUsage] <= 30",
-                    "test3: maximum memory": "[maxMemUsage] <= 15"
+                    "test3: maximum memory": "[maxMemUsage] <= 20"
                 }
             }
         ]

TimberWinR.TestGenerator/results4.json

@@ -0,0 +1,20 @@
+{
+    "Results": {
+        "Inputs": [
+            {
+                "taillog": {
+                    "test1: message sent count": "[messages] == 7404",
+                    "test2: average cpu": "[avgCpuUsage] <= 30",
+                    "test3: maximum memory": "[maxMemUsage] <= 15"
+                }
+            },
+            {
+                "tcp": {
+                    "test4: message sent count": "[messages] == 1234",
+                    "test5: average cpu": "[avgCpuUsage] <= 30",
+                    "test6: maximum memory": "[maxMemUsage] <= 15"
+                }
+            }
+        ]
+    }
+}

TimberWinR.TestGenerator/test1-twconfig.json

@@ -1,14 +1,14 @@
 {
     "TimberWinR": {
-        "Inputs": {           
+        "Inputs": {
             "Udp": [
                 {
                     "_comment": "Output from NLog",
                     "port": 5140
                 }
             ],
-             "TailFiles": [
+            "TailFiles": [
-                {  
+                {
                     "interval": 5,
                     "logSource": "log files",
                     "location": "*.jlog",
@@ -25,6 +25,11 @@
                         ""
                     ],
                     "drop": "true"
+                },
+                "json": {
+                    "type": "Win32-TailFile",
+                    "source": "Text",
+                    "promote": "Text"
                 }
             }
         ],
@@ -34,7 +39,7 @@
                     "_comment": "Change the host to your Redis instance",
                     "port": 6379,
                     "batch_count": 500,
-                    "threads":  2,
+                    "threads": 2,
                     "host": [
                         "tstlexiceapp006.vistaprint.svc"
                     ]

TimberWinR.TestGenerator/test1.json

@@ -1,13 +1,14 @@
 {
     "test": "Test 1",
     "arguments": {
+        "--start":  "",
         "--testFile": "test1.json",
         "--testDir": "test1",
         "--timberWinRConfig": "test1-twconfig.json",
         "--numMessages": 1234,
         "--logLevel": "debug",
-        "--udp-host":  "::1",
+        "--udp-host":  "localhost",
-        "--udp": "5140",
+        "--udp": "5140",       
         "--jroll": ["r1.jlog", "r2.jlog"],
         "--json": ["1.jlog", "2.jlog", "3.jlog", "4.jlog"],
         "--resultsFile": "results1.json"

TimberWinR.TestGenerator/test2.json

@@ -5,7 +5,7 @@
         "--testDir": "test2",
         "--timberWinRConfig": "test2-tw.json",
         "--numMessages": 1234,
-        "--logLevel": "debug",
+        "--logLevel": "trace",
         "--udp": "5140",
         "--jroll": ["r1.jlog", "r2.jlog"],
         "--json": ["1.jlog", "2.jlog", "3.jlog", "4.jlog"],

TimberWinR.TestGenerator/test3-tw.json

@@ -1,17 +1,17 @@
 {
     "TimberWinR": {
-        "Inputs": {           
+        "Inputs": {
-            "Tcp": [
+            "Udp": [
                 {
                     "_comment": "Output from NLog",
                     "port": 5140
                 }
             ],
-             "Logs": [
+            "TailFiles": [
-                {  
+                {
                     "interval": 5,
                     "logSource": "log files",
-                    "location": "*.jlog",
+                    "location": "d:\\logs\\sta\\sta.log",
                     "recurse": -1
                 }
             ]
@@ -25,6 +25,11 @@
                         ""
                     ],
                     "drop": "true"
+                },
+                "json": {
+                    "type": "Win32-TailFile",
+                    "source": "Text",
+                    "promote": "Text"
                 }
             }
         ],
@@ -34,7 +39,7 @@
                     "_comment": "Change the host to your Redis instance",
                     "port": 6379,
                     "batch_count": 500,
-                    "threads":  2,
+                    "threads": 2,
                     "host": [
                         "tstlexiceapp006.vistaprint.svc"
                     ]

TimberWinR.TestGenerator/test3.json

@@ -1,14 +1,12 @@
 {
     "test": "Test 3",
     "arguments": {
+        "--start":  "",
         "--testFile": "test3.json",
         "--testDir": "test3",
         "--timberWinRConfig": "test3-tw.json",
         "--numMessages": 1234,
-        "--logLevel": "debug",
+        "--logLevel": "debug",         
-        "--tcp": "5140",
-        "--jroll": ["r1.jlog", "r2.jlog"],
-        "--json": ["1.jlog", "2.jlog", "3.jlog", "4.jlog"],
         "--resultsFile": "results3.json"
     }
 }

TimberWinR.TestGenerator/test4-tw.json

@@ -0,0 +1,50 @@
+{
+    "TimberWinR": {
+        "Inputs": {
+            "Udp": [
+                {
+                    "_comment": "Output from NLog",
+                    "port": 5140
+                }
+            ],
+            "TailFiles": [
+                {
+                    "interval": 5,
+                    "logSource": "log files",
+                    "location": "d:\\logs\\sta\\sta.log",
+                    "recurse": -1
+                }
+            ]
+        },
+        "Filters": [
+            {
+                "grok": {
+                    "condition": "\"[EventTypeName]\" == \"Information Event\"",
+                    "match": [
+                        "Text",
+                        ""
+                    ],
+                    "drop": "true"
+                },
+                "json": {
+                    "type": "Win32-TailFile",
+                    "source": "Text",
+                    "promote": "Text"
+                }
+            }
+        ],
+        "Outputs": {
+            "Redis": [
+                {
+                    "_comment": "Change the host to your Redis instance",
+                    "port": 6379,
+                    "batch_count": 500,
+                    "threads": 2,
+                    "host": [
+                        "tstlexiceapp006.vistaprint.svc"
+                    ]
+                }
+            ]
+        }
+    }
+}

TimberWinR.TestGenerator/test4.json

@@ -0,0 +1,11 @@
+{
+    "test": "Test 4",
+    "arguments": {
+        "--testFile": "test4.json",                
+        "--testDir": "test4",
+        "--timberWinRConfig": "test4-tw.json",
+        "--numMessages": 1234,
+        "--logLevel": "debug",         
+        "--resultsFile": "results4.json"
+    }
+}

TimberWinR/Codecs/JsonCodec.cs

@@ -0,0 +1,26 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using Newtonsoft.Json.Linq;
+using TimberWinR.Parser;
+
+namespace TimberWinR.Codecs
+{
+    class JsonCodec : ICodec
+    {
+        private CodecArguments _codecArguments;
+
+        public void Apply(string msg, Inputs.InputListener listener)
+        {
+            JObject jobject = JObject.Parse(msg);
+            listener.AddDefaultFields(jobject);
+            listener.ProcessJson(jobject);
+        }
+
+        public JsonCodec(CodecArguments args)
+        {
+            _codecArguments = args;
+        }
+    }
+}

TimberWinR/Codecs/PlainCodec.cs

@@ -0,0 +1,39 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using Newtonsoft.Json.Linq;
+using TimberWinR.Parser;
+
+namespace TimberWinR.Codecs
+{
+    public class PlainCodec : ICodec
+    {
+        private CodecArguments _codecArguments;
+
+        public void Apply(string msg, Inputs.InputListener listener)
+        {
+            JObject json = new JObject();
+            listener.AddDefaultFields(json);
+            json["message"] = ExpandField(msg, json);
+            listener.ProcessJson(json);
+        }
+
+        protected string ExpandField(string fieldName, JObject json)
+        {
+            foreach (var token in json.Children())
+            {
+                string replaceString = "%{" + token.Path + "}";
+                fieldName = fieldName.Replace(replaceString, json[token.Path].ToString());
+            }
+            return fieldName;
+        }
+
+
+        public PlainCodec(CodecArguments args)
+        {
+            _codecArguments = args;
+        }
+
+    }
+}

TimberWinR/Configuration.cs

@@ -102,6 +102,12 @@ namespace TimberWinR
             get { return _stdins; }
         }
 
+        private List<GeneratorParameters> _generators = new List<GeneratorParameters>();
+        public IEnumerable<GeneratorParameters> Generators
+        {
+            get { return _generators; }
+        }
+
         private List<LogstashFilter> _filters = new List<LogstashFilter>();
 
         public IEnumerable<LogstashFilter> Filters
@@ -239,6 +245,8 @@ namespace TimberWinR
                     c._iisw3clogs.AddRange(x.TimberWinR.Inputs.IISW3CLogs.ToList());
                 if (x.TimberWinR.Inputs.Stdins != null)
                     c._stdins.AddRange(x.TimberWinR.Inputs.Stdins.ToList());
+                if (x.TimberWinR.Inputs.Generators != null)
+                    c._generators.AddRange(x.TimberWinR.Inputs.Generators.ToList());
                 if (x.TimberWinR.Inputs.Logs != null)
                     c._logs.AddRange(x.TimberWinR.Inputs.Logs.ToList());
                 if (x.TimberWinR.Inputs.TailFilesArguments != null)

TimberWinR/Filters/JsonFilter.cs

@@ -3,6 +3,7 @@ using System.Collections.Generic;
 using System.IO;
 using System.Linq;
 using System.Text;
+using System.Threading;
 using System.Xml.Linq;
 using Newtonsoft.Json;
 using Newtonsoft.Json.Linq;
@@ -12,6 +13,8 @@ namespace TimberWinR.Parser
 {
     public partial class Json : LogstashFilter
     {
+        private long _errorCount;
+
         public Json()
         {
             RemoveSource = true;
@@ -22,6 +25,7 @@ namespace TimberWinR.Parser
                new JProperty("json",
                    new JObject(
                        new JProperty("condition", Condition),
+                       new JProperty("errors", _errorCount),
                        new JProperty("source", Source),
                        new JProperty("promote", Source),
                        new JProperty("target", Target),
@@ -102,6 +106,7 @@ namespace TimberWinR.Parser
                 catch (Exception ex)
                 {
                     LogManager.GetCurrentClassLogger().Error(ex);
+                    Interlocked.Increment(ref _errorCount);
                     return true;
                 }
             }

TimberWinR/Inputs/GeneratorInput.cs

@@ -0,0 +1,88 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+
+using Newtonsoft.Json;
+using Newtonsoft.Json.Linq;
+using NLog;
+using RestSharp.Extensions;
+using TimberWinR.Codecs;
+using TimberWinR.Parser;
+
+
+namespace TimberWinR.Inputs
+{
+    public class GeneratorInput : InputListener
+    {
+        public override JObject ToJson()
+        {
+            JObject json = new JObject(
+                new JProperty("message", _params.Message),
+                new JProperty("messages", _sentMessages),
+                new JProperty("generator", "enabled"));
+            return json;
+        }
+
+        private TimberWinR.Parser.GeneratorParameters _params;
+        private Thread _listenThread;
+        private ICodec _codec;
+        private int _sentMessages;
+
+        public GeneratorInput(TimberWinR.Parser.GeneratorParameters parameters, CancellationToken cancelToken)
+            : base(cancelToken, "Win32-InputGen")
+        {
+            _params = parameters;
+
+            if (_params.CodecArguments != null)
+            {
+                switch (_params.CodecArguments.Type)
+                {
+                    case CodecArguments.CodecType.json:
+                        _codec = new JsonCodec(_params.CodecArguments);
+                        break;
+                    case CodecArguments.CodecType.multiline:
+                        _codec = new Multiline(_params.CodecArguments);
+                        break;
+                    case CodecArguments.CodecType.plain:
+                        _codec = new PlainCodec(_params.CodecArguments);
+                        break;
+                }
+            }
+
+            _listenThread = new Thread(new ThreadStart(GenerateData));
+            _listenThread.Start();
+        }
+
+        private void GenerateData()
+        {
+            LogManager.GetCurrentClassLogger().Info("Generator Creating {0} Lines", _params.Count);
+
+            int numMessages = _params.Count;         
+
+            // Infinite or until done.
+            for (int i = 0; (_params.Count == 0 || i < numMessages); i++)
+            {
+                if (CancelToken.IsCancellationRequested)
+                    break;
+
+                string msg = ToPrintable(_params.Message);
+
+                if (_codec != null)
+                    _codec.Apply(msg, this);
+                else
+                {
+                    JObject jo = new JObject();
+                    jo["Message"] = msg;
+                    AddDefaultFields(jo);
+                    ProcessJson(jo);
+                }
+
+                Thread.Sleep(_params.Rate);
+            }
+
+            Finished();
+        }
+    }
+}

TimberWinR/Inputs/LogsFileDatabase.cs

@@ -30,18 +30,10 @@ namespace TimberWinR.Inputs
 
         private static LogsFileDatabase instance;
 
-        private bool ExistingFile(string logName)
-        {
-            lock (_locker)
-            {
-                return ExistingFileTest(logName);
-            }
-        }
-       
         //
         // Lookup the database entry for this log file, returns null if there isnt one.
         //
-        private LogsFileDatabaseEntry FindFile(string logName)
+        private LogsFileDatabaseEntry FindFileWithLock(string logName)
         {
             lock (_locker)
             {
@@ -69,53 +61,56 @@ namespace TimberWinR.Inputs
             }
         }
 
-        private LogsFileDatabaseEntry AddFileEntry(string logName)
+        private LogsFileDatabaseEntry AddFileEntryWithLock(string logName)
         {
             var de = new LogsFileDatabaseEntry();
             lock (_locker)
             {
-                var fi = new FileInfo(logName);             
+                var fi = new FileInfo(logName);
                 de.FileName = logName;
                 de.LogFileExists = fi.Exists;
+                de.Previous = "";
                 de.NewFile = true;
                 de.ProcessedFile = false;
                 de.LastPosition = fi.Length;
                 de.SampleTime = DateTime.UtcNow;
-                de.CreationTimeUtc = fi.CreationTimeUtc;               
+                de.CreationTimeUtc = fi.CreationTimeUtc;
 
                 Entries.Add(de);
                 WriteDatabaseFileNoLock();
             }
             return de;
-        }      
+        }
 
         public static LogsFileDatabaseEntry LookupLogFile(string logName)
         {
-            LogsFileDatabaseEntry dbe = Instance.FindFile(logName);
+            LogsFileDatabaseEntry dbe = Instance.FindFileWithLock(logName);
             if (dbe == null)
-                dbe = Instance.AddFileEntry(logName);
+                dbe = Instance.AddFileEntryWithLock(logName);
 
-            FileInfo fi = new FileInfo(logName);          
+            FileInfo fi = new FileInfo(logName);
 
             dbe.LogFileExists = fi.Exists;
             var creationTime = fi.CreationTimeUtc;
 
             if (dbe.LogFileExists && creationTime != dbe.CreationTimeUtc)
-                dbe.NewFile = true;           
+            {
-
+                dbe.NewFile = true;
+                dbe.Previous = "";
+            }
             dbe.CreationTimeUtc = creationTime;
-      
+
             return dbe;
         }
 
         // Find all the non-existent entries and remove them.
-        private void PruneFiles()
+        private void PruneFilesWithLock()
         {
             lock (_locker)
             {
-                foreach(var entry in Entries.ToList())
+                foreach (var entry in Entries.ToList())
                 {
-                    FileInfo fi = new FileInfo(entry.FileName);
+                    var fi = new FileInfo(entry.FileName);
                     if (!fi.Exists)
                         Entries.Remove(entry);
                 }
@@ -127,27 +122,28 @@ namespace TimberWinR.Inputs
         {
             dbe.ProcessedFile = processedFile;
             dbe.LogFileExists = File.Exists(dbe.FileName);
-            Instance.UpdateEntry(dbe, lastOffset);           
+            Instance.UpdateEntryWithLock(dbe, lastOffset);
         }
 
         public static void Roll(LogsFileDatabaseEntry dbe)
         {
             dbe.ProcessedFile = false;
-            dbe.LastPosition = 0;          
+            dbe.LastPosition = 0;
-            Instance.UpdateEntry(dbe, 0);
+            dbe.Previous = "";
-            dbe.NewFile = true;           
+            Instance.UpdateEntryWithLock(dbe, 0);
+            dbe.NewFile = true;
         }
 
-        private void UpdateEntry(LogsFileDatabaseEntry dbe, long lastOffset)
+        private void UpdateEntryWithLock(LogsFileDatabaseEntry dbe, long lastOffset)
         {
             lock (_locker)
             {
-                var fi = new FileInfo(dbe.FileName);                
+                var fi = new FileInfo(dbe.FileName);
                 dbe.NewFile = !fi.Exists;
                 dbe.CreationTimeUtc = fi.CreationTimeUtc;
-                dbe.SampleTime = DateTime.UtcNow;                 
+                dbe.SampleTime = DateTime.UtcNow;
                 dbe.LastPosition = lastOffset;
-              
+
                 WriteDatabaseFileNoLock();
             }
         }
@@ -171,7 +167,7 @@ namespace TimberWinR.Inputs
                         if (instance.Entries == null)
                             instance.Entries = new List<LogsFileDatabaseEntry>();
 
-                        instance.PruneFiles();
+                        instance.PruneFilesWithLock();
                     }
                 }
                 return instance;
@@ -191,18 +187,17 @@ namespace TimberWinR.Inputs
             }
             catch (Exception ex)
             {
-                LogManager.GetCurrentClassLogger()
+                LogManager.GetCurrentClassLogger().Error("Error reading database '{0}': {1}", DatabaseFileName, ex.ToString());
-                    .Error("Error reading database '{0}': {1}", DatabaseFileName, ex.ToString());
                 try
                 {
                     if (File.Exists(DatabaseFileName))
                         File.Delete(DatabaseFileName);
-                    LogManager.GetCurrentClassLogger().Info("Creating New Database '{0}'", DatabaseFileName);
+                    LogManager.GetCurrentClassLogger().Error("Creating New Database '{0}'", DatabaseFileName);
                     WriteDatabaseLock();
                 }
                 catch (Exception ex2)
                 {
-                    LogManager.GetCurrentClassLogger().Info("Error Creating New Database '{0}': {1}", DatabaseFileName, ex2.ToString());
+                    LogManager.GetCurrentClassLogger().Error("Error Creating New Database '{0}': {1}", DatabaseFileName, ex2.ToString());
                 }
             }
         }
@@ -254,7 +249,7 @@ namespace TimberWinR.Inputs
         public bool NewFile { get; set; }
         public bool ProcessedFile { get; set; }
         public bool LogFileExists { get; set; }
-        public string FileName { get; set; }       
+        public string FileName { get; set; }
         public DateTime CreationTimeUtc { get; set; }
         public DateTime SampleTime { get; set; }
         public long LastPosition { get; set; }
@@ -268,6 +263,7 @@ namespace TimberWinR.Inputs
         {
             Interlocked.Increment(ref _linesProcessed);
         }
+        public string Previous { get; set; }
     }
 
 }

TimberWinR/Inputs/TailFileListener.cs

@@ -30,18 +30,19 @@ namespace TimberWinR.Inputs
         private int _pollingIntervalInSeconds;
         private TimberWinR.Parser.TailFileArguments _arguments;
         private long _receivedMessages;
-
+        private long _errorCount;      
         private CodecArguments _codecArguments;
         private ICodec _codec;
-       
+
         public bool Stop { get; set; }
 
-        public TailFileListener(TimberWinR.Parser.TailFileArguments arguments, CancellationToken cancelToken)
+        public TailFileListener(TimberWinR.Parser.TailFileArguments arguments,
+            CancellationToken cancelToken)
             : base(cancelToken, "Win32-TailLog")
-        {
+        {         
             Stop = false;
-         
+
-            EnsureRollingCaught();    
+            EnsureRollingCaught();
 
             _codecArguments = arguments.CodecArguments;
             if (_codecArguments != null && _codecArguments.Type == CodecArguments.CodecType.multiline)
@@ -60,7 +61,9 @@ namespace TimberWinR.Inputs
 
         public override void Shutdown()
         {
-            LogManager.GetCurrentClassLogger().Info("{0}: Shutting Down {1} for {2}", Thread.CurrentThread.ManagedThreadId, InputType, _arguments.Location);
+            LogManager.GetCurrentClassLogger()
+                .Info("{0}: Shutting Down {1} for {2}", Thread.CurrentThread.ManagedThreadId, InputType,
+                    _arguments.Location);
             Stop = true;
             base.Shutdown();
         }
@@ -71,6 +74,7 @@ namespace TimberWinR.Inputs
                 new JProperty("taillog",
                     new JObject(
                         new JProperty("messages", _receivedMessages),
+                        new JProperty("errors", _errorCount),
                         new JProperty("type", InputType),
                         new JProperty("location", _arguments.Location),
                         new JProperty("logSource", _arguments.LogSource),
@@ -103,66 +107,100 @@ namespace TimberWinR.Inputs
 
         private void TailFileContents(string fileName, long offset, LogsFileDatabaseEntry dbe)
         {
-            using (StreamReader reader = new StreamReader(new FileStream(fileName, FileMode.Open, FileAccess.Read, FileShare.ReadWrite)))
+            const int bufSize = 16535;
+            long prevLen = offset;
+
+            var fi = new FileInfo(fileName);
+            if (!fi.Exists)
+                return;
+
+            LogManager.GetCurrentClassLogger().Trace(":{0} Tailing File: {1} as Pos: {2}", Thread.CurrentThread.ManagedThreadId, fileName, prevLen);
+
+            using (var stream = new FileStream(fi.FullName, FileMode.Open, FileAccess.Read, FileShare.Delete | FileShare.ReadWrite))
             {
-                //start at the end of the file
+                stream.Seek(prevLen, SeekOrigin.Begin);
-                long lastMaxOffset = offset;
 
-                //if the file size has not changed, idle
+                var buffer = new char[bufSize];
-                if (reader.BaseStream.Length == lastMaxOffset)
+                var current = new StringBuilder();
-                    return;
+                using (var sr = new StreamReader(stream))
-
-                //seek to the last max offset
-                LogManager.GetCurrentClassLogger().Trace("{0}: File: {1} Seek to: {2}", Thread.CurrentThread.ManagedThreadId, fileName, lastMaxOffset);
-
-                reader.BaseStream.Seek(lastMaxOffset, SeekOrigin.Begin);
-
-                //read out of the file until the EOF
-                string line = "";
-                long lineOffset = 0;
-                while ((line = reader.ReadLine()) != null)
                 {
-                    if (string.IsNullOrEmpty(line))
+                    int nRead;
-                        continue;
+                    do
-
-                    long index = lastMaxOffset + lineOffset;
-                    string text = line;
-                    string logFileName = fileName;
-                    var json = new JObject();
-
-                    if (json["logSource"] == null)
                     {
-                        if (string.IsNullOrEmpty(_arguments.LogSource))
+                        // Read a buffered amount
-                            json.Add(new JProperty("logSource", fileName));
+                        nRead = sr.ReadBlock(buffer, 0, bufSize);
-                        else
+                        for (int i = 0; i < nRead; ++i)
-                            json.Add(new JProperty("logSource", _arguments.LogSource));
+                        {
+                            // We need the terminator!
+                            if (buffer[i] == '\n' || buffer[i] == '\r')
+                            {
+                                if (current.Length > 0)
+                                {
+                                    string line = string.Concat(dbe.Previous, current);
+                                    var json = new JObject();
+
+                                    if (json["logSource"] == null)
+                                    {
+                                        if (string.IsNullOrEmpty(_arguments.LogSource))
+                                            json.Add(new JProperty("logSource", fileName));
+                                        else
+                                            json.Add(new JProperty("logSource", _arguments.LogSource));
+                                    }
+
+                                    //LogManager.GetCurrentClassLogger().Debug(":{0} File: {1}:{2}  {3}", Thread.CurrentThread.ManagedThreadId, fileName, dbe.LinesProcessed, line);
+
+                                    // We've processed the partial input
+                                    dbe.Previous = "";
+                                    json["Text"] = line;
+                                    json["Index"] = dbe.LinesProcessed;
+                                    json["LogFileName"] = fileName;
+                                    if (_codecArguments != null && _codecArguments.Type == CodecArguments.CodecType.multiline)
+                                    {
+                                        try
+                                        {                                         
+                                            _codec.Apply(line, this);
+                                            Interlocked.Increment(ref _receivedMessages);
+                                            dbe.IncrementLineCount();
+                                        }
+                                        catch (Exception ex)
+                                        {
+                                            Interlocked.Increment(ref _errorCount);
+                                            LogManager.GetCurrentClassLogger().ErrorException("Filter Error", ex);
+                                        }
+                                    }
+                                    else
+                                    {
+                                        try
+                                        {
+                                            ProcessJson(json);
+                                            dbe.IncrementLineCount();                                                                                      
+                                            Interlocked.Increment(ref _receivedMessages);
+                                            LogsFileDatabase.Update(dbe, true, sr.BaseStream.Position);
+                                        }
+                                        catch (Exception ex)
+                                        {
+                                            Interlocked.Increment(ref _errorCount);
+                                            LogManager.GetCurrentClassLogger().ErrorException("Process Error", ex);
+                                        }
+                                    }
+
+                                }
+                                current = new StringBuilder();
+                            }
+                            else // Copy character into the buffer
+                            {
+                                current.Append(buffer[i]);
+                            }
+                        }
+                    } while (nRead > 0);
+
+                    // We didn't encounter the newline, so save it.
+                    if (current.Length > 0)
+                    {
+                        dbe.Previous = current.ToString();
                     }
-
-                    json["Text"] = line;
-                    json["Index"] = index;
-                    json["LogFileName"] = fileName;
-
-
-                    if (_codecArguments != null && _codecArguments.Type == CodecArguments.CodecType.multiline)
-                    {
-                        _codec.Apply(line, this);
-                        Interlocked.Increment(ref _receivedMessages);
-                        dbe.IncrementLineCount();
-                    }
-                    else
-                    {
-                        ProcessJson(json);
-                        Interlocked.Increment(ref _receivedMessages);
-                        dbe.IncrementLineCount();
-                        //LogManager.GetCurrentClassLogger().Info("{0}: File: {1} {2} {3}", Thread.CurrentThread.ManagedThreadId, fileName, dbe.LinesProcessed, line);
-                    }                 
-
-                    lineOffset += line.Length;
                 }
-                //update the last max offset
+            }         
-                lastMaxOffset = reader.BaseStream.Position;
-                LogsFileDatabase.Update(dbe, true, lastMaxOffset);               
-            }
         }
         // One thread for each kind of file to watch, i.e. "*.log,*.txt" would be two separate
         // threads.
@@ -202,12 +240,13 @@ namespace TimberWinR.Inputs
                                 {
                                     LogManager.GetCurrentClassLogger().Debug(":{0} Starting Thread Tailing File: {1}", Thread.CurrentThread.ManagedThreadId, dbe.FileName);
                                     LogsFileDatabase.Update(dbe, false, dbe.LastPosition);
-                                    SaveVisitedFileName(fileName);
+
                                     Task.Factory.StartNew(() => TailFileWatcher(fileName));
                                 }
                                 else if (!isWildcardPattern)
                                 {
                                     FileInfo fi = new FileInfo(dbe.FileName);
+                                    SaveVisitedFileName(fileName);
 
                                     //LogManager.GetCurrentClassLogger().Info("Located File: {0}, New: {1}", dbe.FileName, dbe.NewFile);                                
                                     long length = fi.Length;
@@ -219,7 +258,7 @@ namespace TimberWinR.Inputs
                                         LogsFileDatabase.Roll(dbe);
                                     }
                                     // Log has rolled or this is a file we are seeing for the first time.
-                                    bool processWholeFile = logHasRolled || !dbe.ProcessedFile;
+                                    bool processWholeFile = logHasRolled || !dbe.ProcessedFile || dbe.NewFile;
                                     if (processWholeFile)
                                     {
                                         LogsFileDatabase.Update(dbe, true, 0);
@@ -229,7 +268,7 @@ namespace TimberWinR.Inputs
                                     else
                                     {
                                         TailFileContents(dbe.FileName, dbe.LastPosition, dbe);
-                                    }                                  
+                                    }
                                 }
                             }
                         }
@@ -273,7 +312,7 @@ namespace TimberWinR.Inputs
                 }
             }
             Finished();
-        }       
+        }
     }
 }
 

TimberWinR/Inputs/UdpInputListener.cs

@@ -1,48 +1,53 @@
 using System;
-using System.IO;
+using System.Collections.Concurrent;
 using System.Text;
-using System.Text.RegularExpressions;
 using System.Threading;
 using System.Net;
 using System.Net.Sockets;
-using Newtonsoft.Json;
 using Newtonsoft.Json.Linq;
 using NLog;
 
 namespace TimberWinR.Inputs
 {
     public class UdpInputListener : InputListener
-    {      
+    {
-        private UdpClient _udpListenerV4;      
+        private UdpClient _udpListenerV4;
-        private readonly Thread _listenThreadV4;
+        private IPEndPoint _udpEndpointV4;
-
+        private readonly BlockingCollection<byte[]> _unprocessedRawData;
+        private readonly Thread _rawDataProcessingThread;
         private readonly int _port;
         private long _receivedMessages;
-        private long _parsedErrors;
+        private long _parseErrors;
-       
+        private long _receiveErrors;
+        private long _parsedMessages;
+
         public override JObject ToJson()
         {
-            JObject json = new JObject(
+            var json =
-                new JProperty("udp",
+                new JObject(new JProperty("udp",
-                    new JObject(
+                    new JObject(new JProperty("port", _port),
-                        new JProperty("port", _port),
+                        new JProperty("receive_errors", _receiveErrors),
-                        new JProperty("errors", _parsedErrors),
+                        new JProperty("parse_errors", _parseErrors),
-                        new JProperty("messages", _receivedMessages)
+                        new JProperty("messages", _receivedMessages),
-                        )));
+                        new JProperty("parsed_messages", _parsedMessages),
+                        new JProperty("unprocessed_messages", _unprocessedRawData.Count))));
 
             return json;
         }
 
-        public UdpInputListener(CancellationToken cancelToken, int port = 5140)
+        public UdpInputListener(CancellationToken cancelToken, int port = 5140) : base(cancelToken, "Win32-Udp")
-            : base(cancelToken, "Win32-Udp")
         {
-            _port = port;     
+            _port = port;
             _receivedMessages = 0;
 
-            _listenThreadV4 = new Thread(StartListener);
+            // setup raw data processor
-            _listenThreadV4.Start();
+            _unprocessedRawData = new BlockingCollection<byte[]>();
-        }
+            _rawDataProcessingThread = new Thread(ProcessDataLoop) { Name = "Win32-Udp-DataProcessor"};
+            _rawDataProcessingThread.Start();
 
+            // start listing to udp port
+            StartListener();
+        }
 
         public override void Shutdown()
         {
@@ -51,64 +56,115 @@ namespace TimberWinR.Inputs
             // close UDP listeners, which will end the listener threads          
             _udpListenerV4.Close();
 
-            // wait for completion of the threads         
-            _listenThreadV4.Join();          
-
             base.Shutdown();
         }
 
         private void StartListener()
-        {                      
+        {
-            var groupV4 = new IPEndPoint(IPAddress.Any, _port);
+            _udpEndpointV4 = new IPEndPoint(IPAddress.Any, _port);
 
-            _udpListenerV4 = new UdpClient(_port);            
+            // setup listener
-    
+            _udpListenerV4 = new UdpClient(_port);
-            LogManager.GetCurrentClassLogger().Info("Udp Input on Port {0} Ready", groupV4);
+
+            // start listening on UDP port
+            StartReceiving();
+
+            // all started; log details
+            LogManager.GetCurrentClassLogger().Info("Udp Input on Port {0} Ready", _udpEndpointV4);
+        }
+
+        private void StartReceiving()
+        {
+            if (!CancelToken.IsCancellationRequested)
+                _udpListenerV4.BeginReceive(DataReceived, null);
+        }
+
+        private void DataReceived(IAsyncResult result)
+        {
+            if (CancelToken.IsCancellationRequested)
+            {
+                _unprocessedRawData.CompleteAdding();
+                return;
+            }
 
-            string lastMessage = "";
             try
             {
-                while (!CancelToken.IsCancellationRequested)
+                byte[] bytes = _udpListenerV4.EndReceive(result, ref _udpEndpointV4);
-                {
+                Interlocked.Increment(ref _receivedMessages);
-                    try
+                StartReceiving();
-                    {
+                _unprocessedRawData.Add(bytes);
-                        byte[] bytes = _udpListenerV4.Receive(ref groupV4);  
+            }
-                        var data = Encoding.UTF8.GetString(bytes, 0, bytes.Length);                                                
+            catch (SocketException)
-                        lastMessage = data;
+            {
-                        var json = JObject.Parse(data);
+                LogManager.GetCurrentClassLogger().Info("Socked exception. Ending UDP Listener.");
-                        ProcessJson(json);
+                _unprocessedRawData.CompleteAdding();
-                        Interlocked.Increment(ref _receivedMessages);                       
+            }
-                    }
+            catch (ObjectDisposedException)
-                    catch(ArgumentException aex)
+            {
-                    {
+                LogManager.GetCurrentClassLogger().Info("Object disposed. Ending UDP Listener");
-                        LogManager.GetCurrentClassLogger().Error(aex);
+                _unprocessedRawData.CompleteAdding();
-                        break;
-                    }
-                    catch(SocketException)
-                    {                       
-                        break;
-                    }
-                    catch (Exception ex)
-                    {                      
-                        var jex1 = LogErrors.LogException(string.Format("Invalid JSON: {0}", lastMessage), ex);
-                        if (jex1 != null)
-                            ProcessJson(jex1);
-      
-                        LogManager.GetCurrentClassLogger().Warn("Bad JSON: {0}", lastMessage);
-                        LogManager.GetCurrentClassLogger().Warn(ex);
-
-                        Interlocked.Increment(ref _parsedErrors);
-                    }
-                }
-                _udpListenerV4.Close();
             }
             catch (Exception ex)
             {
-                if (!CancelToken.IsCancellationRequested)
+                LogManager.GetCurrentClassLogger().Warn("Error while receiving data.", ex);
-                    LogManager.GetCurrentClassLogger().Error(ex);
+
+                Interlocked.Increment(ref _receiveErrors);
+                StartReceiving();
+            }
+        }
+
+        private void ProcessDataLoop()
+        {           
+            while (!_unprocessedRawData.IsCompleted)
+            {
+                try
+                {
+                    ProcessData(_unprocessedRawData.Take());
+                }
+                catch (OperationCanceledException)
+                {
+                    // we are shutting down.
+                    break;
+                }
+                catch (InvalidOperationException)
+                {
+                    // when the collection is marked as completed
+                    break;
+                }
+                catch (Exception ex)
+                {
+                    LogManager.GetCurrentClassLogger().ErrorException("Error while processing data", ex);
+                    Thread.Sleep(100);
+                }
             }
 
             Finished();
         }
+
+        private void ProcessData(byte[] bytes)
+        {
+            var data = Encoding.UTF8.GetString(bytes, 0, bytes.Length);
+
+            try
+            {
+                var json = JObject.Parse(data);
+                ProcessJson(json);
+
+                _parsedMessages++;
+            }
+            catch (Exception ex)
+            {
+                var jex1 = LogErrors.LogException(string.Format("Invalid JSON: {0}", data), ex);
+                if (jex1 != null)
+                {
+                    ProcessJson(jex1);
+                }
+
+                var msg = string.Format("Bad JSON: {0}", data);
+                LogManager.GetCurrentClassLogger().Warn(msg, ex);
+
+                _parseErrors++;
+            }
+        }
     }
 }

TimberWinR/Manager.cs

@@ -22,9 +22,7 @@ namespace TimberWinR
     public class Manager
     {
         public Configuration Config { get; set; }
-        public List<OutputSender> Outputs { get; set; }
+        public List<OutputSender> Outputs { get; set; }       
-        public List<TcpInputListener> Tcps { get; set; }
-        public List<TcpInputListener> Udps { get; set; }
         public List<InputListener> Listeners { get; set; }
         public bool LiveMonitor { get; set; }
 
@@ -66,13 +64,13 @@ namespace TimberWinR
 
         public Manager()
         {
-            LogsFileDatabase.Manager = this;                   
+            LogsFileDatabase.Manager = this;
         }
 
         public Manager(string jsonConfigFile, string logLevel, string logfileDir, bool liveMonitor, CancellationToken cancelToken, bool processConfiguration = true)
         {
-            LogsFileDatabase.Manager = this;           
+            LogsFileDatabase.Manager = this;
-  
+
             StartedOn = DateTime.UtcNow;
             LiveMonitor = liveMonitor;
 
@@ -119,17 +117,9 @@ namespace TimberWinR
 
             try
             {
-                // Is it a directory?
+                var fi = new FileInfo(jsonConfigFile);
-                if (Directory.Exists(jsonConfigFile))
+                if (fi.Exists)
                 {
-                    DirectoryInfo di = new DirectoryInfo(jsonConfigFile);
-                    LogManager.GetCurrentClassLogger().Info("Initialized, Reading Configurations From {0}", di.FullName);
-                    Config = Configuration.FromDirectory(jsonConfigFile, cancelToken, this);
-                }
-                else
-                {
-                    var fi = new FileInfo(jsonConfigFile);
-
                     LogManager.GetCurrentClassLogger().Info("Initialized, Reading Configurations From File: {0}", fi.FullName);
 
                     if (!fi.Exists)
@@ -138,6 +128,12 @@ namespace TimberWinR
                     LogManager.GetCurrentClassLogger().Info("Initialized, Reading Config: {0}", fi.FullName);
                     Config = Configuration.FromFile(jsonConfigFile);
                 }
+                else if (Directory.Exists(jsonConfigFile))
+                {
+                    DirectoryInfo di = new DirectoryInfo(jsonConfigFile);
+                    LogManager.GetCurrentClassLogger().Info("Initialized, Reading Configurations From {0}", di.FullName);
+                    Config = Configuration.FromDirectory(jsonConfigFile, cancelToken, this);
+                }
             }
             catch (JsonSerializationException jse)
             {
@@ -145,30 +141,30 @@ namespace TimberWinR
             }
             catch (Exception ex)
             {
-                LogManager.GetCurrentClassLogger().Error(ex);               
+                LogManager.GetCurrentClassLogger().Error(ex);
             }
             LogManager.GetCurrentClassLogger().Info("Log Directory {0}", logfileDir);
             LogManager.GetCurrentClassLogger().Info("Logging Level: {0}", LogManager.GlobalThreshold);
 
             if (processConfiguration)
             {
-                ProcessConfiguration(cancelToken, Config);               
+                ProcessConfiguration(cancelToken, Config);
             }
         }
 
         public void Start(CancellationToken cancelToken)
         {
-            ProcessConfiguration(cancelToken, Config);          
+            ProcessConfiguration(cancelToken, Config);
         }
 
         public void ProcessConfiguration(CancellationToken cancelToken, Configuration config)
         {
-// Read the Configuration file
+            // Read the Configuration file
             if (config != null)
             {
                 if (OnConfigurationProcessed != null)
                     OnConfigurationProcessed(config);
-  
+
                 if (config.RedisOutputs != null)
                 {
                     foreach (var ro in config.RedisOutputs)
@@ -258,6 +254,15 @@ namespace TimberWinR
                         output.Connect(elistner);
                 }
 
+                foreach (var stdin in config.Generators)
+                {
+                    var elistner = new GeneratorInput(stdin, cancelToken);
+                    Listeners.Add(elistner);
+                    foreach (var output in Outputs)
+                        output.Connect(elistner);
+                }
+
+
                 var computerName = System.Environment.MachineName + "." +
                                    Microsoft.Win32.Registry.LocalMachine.OpenSubKey(
                                        @"SYSTEM\CurrentControlSet\services\Tcpip\Parameters")
@@ -272,7 +277,7 @@ namespace TimberWinR
                             new JObject(
                                 new JProperty("version",
                                     Assembly.GetEntryAssembly().GetName().Version.ToString()),
-                                    //GetAssemblyByName("TimberWinR.ServiceHost").GetName().Version.ToString()),
+                        //GetAssemblyByName("TimberWinR.ServiceHost").GetName().Version.ToString()),
                                 new JProperty("host", computerName),
                                 new JProperty("output", output.Name),
                                 new JProperty("initialized", DateTime.UtcNow)
@@ -280,7 +285,7 @@ namespace TimberWinR
                     json.Add(new JProperty("type", "Win32-TimberWinR"));
                     json.Add(new JProperty("host", computerName));
                     output.Startup(json);
-                }              
+                }
             }
         }
 

TimberWinR/Outputs/Elasticsearch.cs

@@ -41,7 +41,10 @@ namespace TimberWinR.Outputs
         private long _sentMessages;
         private long _errorCount;
         private readonly int _maxQueueSize;
-        private readonly bool _queueOverflowDiscardOldest;    
+        private readonly bool _queueOverflowDiscardOldest;
+        private readonly bool _disablePing;
+        private readonly int _pingTimeout;
+
         private Parser.ElasticsearchOutputParameters _parameters;
         public bool Stop { get; set; }
       
@@ -61,6 +64,11 @@ namespace TimberWinR.Outputs
             var settings = new ConnectionSettings(pool)
                 .ExposeRawResponse();
 
+            if (_disablePing)
+                settings.DisablePing();
+            else if (_pingTimeout != 0)
+                settings.SetPingTimeout(_pingTimeout);
+
             var client = new ElasticClient(settings);
             return client;
         }
@@ -70,7 +78,7 @@ namespace TimberWinR.Outputs
         {
             _sentMessages = 0;
             _errorCount = 0;
-
+           
             _parameters = parameters;
             _flushSize = parameters.FlushSize;
             _idleFlushTimeSeconds = parameters.IdleFlushTimeInSeconds;
@@ -84,7 +92,8 @@ namespace TimberWinR.Outputs
             _numThreads = parameters.NumThreads;
             _maxQueueSize = parameters.MaxQueueSize;
             _queueOverflowDiscardOldest = parameters.QueueOverflowDiscardOldest;
-      
+            _disablePing = !parameters.EnablePing;
+            _pingTimeout = parameters.PingTimeout;
 
             for (int i = 0; i < parameters.NumThreads; i++)
             {
@@ -99,7 +108,7 @@ namespace TimberWinR.Outputs
                     new JObject(
                         new JProperty("host", string.Join(",", _hosts)),
                         new JProperty("errors", _errorCount),
-                        new JProperty("sentMmessageCount", _sentMessages),
+                        new JProperty("messages", _sentMessages),
                         new JProperty("queuedMessageCount", _jsonQueue.Count),
                         new JProperty("port", _port),
                         new JProperty("flushSize", _flushSize),                                       
@@ -123,6 +132,10 @@ namespace TimberWinR.Outputs
             // Force an inital flush
             DateTime lastFlushTime = DateTime.MinValue;
 
+            LogManager.GetCurrentClassLogger()
+                            .Info("{0}: Elasticsarch Output To {1} Ready", Thread.CurrentThread.ManagedThreadId, string.Join(",", _hosts));
+         
+
             using (var syncHandle = new ManualResetEventSlim())
             {               
                 // Execute the query
@@ -188,7 +201,7 @@ namespace TimberWinR.Outputs
                             }
                             GC.Collect();
                             if (!Stop)
-                            {
+                            {                               
                                 syncHandle.Wait(TimeSpan.FromMilliseconds(_interval), CancelToken);
                             }
                         }
@@ -203,6 +216,10 @@ namespace TimberWinR.Outputs
                     }
                 }
             }
+
+            LogManager.GetCurrentClassLogger()
+                          .Info("{0}: Elasticsarch Output To {1} Terminated", Thread.CurrentThread.ManagedThreadId, string.Join(",", _hosts));         
+
         }
 
         //

TimberWinR/Parser.cs

@@ -263,12 +263,46 @@ namespace TimberWinR.Parser
         }
     }
 
+    public class GeneratorParameters : IValidateSchema
+    {
+        [JsonProperty(PropertyName = "type")]
+        public string Type { get; set; }
+
+        [JsonProperty(PropertyName = "codec")]
+        public CodecArguments CodecArguments { get; set; }       
+
+        [JsonProperty(PropertyName = "message")]
+        public string Message { get; set; }
+
+        [JsonProperty(PropertyName = "count")]
+        public int Count { get; set; }
+
+        [JsonProperty(PropertyName = "rate")]
+        public int Rate { get; set; }
+
+        public void Validate()
+        {                      
+        }
+
+        public GeneratorParameters()
+        {
+            Count = 0; // Infinity messages
+            Rate = 10; // Milliseconds
+            Message = "Hello, world!";
+            CodecArguments = new CodecArguments();
+            CodecArguments.Type = CodecArguments.CodecType.plain;
+        }
+    }
+
+
     public class CodecArguments
     {
         public enum CodecType
         {
             singleline,
-            multiline
+            multiline,
+            json,
+            plain
         };
 
         public enum WhatType
@@ -519,7 +553,11 @@ namespace TimberWinR.Parser
         [JsonProperty(PropertyName = "max_queue_size")]
         public int MaxQueueSize { get; set; }
         [JsonProperty(PropertyName = "queue_overflow_discard_oldest")]
-        public bool QueueOverflowDiscardOldest { get; set; }
+        public bool QueueOverflowDiscardOldest { get; set; }        
+        [JsonProperty(PropertyName = "enable_ping")]
+        public bool EnablePing { get; set; }      
+        [JsonProperty(PropertyName = "ping_timeout")]
+        public int PingTimeout { get; set; }
 
         public ElasticsearchOutputParameters()
         {
@@ -534,6 +572,8 @@ namespace TimberWinR.Parser
             Interval = 1000;
             QueueOverflowDiscardOldest = true;
             MaxQueueSize = 50000;
+            EnablePing = false;
+            PingTimeout = 0;
         }
 
         public string GetIndexName(JObject json)
@@ -662,6 +702,9 @@ namespace TimberWinR.Parser
 
         [JsonProperty("Stdin")]
         public Stdin[] Stdins { get; set; }
+
+        [JsonProperty("Generator")]
+        public GeneratorParameters[] Generators { get; set; }
     }
 
     public partial class Grok : LogstashFilter, IValidateSchema

TimberWinR/ReleaseNotes.md

@@ -3,6 +3,20 @@
 A Native Windows to Redis/Elasticsearch Logstash Agent which runs as a service.
 
 Version / Date
+### 1.3.24.0 - 2015-04-29
+1. Fixed potential bug in TailFiles when tailing log files which are partially flushed
+   to disk, it now will not process the line until the \r\n has been seen.
+2. Added Generator input.
+
+### 1.3.23.0 - 2015-04-23
+1. Fixed bug with parsing a single json config file, rather than reading
+   JSON files from a directory.
+2. Diabled elasticsearch outputter ping by default and parameterized the ping capability.
+
+### 1.3.22.0 - 2015-04-14
+1. Fixed minor bug with TailFiles and service re-starts not picking up
+   rolled files right away.
+
 ### 1.3.21.0 - 2015-04-13
 1. Rolled Udp listener support to V4 only, too many issues with dual mode sockets
    and hosts file.  If we want to add this back, I will add a Udpv6 input.

TimberWinR/TimberWinR.csproj

@@ -82,7 +82,9 @@
     </Reference>
   </ItemGroup>
   <ItemGroup>
+    <Compile Include="Codecs\JsonCodec.cs" />
     <Compile Include="Codecs\Multiline.cs" />
+    <Compile Include="Codecs\PlainCodec.cs" />
     <Compile Include="Configuration.cs" />
     <Compile Include="ConfigurationErrors.cs" />
     <Compile Include="Diagnostics\Diagnostics.cs" />
@@ -94,6 +96,7 @@
     <Compile Include="Filters\MutateFilter.cs" />
     <Compile Include="ICodec.cs" />
     <Compile Include="Inputs\FieldDefinitions.cs" />
+    <Compile Include="Inputs\GeneratorInput.cs" />
     <Compile Include="Inputs\IISW3CRowReader.cs" />
     <Compile Include="Inputs\LogsFileDatabase.cs" />
     <Compile Include="Inputs\TailFileListener.cs" />
@@ -137,6 +140,7 @@
     <None Include="mdocs\DateFilter.md" />
     <None Include="mdocs\Filters.md" />
     <None Include="mdocs\GeoIPFilter.md" />
+    <None Include="mdocs\Generator.md" />
     <None Include="mdocs\TailFiles.md" />
     <None Include="mdocs\UdpInput.md" />
     <None Include="mdocs\W3CInput.md" />

TimberWinR/mdocs/ElasticsearchOutput.md

@@ -16,6 +16,8 @@ The following parameters are allowed when configuring the Redis output.
 | *port*                          | integer  | Elasticsearch port number                                   | This port must be open  | 9200  |
 | *queue_overflow_discard_oldest* | bool  | If true, discard oldest messages when max_queue_size reached otherwise discard newest |  | true |
 | *threads*                       | [string]    | Number of Threads                         | Number of worker threads processing messages | 1 |
+| *enable_ping* | bool  | If true, pings the server to test for keep alive |  | false |
+| *ping_timeout* | integer  | Default ping timeout when enable_ping is true | milliseconds | 200 |
 
 ### Index parameter
 If you want to output your data everyday to a new index, use following index format: "index-%{yyyy.MM.dd}". Here date format could be any forwat which you need.

TimberWinR/mdocs/Generator.md

@@ -0,0 +1,37 @@
+# Input: Generator
+
+The Generator input can be used to Generate log files for test purposes.
+
+## Parameters
+The following parameters are allowed when configuring the test log Generator.
+
+| Parameter         |     Type       |  Description                                                             | Details               |  Default |
+| :---------------- |:---------------| :----------------------------------------------------------------------- | :---------------------------  | :-- |
+| *type*            | string  |Message type                                |  |  Win32-InputGen   |
+| *message*         | string  |Message format to send                                  |  |  Hello, World!   |
+| *count*           | integer |Number of messages to generate                                       | 0 - Infinite, otherwise that number | 0 |
+| *rate*            | integer |Sleep time between generated messages                                    | Milliseconds | 10 |
+| [codec](https://github.com/Cimpress-MCP/TimberWinR/blob/master/TimberWinR/mdocs/Codec.md)  | object | Codec to use  |
+
+Example: Generate 100000 "Hello Win32-InputGen" messages
+
+```json
+{
+    "TimberWinR": {
+        "Inputs": {
+            "Generator": [
+                {
+                   "message":  "Hello %{type}",
+                   "count": 100000
+                }
+            ]
+		}
+	}
+}
+```
+## Fields
+After a successful parse of the generated line, the following fields are added:
+
+| Name | Type | Description |
+| ---- |:-----| :-----------|
+| Message | STRING | Text line content  |

TimberWix/Product.wxs

@@ -49,6 +49,8 @@
         <File Id="TimberWinR.ServiceHost.exe.config" Source="$(var.TimberWinR.ServiceHost.TargetDir)\TimberWinR.ServiceHost.exe.config" />
         <File Id="Interop.MSUtil.dll" Source="$(var.TimberWinR.ServiceHost.TargetDir)\Interop.MSUtil.dll"  />
         <File Id="csredis.dll" Source="$(var.TimberWinR.ServiceHost.TargetDir)\csredis.dll"  />
+        <File Id="Nest.dll" Source="$(var.TimberWinR.ServiceHost.TargetDir)\Nest.dll"  />
+        <File Id="Elasticsearch.Net.dll" Source="$(var.TimberWinR.ServiceHost.TargetDir)\Elasticsearch.Net.dll"  />
         <File Id="default.json" Source="$(var.TimberWinR.ServiceHost.TargetDir)\default.json"  />
         <File Id="Newtonsoft.Json.dll" Source="$(var.TimberWinR.ServiceHost.TargetDir)\Newtonsoft.Json.dll"  />
         <File Id="Nlog.dll" Source="$(var.TimberWinR.ServiceHost.TargetDir)\Nlog.dll"  />

chocolateyInstall.ps1.template

@@ -2,7 +2,7 @@
 $fileType = 'msi'
 $silentArgs = '/quiet' 
 $scriptPath =  $(Split-Path $MyInvocation.MyCommand.Path)
-$fileFullPath = Join-Path $scriptPath 'TimberWinR-${version}'
+$fileFullPath = Join-Path $scriptPath 'TimberWinR-${version}.0.msi'
 try { 
   Install-ChocolateyInstallPackage $packageName $fileType $silentArgs $fileFullPath
 } catch {

chocolateyUninstall.ps1.guid

@@ -1,8 +1,7 @@
 $packageName = 'TimberWinR-${version}' # arbitrary name for the package, used in messages
 $installerType = 'msi' #only one of these: exe, msi, msu
-$url = 'http://www.ericfontana.com/TimberWinR/TimberWinR-${version}.0.msi' # download url
+$scriptPath =  $(Split-Path $MyInvocation.MyCommand.Path)
+$fileFullPath = Join-Path $scriptPath 'TimberWinR-${version}.0.msi'
 $silentArgs = '${PROJECTGUID} /quiet' 
 $validExitCodes = @(0) #please insert other valid exit codes here, exit codes for ms http://msdn.microsoft.com/en-us/library/aa368542(VS.85).aspx
-UnInstall-ChocolateyPackage "$packageName" "$installerType" "$silentArgs" "$url" -validExitCodes $validExitCodes
+UnInstall-ChocolateyPackage "$packageName" "$installerType" "$silentArgs" "fileFullPath" -validExitCodes $validExitCodes
-
-

chocolateyUninstall.ps1.template

@@ -1,9 +1,7 @@
 $packageName = 'TimberWinR-${version}' # arbitrary name for the package, used in messages
 $installerType = 'msi' #only one of these: exe, msi, msu
 $scriptPath =  $(Split-Path $MyInvocation.MyCommand.Path)
-$fileFullPath = Join-Path $scriptPath 'TimberWinR-${version}'
+$fileFullPath = Join-Path $scriptPath 'TimberWinR-${version}.0.msi'
-$silentArgs = '${PROJECTGUID} /quiet' 
+$silentArgs = '{CC4DF908-07C4-4BD8-A9FA-6E6AC315E30B} /quiet' 
 $validExitCodes = @(0) #please insert other valid exit codes here, exit codes for ms http://msdn.microsoft.com/en-us/library/aa368542(VS.85).aspx
 UnInstall-ChocolateyPackage "$packageName" "$installerType" "$silentArgs" "fileFullPath" -validExitCodes $validExitCodes
-
-

chocolateyUninstall.ps1.template.orig

@@ -1,7 +1,7 @@
 $packageName = 'TimberWinR-${version}' # arbitrary name for the package, used in messages
 $installerType = 'msi' #only one of these: exe, msi, msu
 $scriptPath =  $(Split-Path $MyInvocation.MyCommand.Path)
-$fileFullPath = Join-Path $scriptPath 'TimberWinR-${version}'
+$fileFullPath = Join-Path $scriptPath 'TimberWinR-${version}.0.msi'
 $silentArgs = '${PROJECTGUID} /quiet' 
 $validExitCodes = @(0) #please insert other valid exit codes here, exit codes for ms http://msdn.microsoft.com/en-us/library/aa368542(VS.85).aspx
 UnInstall-ChocolateyPackage "$packageName" "$installerType" "$silentArgs" "fileFullPath" -validExitCodes $validExitCodes