Merge pull request #2 from YPermitin/develop

Реализован общий буфер и оптимизация производительности

.github/workflows/dotnet.yml

@@ -16,7 +16,7 @@ jobs:
     - name: Setup .NET
       uses: actions/setup-dotnet@v1
       with:
-        dotnet-version: 3.1.301
+        dotnet-version: 5.0.202
     - name: Install ClickHouse
       run: sudo chmod +x ./Scripts/install-clickhouse.sh && ./Scripts/install-clickhouse.sh
     - name: Restore dependencies

Apps/YY.TechJournalExportAssistantConsoleApp/Program.cs

@@ -73,7 +73,6 @@ namespace YY.TechJournalExportAssistantConsoleApp
                         target.SetInformationSystem(new TechJournalLogBase()
                         {
                             Name = techJournalName,
-                            DirectoryName = tjDirectory.DirectoryData.Name,
                             Description = techJournalDescription
                         });
                         exporter.SetTarget(target);

Apps/YY.TechJournalExportAssistantConsoleApp/YY.TechJournalExportAssistantConsoleApp.csproj

@@ -2,7 +2,7 @@
 
   <PropertyGroup>
     <OutputType>Exe</OutputType>
-    <TargetFramework>netcoreapp3.1</TargetFramework>
+    <TargetFramework>net5.0</TargetFramework>
   </PropertyGroup>
 
   <ItemGroup>
@@ -15,4 +15,6 @@
     </None>
   </ItemGroup>
 
+  <ProjectExtensions><VisualStudio><UserProperties appsettings_1json__JsonSchema="" /></VisualStudio></ProjectExtensions>
+
 </Project>

Apps/YY.TechJournalExportAssistantWithSharedBufferConsoleApp/Program.cs

@@ -0,0 +1,42 @@
+using Microsoft.Extensions.Configuration;
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using YY.TechJournalExportAssistant.ClickHouse;
+using YY.TechJournalExportAssistant.Core;
+using YY.TechJournalExportAssistant.Core.SharedBuffer;
+using YY.TechJournalExportAssistant.Core.SharedBuffer.EventArgs;
+
+namespace YY.TechJournalExportAssistantWithSharedBufferConsoleApp
+{
+    class Program
+    {
+        public static IConfiguration Configuration = new ConfigurationBuilder()
+            .AddJsonFile("appsettings.json", optional: false, reloadOnChange: false)
+            .Build();
+        
+        static async Task Main(string[] args)
+        {
+            TechJournalSettings settings = TechJournalSettings.Create(Configuration);
+            
+            TechJournalExport exportMaster = new TechJournalExport(settings, new TechJournalOnClickHouseTargetBuilder());
+            exportMaster.OnErrorEvent += OnError;
+            exportMaster.OnSendLogEvent += OnSend;
+            await exportMaster.StartExport();
+
+            Console.WriteLine("Good luck & bye!");
+        }
+
+        private static void OnError(OnErrorExportSharedBufferEventArgs args)
+        {
+            Console.WriteLine($"Ошибка при экспорте логов: {args.Exception}");
+        }
+
+        private static void OnSend(OnSendLogFromSharedBufferEventArgs args)
+        {
+            Console.WriteLine($"Отправка данных в хранилище:\n" +
+                              $"Записей: {args.DataFromBuffer.Values.SelectMany(i => i.LogRows).Select(i => i.Value).Count()}\n" +
+                              $"Актуальных позиций чтения: {args.DataFromBuffer.Values.Select(i => i.LogPosition).Count() }");
+        }
+    }
+}

Apps/YY.TechJournalExportAssistantWithSharedBufferConsoleApp/YY.TechJournalExportAssistantWithSharedBufferConsoleApp.csproj

@@ -0,0 +1,26 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net5.0</TargetFramework>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Configuration" Version="5.0.0" />
+    <PackageReference Include="Microsoft.Extensions.Configuration.Binder" Version="5.0.0" />
+    <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="5.0.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\Libs\YY.TechJournalExportAssistant.ClickHouse\YY.TechJournalExportAssistant.ClickHouse.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <None Update="appsettings.json">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+  </ItemGroup>
+
+  <ProjectExtensions><VisualStudio><UserProperties appsettings_1json__JsonSchema="" /></VisualStudio></ProjectExtensions>
+
+</Project>

Libs/YY.TechJournalExportAssistant.ClickHouse/ClickHouseContext.cs

@@ -8,6 +8,7 @@ using ClickHouse.Client.Copy;
 using YY.TechJournalExportAssistant.ClickHouse.Helpers;
 using YY.TechJournalExportAssistant.Core;
 using YY.TechJournalExportAssistant.Core.Helpers;
+using YY.TechJournalExportAssistant.Core.SharedBuffer;
 using YY.TechJournalReaderAssistant;
 using YY.TechJournalReaderAssistant.Helpers;
 using YY.TechJournalReaderAssistant.Models;
@@ -19,7 +20,6 @@ namespace YY.TechJournalExportAssistant.ClickHouse
         #region Private Members
 
         private ClickHouseConnection _connection;
-        private long logFileLastId = -1;
 
         #endregion
 
@@ -31,16 +31,12 @@ namespace YY.TechJournalExportAssistant.ClickHouse
 
             _connection = new ClickHouseConnection(connectionSettings);
             _connection.Open();
-            
+
             var cmdDDL = _connection.CreateCommand();
             cmdDDL.CommandText = Resource.Query_CreateTable_EventDataStorage;
             cmdDDL.ExecuteNonQuery();
             cmdDDL.CommandText = Resource.Query_CreateTable_LogFilesStorage;
             cmdDDL.ExecuteNonQuery();
-            cmdDDL.CommandText = Resource.Query_CreateTable_EventData;
-            cmdDDL.ExecuteNonQuery();
-            cmdDDL.CommandText = Resource.Query_CreateTable_LogFiles;
-            cmdDDL.ExecuteNonQuery();
         }
 
         #endregion
@@ -49,111 +45,243 @@ namespace YY.TechJournalExportAssistant.ClickHouse
 
         #region RowsData
 
-        public void SaveRowsData(TechJournalLogBase techJournalLog, List<EventData> eventData, string fileName)
+        public void SaveRowsData(Dictionary<LogBufferItemKey, LogBufferItem> sourceDataFromBuffer)
         {
+            List<object[]> rowsForInsert = new List<object[]>();
+            List<object[]> positionsForInsert = new List<object[]>();
+            Dictionary<string, LastRowsInfoByLogFile> maxPeriodByDirectories = new Dictionary<string, LastRowsInfoByLogFile>();
+
+            var dataFromBuffer = sourceDataFromBuffer
+                .OrderBy(i => i.Key.Period)
+                .ThenBy(i => i.Value.LogPosition.EventNumber)
+                .ToList();
+            long itemNumber = 0;
+            foreach (var dataItem in dataFromBuffer)
+            {
+                itemNumber++;
+                FileInfo logFileInfo = new FileInfo(dataItem.Key.LogFile);
+
+                positionsForInsert.Add(new object[]
+                {
+                    dataItem.Key.Settings.TechJournalLog.Name,
+                    logFileInfo.Directory?.Name ?? string.Empty,
+                    DateTime.Now.Ticks + itemNumber,
+                    logFileInfo.Name,
+                    logFileInfo.CreationTimeUtc,
+                    logFileInfo.LastWriteTimeUtc,
+                    dataItem.Value.LogPosition.EventNumber,
+                    dataItem.Value.LogPosition.CurrentFileData.Replace("\\", "\\\\"),
+                    dataItem.Value.LogPosition.StreamPosition ?? 0
+                });
+
+                foreach (var rowData in dataItem.Value.LogRows)
+                {
+                    if (!maxPeriodByDirectories.TryGetValue(logFileInfo.FullName, out LastRowsInfoByLogFile lastInfo))
+                    {
+                        if (logFileInfo.Directory != null)
+                        {
+                            GetRowsDataMaxPeriodAndId(
+                                dataItem.Key.Settings.TechJournalLog,
+                                logFileInfo.Directory.Name,
+                                logFileInfo.Name,
+                                rowData.Value.Period,
+                                out var maxPeriod,
+                                out var maxId
+                            );
+                            lastInfo = new LastRowsInfoByLogFile(maxPeriod, maxId);
+                            maxPeriodByDirectories.Add(logFileInfo.FullName, lastInfo);
+                        }
+                    }
+
+                    bool existByPeriod = lastInfo.MaxPeriod > ClickHouseHelpers.MinDateTimeValue &&
+                                         rowData.Value.Period.Truncate(TimeSpan.FromSeconds(1)) <= lastInfo.MaxPeriod;
+                    bool existById = lastInfo.MaxId > 0 &&
+                                     rowData.Value.Id <= lastInfo.MaxId;
+                    if (existByPeriod && existById)
+                        continue;
+
+                    var eventItem = rowData.Value;
+                    rowsForInsert.Add(new object[]
+                        {
+                            dataItem.Key.Settings.TechJournalLog.Name,
+                            logFileInfo.Directory.Name,
+                            logFileInfo.Name,
+                            eventItem.Id,
+                            eventItem.Period,
+                            eventItem.Level,
+                            eventItem.Duration,
+                            eventItem.DurationSec,
+                            eventItem.EventName ?? string.Empty,
+                            eventItem.ServerContextName ?? string.Empty,
+                            eventItem.ProcessName ?? string.Empty,
+                            eventItem.SessionId ?? 0,
+                            eventItem.ApplicationName ?? string.Empty,
+                            eventItem.ClientId ?? 0,
+                            eventItem.ComputerName ?? string.Empty,
+                            eventItem.ConnectionId ?? 0,
+                            eventItem.UserName ?? string.Empty,
+                            eventItem.ApplicationId ?? 0,
+                            eventItem.Context ?? string.Empty,
+                            eventItem.ActionType.GetDescription() ?? string.Empty,
+                            eventItem.Database ?? string.Empty,
+                            eventItem.DatabaseCopy ?? string.Empty,
+                            eventItem.DBMS.GetPresentation() ?? string.Empty,
+                            eventItem.DatabasePID ?? string.Empty,
+                            eventItem.PlanSQLText ?? string.Empty,
+                            eventItem.Rows ?? 0,
+                            eventItem.RowsAffected ?? 0,
+                            eventItem.SQLText ?? string.Empty,
+                            eventItem.SQLQueryOnly ?? string.Empty,
+                            eventItem.SQLQueryParametersOnly ?? string.Empty,
+                            eventItem.SQLQueryHash ?? string.Empty,
+                            eventItem.SDBL ?? string.Empty,
+                            eventItem.Description ?? string.Empty,
+                            eventItem.Message ?? string.Empty,
+                            eventItem.GetCustomFieldsAsJSON() ?? string.Empty
+                        });
+                }
+            }
+
+            if (rowsForInsert.Count > 0)
+            {
+                using (ClickHouseBulkCopy bulkCopyInterface = new ClickHouseBulkCopy(_connection)
+                {
+                    DestinationTableName = "EventData",
+                    BatchSize = 100000,
+                    MaxDegreeOfParallelism = 4
+                })
+                {
+                    var bulkResult = bulkCopyInterface.WriteToServerAsync(rowsForInsert);
+                    bulkResult.Wait();
+                    rowsForInsert.Clear();
+                }
+            }
+
+            if (positionsForInsert.Count > 0)
+            {
+                using (ClickHouseBulkCopy bulkCopyInterface = new ClickHouseBulkCopy(_connection)
+                {
+                    DestinationTableName = "LogFiles",
+                    BatchSize = 100000
+                })
+                {
+                    var bulkResult = bulkCopyInterface.WriteToServerAsync(positionsForInsert);
+                    bulkResult.Wait();
+                }
+            }
+        }
+
+        public void SaveRowsData(TechJournalLogBase techJournalLog, 
+            List<EventData> eventData, 
+            string fileName,
+            Dictionary<string, DateTime> maxPeriodByFiles = null)
+        {
+            IDictionary<string, List<EventData>> eventDataToInsert = new Dictionary<string, List<EventData>>();
+            eventDataToInsert.Add(fileName, eventData);
+
+            SaveRowsData(techJournalLog, eventDataToInsert);
+        }
+        
+        public void SaveRowsData(TechJournalLogBase techJournalLog, 
+            IDictionary<string, List<EventData>> eventData,
+            Dictionary<string, LastRowsInfoByLogFile> maxPeriodByFiles = null)
+        {
+            if(maxPeriodByFiles == null) maxPeriodByFiles = new Dictionary<string, LastRowsInfoByLogFile>();
+            List<object[]> rowsForInsert = new List<object[]>();
+            foreach (var eventInfo in eventData)
+            {
+                FileInfo logFileInfo = new FileInfo(eventInfo.Key);
+                foreach (var eventItem in eventInfo.Value)
+                {
+                    if (!maxPeriodByFiles.TryGetValue(logFileInfo.Name, out LastRowsInfoByLogFile lastInfo))
+                    {
+                        if (logFileInfo.Directory != null)
+                        {
+                            GetRowsDataMaxPeriodAndId(
+                                techJournalLog,
+                                logFileInfo.Directory.Name,
+                                logFileInfo.Name,
+                                eventItem.Period,
+                                out var maxPeriod,
+                                out var maxId
+                            );
+                            lastInfo = new LastRowsInfoByLogFile(maxPeriod, maxId);
+                            maxPeriodByFiles.Add(logFileInfo.Name, lastInfo);
+                        }
+                    }
+
+                    bool existByPeriod = lastInfo.MaxPeriod > ClickHouseHelpers.MinDateTimeValue &&
+                                         eventItem.Period.Truncate(TimeSpan.FromSeconds(1)) <= lastInfo.MaxPeriod;
+                    bool existById = lastInfo.MaxId > 0 &&
+                                         eventItem.Id <= lastInfo.MaxId;
+                    if (existByPeriod && existById)
+                        continue;
+
+                    if (logFileInfo.Directory != null)
+                        rowsForInsert.Add(new object[]
+                        {
+                            techJournalLog.Name,
+                            logFileInfo.Directory.Name,
+                            logFileInfo.Name,
+                            eventItem.Id,
+                            eventItem.Period,
+                            eventItem.Level,
+                            eventItem.Duration,
+                            eventItem.DurationSec,
+                            eventItem.EventName ?? string.Empty,
+                            eventItem.ServerContextName ?? string.Empty,
+                            eventItem.ProcessName ?? string.Empty,
+                            eventItem.SessionId ?? 0,
+                            eventItem.ApplicationName ?? string.Empty,
+                            eventItem.ClientId ?? 0,
+                            eventItem.ComputerName ?? string.Empty,
+                            eventItem.ConnectionId ?? 0,
+                            eventItem.UserName ?? string.Empty,
+                            eventItem.ApplicationId ?? 0,
+                            eventItem.Context ?? string.Empty,
+                            eventItem.ActionType.GetDescription() ?? string.Empty,
+                            eventItem.Database ?? string.Empty,
+                            eventItem.DatabaseCopy ?? string.Empty,
+                            eventItem.DBMS.GetPresentation() ?? string.Empty,
+                            eventItem.DatabasePID ?? string.Empty,
+                            eventItem.PlanSQLText ?? string.Empty,
+                            eventItem.Rows ?? 0,
+                            eventItem.RowsAffected ?? 0,
+                            eventItem.SQLText ?? string.Empty,
+                            eventItem.SQLQueryOnly ?? string.Empty,
+                            eventItem.SQLQueryParametersOnly ?? string.Empty,
+                            eventItem.SQLQueryHash ?? string.Empty,
+                            eventItem.SDBL ?? string.Empty,
+                            eventItem.Description ?? string.Empty,
+                            eventItem.Message ?? string.Empty,
+                            eventItem.GetCustomFieldsAsJSON() ?? string.Empty
+                        });
+                }
+            }
+
+            if (rowsForInsert.Count == 0)
+                return;
+
             using (ClickHouseBulkCopy bulkCopyInterface = new ClickHouseBulkCopy(_connection)
             {
                 DestinationTableName = "EventData",
-                BatchSize = 100000
+                BatchSize = 100000,
+                MaxDegreeOfParallelism = 4
             })
             {
-                var values = eventData.Select(i => new object[]
-                {
-                    techJournalLog.Name,
-                    techJournalLog.DirectoryName,
-                    fileName,
-                    i.Id,
-                    i.Period,
-                    i.Level,
-                    i.Duration,
-                    i.DurationSec,
-                    i.EventName ?? string.Empty,
-                    i.ServerContextName ?? string.Empty,
-                    i.ProcessName ?? string.Empty,
-                    i.SessionId ?? 0,
-                    i.ApplicationName ?? string.Empty,
-                    i.ClientId ?? 0,
-                    i.ComputerName ?? string.Empty, 
-                    i.ConnectionId ?? 0,
-                    i.UserName ?? string.Empty,
-                    i.ApplicationId ?? 0,
-                    i.Context ?? string.Empty,
-                    i.ActionType.GetDescription() ?? string.Empty,
-                    i.Database ?? string.Empty,
-                    i.DatabaseCopy ?? string.Empty,
-                    i.DBMS.GetPresentation() ?? string.Empty,
-                    i.DatabasePID ?? string.Empty,
-                    i.PlanSQLText ?? string.Empty,
-                    i.Rows ?? 0,
-                    i.RowsAffected ?? 0,
-                    i.SQLText ?? string.Empty,
-                    i.SQLQueryOnly ?? string.Empty,
-                    i.SQLQueryParametersOnly ?? string.Empty,
-                    i.SQLQueryHash ?? string.Empty,
-                    i.SDBL?? string.Empty,
-                    i.Description?? string.Empty,
-                    i.Message?? string.Empty,
-                    i.GetCustomFieldsAsJSON() ?? string.Empty
-                }).AsEnumerable();
-
-                var bulkResult = bulkCopyInterface.WriteToServerAsync(values);
+                var bulkResult = bulkCopyInterface.WriteToServerAsync(rowsForInsert);
                 bulkResult.Wait();
+                rowsForInsert.Clear();
             }
         }
-        public DateTime GetRowsDataMaxPeriod(TechJournalLogBase techJournalLog, string fileName)
-        {
-            DateTime output = DateTime.MinValue;
-
-            using (var command = _connection.CreateCommand())
-            {
-                command.CommandText =
-                    @"SELECT
-                        MAX(Period) AS MaxPeriod
-                    FROM EventData AS RD
-                    WHERE TechJournalLog = {techJournalLog:String}
-                        AND DirectoryName = {directoryName:String}
-                        AND FileName = {fileName:String}";
-                command.AddParameterToCommand("techJournalLog", techJournalLog.Name);
-                command.AddParameterToCommand("directoryName", techJournalLog.DirectoryName);
-                command.AddParameterToCommand("fileName", fileName);
-                using (var cmdReader = command.ExecuteReader())
-                    if (cmdReader.Read()) output = cmdReader.GetDateTime(0);
-            }
-
-            return output;
-        }
-        public bool RowDataExistOnDatabase(TechJournalLogBase techJournalLog, EventData eventData, string fileName)
-        {
-            bool output = false;
-
-            using (var command = _connection.CreateCommand())
-            {
-                command.CommandText =
-                    @"SELECT
-                        TechJournalLog,
-                        Id,
-                        Period
-                    FROM EventData AS RD
-                    WHERE TechJournalLog = {techJournalLog:String}
-                        AND Id = {existId:Int64}
-                        AND FileName = {fileName:String}
-                        AND Period = {existPeriod:DateTime}";
-                command.AddParameterToCommand("techJournalLog", DbType.AnsiString, techJournalLog.Name);
-                command.AddParameterToCommand("existId", DbType.Int64, eventData.Id);
-                command.AddParameterToCommand("existPeriod", DbType.DateTime, eventData.Period);
-                command.AddParameterToCommand("fileName", DbType.String, fileName);
-                using (var cmdReader = command.ExecuteReader())
-                    if (cmdReader.Read()) output = true;
-            }
-
-            return output;
-        }
 
+        
+        
         #endregion
 
         #region LogFiles
 
-        public TechJournalPosition GetLogFilePosition(TechJournalLogBase techJournalLog)
+        public TechJournalPosition GetLogFilePosition(TechJournalLogBase techJournalLog, string directoryName)
         {
             var cmdGetLastLogFileInfo = _connection.CreateCommand();
             cmdGetLastLogFileInfo.CommandText =
@@ -172,7 +300,7 @@ namespace YY.TechJournalExportAssistant.ClickHouse
                             AND LF_LAST.DirectoryName = {directoryName:String}
                     )";
             cmdGetLastLogFileInfo.AddParameterToCommand("techJournalLog", DbType.AnsiString, techJournalLog.Name);
-            cmdGetLastLogFileInfo.AddParameterToCommand("directoryName", DbType.AnsiString, techJournalLog.DirectoryName);
+            cmdGetLastLogFileInfo.AddParameterToCommand("directoryName", DbType.AnsiString, directoryName);
 
             TechJournalPosition output = null;
             using (var cmdReader = cmdGetLastLogFileInfo.ExecuteReader())
@@ -191,75 +319,76 @@ namespace YY.TechJournalExportAssistant.ClickHouse
 
             return output;
         }
-        public void SaveLogPosition(TechJournalLogBase techJournalLog, FileInfo logFileInfo, TechJournalPosition position)
+        
+        public IDictionary<string, TechJournalPosition> GetCurrentLogPositions(
+            TechJournalLogBase techJournalLog)
         {
-            using (ClickHouseBulkCopy bulkCopyInterface = new ClickHouseBulkCopy(_connection)
-            {
-                DestinationTableName = "LogFiles", BatchSize = 100000
-            })
-            {
-                IEnumerable<object[]> values = new List<object[]>()
-                {
-                    new object[]
-                    {
-                        techJournalLog.Name,
-                        techJournalLog.DirectoryName,
-                        GetLogFileInfoNewId(techJournalLog),
-                        logFileInfo.Name,
-                        logFileInfo.CreationTimeUtc,
-                        logFileInfo.LastWriteTimeUtc,
-                        position.EventNumber,
-                        position.CurrentFileData.Replace("\\", "\\\\"),
-                        position.StreamPosition ?? 0
-                    }
-                }.AsEnumerable();
-                var bulkResult = bulkCopyInterface.WriteToServerAsync(values);
-                bulkResult.Wait();
-            }
-        }
-        public long GetLogFileInfoNewId(TechJournalLogBase techJournalLog)
-        {
-            long output = 0;
+            var cmdGetLastLogFileInfo = _connection.CreateCommand();
+            cmdGetLastLogFileInfo.CommandText = Resource.Query_GetActualPositions;
+            cmdGetLastLogFileInfo.AddParameterToCommand("techJournalLog", DbType.AnsiString, techJournalLog.Name);
 
-            if (logFileLastId < 0)
+            IDictionary<string, TechJournalPosition> output = new Dictionary<string, TechJournalPosition>();
+            using (var cmdReader = cmdGetLastLogFileInfo.ExecuteReader())
             {
-                using (var command = _connection.CreateCommand())
+                while (cmdReader.Read())
                 {
-                    command.CommandText =
-                        @"SELECT
-                            MAX(Id)
-                        FROM LogFiles
-                        WHERE TechJournalLog = {techJournalLog:String}
-                            AND DirectoryName = {directoryName:String}";
-                    command.AddParameterToCommand("techJournalLog", DbType.AnsiString, techJournalLog.Name);
-                    command.AddParameterToCommand("directoryName", DbType.AnsiString, techJournalLog.DirectoryName);
-                    using (var cmdReader = command.ExecuteReader())
-                        if (cmdReader.Read()) output = cmdReader.GetInt64(0);
+                    string fileName = cmdReader.GetString(7).Replace("\\\\", "\\");
+                    string directoryName = cmdReader.GetString(1);
+                    output.Add(directoryName, new TechJournalPosition(
+                        cmdReader.GetInt64(6),
+                        fileName,
+                        cmdReader.GetInt64(8)
+                    ));
                 }
             }
-            else output = logFileLastId;
-
-            output += 1;
-            logFileLastId = output;
 
             return output;
         }
-        public void RemoveArchiveLogFileRecords(TechJournalLogBase techJournalLog)
+        
+        public void SaveLogPosition(TechJournalLogBase techJournalLog, TechJournalPosition position)
         {
-            var commandRemoveArchiveLogInfo = _connection.CreateCommand();
-            commandRemoveArchiveLogInfo.CommandText =
-                @"ALTER TABLE LogFilesStorage DELETE
-                WHERE TechJournalLog = {techJournalLog:String}
-                    AND DirectoryName = {directoryName:String}
-                    AND Id < (
-                    SELECT MAX(Id) AS LastId
-                    FROM LogFilesStorage lf
-                    WHERE TechJournalLog = {techJournalLog:String}
-                        AND DirectoryName = {directoryName:String}
-                )";
-            commandRemoveArchiveLogInfo.AddParameterToCommand("techJournalLog", DbType.AnsiString, techJournalLog.Name);
-            commandRemoveArchiveLogInfo.AddParameterToCommand("directoryName", DbType.AnsiString, techJournalLog.DirectoryName);
-            commandRemoveArchiveLogInfo.ExecuteNonQuery();
+            SaveLogPositions(techJournalLog, new List<TechJournalPosition>()
+            {
+                position
+            });
+        }
+        
+        public void SaveLogPositions(TechJournalLogBase techJournalLog, List<TechJournalPosition> positions)
+        {
+            if(positions.Count == 0)
+                return;
+
+            long itemNumber = 0;
+            var dataForInsert = positions
+                .Select(position =>
+                {
+                    itemNumber++;
+                    var dataFileInfoForPosition = new FileInfo(position.CurrentFileData);
+
+                    return new object[]
+                    {
+                        techJournalLog.Name,
+                        dataFileInfoForPosition.Directory?.Name ?? string.Empty,
+                        DateTime.Now.Ticks + itemNumber,
+                        dataFileInfoForPosition.Name,
+                        dataFileInfoForPosition.CreationTimeUtc,
+                        dataFileInfoForPosition.LastWriteTimeUtc,
+                        position.EventNumber,
+                        position.CurrentFileData.Replace("\\", "\\\\"),
+                        position.StreamPosition ?? 0
+                    };
+                })
+                .ToList();
+
+            using (ClickHouseBulkCopy bulkCopyInterface = new ClickHouseBulkCopy(_connection)
+            {
+                DestinationTableName = "LogFiles",
+                BatchSize = 1000000
+            })
+            {
+                var bulkResult = bulkCopyInterface.WriteToServerAsync(dataForInsert);
+                bulkResult.Wait();
+            }
         }
 
         #endregion
@@ -276,5 +405,57 @@ namespace YY.TechJournalExportAssistant.ClickHouse
         }
 
         #endregion
+
+        #region Service
+
+        private void GetRowsDataMaxPeriodAndId(TechJournalLogBase techJournalLog,
+            string directoryName, string fileName, DateTime fromPeriod,
+            out DateTime maxPeriod, out long maxId)
+        {
+            DateTime outputMaxPeriod = DateTime.MinValue;
+            long outputMaxId = 0;
+
+            using (var command = _connection.CreateCommand())
+            {
+                command.CommandText =
+                    @"SELECT
+                        MAX(Period) AS MaxPeriod,
+                        MAX(Id) AS MaxId
+                    FROM EventData AS RD
+                    WHERE TechJournalLog = {techJournalLog:String}
+                        AND DirectoryName = {directoryName:String}
+                        AND FileName = {fileName:String}
+                        AND Period >= {fromPeriod:DateTime}";
+                command.AddParameterToCommand("techJournalLog", techJournalLog.Name);
+                command.AddParameterToCommand("directoryName", directoryName);
+                command.AddParameterToCommand("fileName", fileName);
+                command.AddParameterToCommand("fromPeriod", fromPeriod);
+                using (var cmdReader = command.ExecuteReader())
+                {
+                    if (cmdReader.Read())
+                    {
+                        outputMaxPeriod = cmdReader.GetDateTime(0);
+                        outputMaxId = cmdReader.GetInt64(1);
+                    }
+                }
+            }
+
+            maxPeriod = outputMaxPeriod;
+            maxId = outputMaxId;
+        }
+
+        public readonly struct LastRowsInfoByLogFile
+        {
+            public LastRowsInfoByLogFile(DateTime maxPeriod, long maxId)
+            {
+                MaxPeriod = maxPeriod;
+                MaxId = maxId;
+            }
+
+            public DateTime MaxPeriod { get; }
+            public long MaxId { get; }
+        }
+
+        #endregion
     }
 }

Libs/YY.TechJournalExportAssistant.ClickHouse/Helpers/ClickHouseHelpers.cs

@@ -9,6 +9,8 @@ namespace YY.TechJournalExportAssistant.ClickHouse.Helpers
 {
     public static class ClickHouseHelpers
     {
+        public static readonly DateTime MinDateTimeValue = new DateTime(1970, 1, 1);
+
         public static Dictionary<string, string> GetConnectionParams(string connectionString)
         {
             var connectionParams = connectionString.Split(';', StringSplitOptions.RemoveEmptyEntries)
@@ -61,9 +63,7 @@ namespace YY.TechJournalExportAssistant.ClickHouse.Helpers
 
         private static void ExecNonReaderQuery(string connectionSettings, string commandTest)
         {
-            string paramName;
-            string paramValue;
-            string databaseName = GetDatabaseName(connectionSettings, out paramName, out paramValue);
+            string databaseName = GetDatabaseName(connectionSettings, out var paramName, out var paramValue);
 
             if (databaseName != null)
             {

Libs/YY.TechJournalExportAssistant.ClickHouse/Resource.Designer.cs

@@ -61,16 +61,7 @@ namespace YY.TechJournalExportAssistant.ClickHouse {
         }
         
         /// <summary>
-        ///   Ищет локализованную строку, похожую на CREATE TABLE IF NOT EXISTS EventData AS EventDataStorage ENGINE = Buffer(currentDatabase(), EventDataStorage, 16, 10, 60, 100000, 1000000, 10000000, 100000000).
-        /// </summary>
-        internal static string Query_CreateTable_EventData {
-            get {
-                return ResourceManager.GetString("Query_CreateTable_EventData", resourceCulture);
-            }
-        }
-        
-        /// <summary>
-        ///   Ищет локализованную строку, похожую на CREATE TABLE IF NOT EXISTS EventDataStorage
+        ///   Ищет локализованную строку, похожую на CREATE TABLE IF NOT EXISTS EventData
         ///(
         ///    TechJournalLog LowCardinality(String),
         ///    DirectoryName LowCardinality(String),
@@ -82,7 +73,8 @@ namespace YY.TechJournalExportAssistant.ClickHouse {
         ///    DurationSec Int64 Codec(DoubleDelta, LZ4),
         ///    EventName LowCardinality(String),
         ///    ServerContextName LowCardinality(String),
-        ///    ProcessName LowCardinality(String), [остаток строки не уместился]&quot;;.
+        ///    ProcessName LowCardinality(String),
+        ///    S [остаток строки не уместился]&quot;;.
         /// </summary>
         internal static string Query_CreateTable_EventDataStorage {
             get {
@@ -91,20 +83,11 @@ namespace YY.TechJournalExportAssistant.ClickHouse {
         }
         
         /// <summary>
-        ///   Ищет локализованную строку, похожую на CREATE TABLE IF NOT EXISTS LogFiles AS LogFilesStorage ENGINE = Buffer(currentDatabase(), LogFilesStorage, 16, 10, 60, 100000, 1000000, 10000000, 100000000).
-        /// </summary>
-        internal static string Query_CreateTable_LogFiles {
-            get {
-                return ResourceManager.GetString("Query_CreateTable_LogFiles", resourceCulture);
-            }
-        }
-        
-        /// <summary>
-        ///   Ищет локализованную строку, похожую на CREATE TABLE IF NOT EXISTS LogFilesStorage
+        ///   Ищет локализованную строку, похожую на CREATE TABLE IF NOT EXISTS LogFiles
         ///(
         ///	TechJournalLog LowCardinality(String),
         ///	DirectoryName LowCardinality(String),
-        ///	Id Int64 Codec(DoubleDelta, LZ4),
+        ///	Id Int64 DEFAULT (toUnixTimestamp64Milli(now64())*1000000 + rowNumberInAllBlocks()) Codec(DoubleDelta, LZ4),
         ///	FileName LowCardinality(String),
         ///	CreateDate DateTime Codec(Delta, LZ4),
         ///	ModificationDate DateTime Codec(Delta, LZ4),
@@ -112,14 +95,47 @@ namespace YY.TechJournalExportAssistant.ClickHouse {
         ///	LastCurrentFileData LowCardinality(String),
         ///	LastStreamPosition Int64 Codec(DoubleDelta, LZ4)
         ///)
-        ///engine = MergeTree()
-        ///PARTITION BY toYYYYMM(CreateDate)
-        ///PRIMARY KEY CreateD [остаток строки не уместился]&quot;;.
+        ///engine = [остаток строки не уместился]&quot;;.
         /// </summary>
         internal static string Query_CreateTable_LogFilesStorage {
             get {
                 return ResourceManager.GetString("Query_CreateTable_LogFilesStorage", resourceCulture);
             }
         }
+        
+        /// <summary>
+        ///   Ищет локализованную строку, похожую на SELECT 
+        ///    l.TechJournalLog, 
+        ///    l.DirectoryName, 
+        ///    l.Id, 
+        ///    l.FileName, 
+        ///    l.CreateDate, 
+        ///    l.ModificationDate, 
+        ///    l.LastEventNumber, 
+        ///    l.LastCurrentFileData, 
+        ///    l.LastStreamPosition
+        ///FROM LogFiles l
+        ///
+        ///INNER JOIN
+        ///
+        ///(
+        ///SELECT
+        ///	TechJournalLog,
+        ///	DirectoryName,
+        ///    MAX(Id) LastId
+        ///FROM LogFiles AS LF_LAST
+        ///WHERE LF_LAST.TechJournalLog = {techJournalLog:String}
+        ///GROUP BY TechJournalLog,
+        ///	DirectoryName
+        ///) lglast
+        ///
+        ///ON l.TechJournalLog = lglast.TechJournalLog 
+        ///	AND l.DirectoryNam [остаток строки не уместился]&quot;;.
+        /// </summary>
+        internal static string Query_GetActualPositions {
+            get {
+                return ResourceManager.GetString("Query_GetActualPositions", resourceCulture);
+            }
+        }
     }
 }

Libs/YY.TechJournalExportAssistant.ClickHouse/Resource.resx

@@ -117,12 +117,8 @@
   <resheader name="writer">
     <value>System.Resources.ResXResourceWriter, System.Windows.Forms, Version=4.0.0.0, Culture=neutral, PublicKeyToken=b77a5c561934e089</value>
   </resheader>
-  <data name="Query_CreateTable_EventData" xml:space="preserve">
-    <value>CREATE TABLE IF NOT EXISTS EventData AS EventDataStorage ENGINE = Buffer(currentDatabase(), EventDataStorage, 16, 10, 60, 100000, 1000000, 10000000, 100000000)</value>
-    <comment>Создание буфера для новых данных технологического журнала</comment>
-  </data>
   <data name="Query_CreateTable_EventDataStorage" xml:space="preserve">
-    <value>CREATE TABLE IF NOT EXISTS EventDataStorage
+    <value>CREATE TABLE IF NOT EXISTS EventData
 (
     TechJournalLog LowCardinality(String),
     DirectoryName LowCardinality(String),
@@ -167,15 +163,12 @@ ORDER BY Period
 SETTINGS index_granularity = 8192;</value>
     <comment>Создание таблицы для хранения данных технологического журнала</comment>
   </data>
-  <data name="Query_CreateTable_LogFiles" xml:space="preserve">
-    <value>CREATE TABLE IF NOT EXISTS LogFiles AS LogFilesStorage ENGINE = Buffer(currentDatabase(), LogFilesStorage, 16, 10, 60, 100000, 1000000, 10000000, 100000000)</value>
-  </data>
   <data name="Query_CreateTable_LogFilesStorage" xml:space="preserve">
-    <value>CREATE TABLE IF NOT EXISTS LogFilesStorage
+    <value>CREATE TABLE IF NOT EXISTS LogFiles
 (
 	TechJournalLog LowCardinality(String),
 	DirectoryName LowCardinality(String),
-	Id Int64 Codec(DoubleDelta, LZ4),
+	Id Int64 DEFAULT (toUnixTimestamp64Milli(now64())*1000000 + rowNumberInAllBlocks()) Codec(DoubleDelta, LZ4),
 	FileName LowCardinality(String),
 	CreateDate DateTime Codec(Delta, LZ4),
 	ModificationDate DateTime Codec(Delta, LZ4),
@@ -190,4 +183,36 @@ ORDER BY CreateDate
 SETTINGS index_granularity = 8192;</value>
     <comment>Создание таблицы для хранения данных лога обработки файлов</comment>
   </data>
+  <data name="Query_GetActualPositions" xml:space="preserve">
+    <value>SELECT 
+    l.TechJournalLog, 
+    l.DirectoryName, 
+    l.Id, 
+    l.FileName, 
+    l.CreateDate, 
+    l.ModificationDate, 
+    l.LastEventNumber, 
+    l.LastCurrentFileData, 
+    l.LastStreamPosition
+FROM LogFiles l
+
+INNER JOIN
+
+(
+SELECT
+	TechJournalLog,
+	DirectoryName,
+    MAX(Id) LastId
+FROM LogFiles AS LF_LAST
+WHERE LF_LAST.TechJournalLog = {techJournalLog:String}
+GROUP BY TechJournalLog,
+	DirectoryName
+) lglast
+
+ON l.TechJournalLog = lglast.TechJournalLog 
+	AND l.DirectoryName = lglast.DirectoryName 
+	AND l.Id = lglast.LastId
+ORDER BY l.DirectoryName</value>
+    <comment>Получение последних событий</comment>
+  </data>
 </root>

Libs/YY.TechJournalExportAssistant.ClickHouse/TechJournalOnClickHouse.cs

@@ -1,6 +1,5 @@
-using System;
-using System.Collections.Generic;
-using System.IO;
+using System.Collections.Generic;
+using System.Linq;
 using Microsoft.Extensions.Configuration;
 using YY.TechJournalExportAssistant.Core;
 using YY.TechJournalReaderAssistant;
@@ -14,12 +13,9 @@ namespace YY.TechJournalExportAssistant.ClickHouse
 
         private const int _defaultPortion = 1000;
         private readonly int _portion;
-        private DateTime _maxPeriodRowData;
         private TechJournalLogBase _techJournalLog;
         private readonly string _connectionString;
         private TechJournalPosition _lastTechJournalFilePosition;
-        private int _stepsToClearLogFiles = 10000;
-        private int _currentStepToClearLogFiles;
 
         #endregion
 
@@ -36,7 +32,6 @@ namespace YY.TechJournalExportAssistant.ClickHouse
         public TechJournalOnClickHouse(string connectionString, int portion)
         {
             _portion = portion;
-            _maxPeriodRowData = DateTime.MinValue;
 
             if (connectionString == null)
             {
@@ -53,33 +48,36 @@ namespace YY.TechJournalExportAssistant.ClickHouse
 
         #region Public Methods
 
-        public override TechJournalPosition GetLastPosition()
+        public override TechJournalPosition GetLastPosition(string directoryName)
         {
             if (_lastTechJournalFilePosition != null)
                 return _lastTechJournalFilePosition;
 
             TechJournalPosition position;
             using(var context = new ClickHouseContext(_connectionString))
-                position = context.GetLogFilePosition(_techJournalLog);
+                position = context.GetLogFilePosition(_techJournalLog, directoryName);
             
             _lastTechJournalFilePosition = position;
             return position;
         }
-        public override void SaveLogPosition(FileInfo logFileInfo, TechJournalPosition position)
+        public override void SaveLogPosition(TechJournalPosition position)
         {
             using (var context = new ClickHouseContext(_connectionString))
             {
-                context.SaveLogPosition(_techJournalLog, logFileInfo, position);
-                if (_currentStepToClearLogFiles == 0 || _currentStepToClearLogFiles >= _stepsToClearLogFiles)
-                {
-                    context.RemoveArchiveLogFileRecords(_techJournalLog);
-                    _currentStepToClearLogFiles = 0;
-                }
-                _currentStepToClearLogFiles += 1;
+                context.SaveLogPosition(_techJournalLog, position);
             }
 
             _lastTechJournalFilePosition = position;
         }
+
+        public override void SaveLogPositions(List<TechJournalPosition> positions)
+        {
+            using (var context = new ClickHouseContext(_connectionString))
+            {
+                context.SaveLogPositions(_techJournalLog, positions);
+            }
+        }
+
         public override int GetPortionSize()
         {
             return _portion;
@@ -100,26 +98,7 @@ namespace YY.TechJournalExportAssistant.ClickHouse
 
             using (var context = new ClickHouseContext(_connectionString))
             {
-                if (_maxPeriodRowData == DateTime.MinValue)
-                {
-                    _maxPeriodRowData = context.GetRowsDataMaxPeriod(
-                        _techJournalLog,
-                        fileName
-                    );
-                }
-
-                List<EventData> newEntities = new List<EventData>();
-                foreach (var itemRow in rowsData)
-                {
-                    if (itemRow == null)
-                        continue;
-                    if (_maxPeriodRowData != DateTime.MinValue && itemRow.Period <= _maxPeriodRowData)
-                        if (context.RowDataExistOnDatabase(_techJournalLog, itemRow, fileName))
-                            continue;
-
-                    newEntities.Add(itemRow);
-                }
-                context.SaveRowsData(_techJournalLog, newEntities, fileName);
+                context.SaveRowsData(_techJournalLog, rowsData.ToList(), fileName);
             }
         }
         public override void SetInformationSystem(TechJournalLogBase techJournalLog)
@@ -127,6 +106,29 @@ namespace YY.TechJournalExportAssistant.ClickHouse
             _techJournalLog = techJournalLog;
         }
 
+        public override void Save(IDictionary<string, List<EventData>> rowsData)
+        {
+            if (rowsData.Count == 0)
+                return;
+
+            using (var context = new ClickHouseContext(_connectionString))
+            {
+                context.SaveRowsData(_techJournalLog, rowsData);
+            }
+        }
+
+        public override IDictionary<string, TechJournalPosition> GetCurrentLogPositions(TechJournalSettings settings)
+        {
+            IDictionary<string, TechJournalPosition> positions;
+
+            using (var context = new ClickHouseContext(_connectionString))
+            {
+                positions = context.GetCurrentLogPositions(_techJournalLog);
+            }
+
+            return positions;
+        }
+
         #endregion
     }
 }

Libs/YY.TechJournalExportAssistant.ClickHouse/TechJournalOnClickHouseTargetBuilder.cs

@@ -0,0 +1,44 @@
+using System.Collections.Generic;
+using YY.TechJournalExportAssistant.Core;
+using YY.TechJournalExportAssistant.Core.SharedBuffer;
+using YY.TechJournalReaderAssistant;
+
+namespace YY.TechJournalExportAssistant.ClickHouse
+{
+    public class TechJournalOnClickHouseTargetBuilder : ITechJournalOnTargetBuilder
+    {
+        public ITechJournalOnTarget CreateTarget(TechJournalSettings settings, KeyValuePair<LogBufferItemKey, LogBufferItem> logBufferItem)
+        {
+            ITechJournalOnTarget target = new TechJournalOnClickHouse(settings.ConnectionString,
+                logBufferItem.Key.Settings.Portion);
+            target.SetInformationSystem(logBufferItem.Key.Settings.TechJournalLog);
+
+            return target;
+        }
+
+        public IDictionary<string, TechJournalPosition> GetCurrentLogPositions(TechJournalSettings settings, KeyValuePair<LogBufferItemKey, LogBufferItem> logBufferItem)
+        {
+            ITechJournalOnTarget target = CreateTarget(settings, logBufferItem);
+            return target.GetCurrentLogPositions(settings);
+        }
+
+        public IDictionary<string, TechJournalPosition> GetCurrentLogPositions(TechJournalSettings settings, TechJournalLogBase techJournalLog)
+        {
+            Dictionary<string, TechJournalPosition> allPositions = new Dictionary<string, TechJournalPosition>();
+            using (ClickHouseContext context = new ClickHouseContext(settings.ConnectionString))
+            {
+                return context.GetCurrentLogPositions(techJournalLog);
+            }
+        }
+
+        public void SaveRowsData(
+            TechJournalSettings settings,
+            Dictionary<LogBufferItemKey, LogBufferItem> dataFromBuffer)
+        {
+            using (ClickHouseContext context = new ClickHouseContext(settings.ConnectionString))
+            {
+                context.SaveRowsData(dataFromBuffer);
+            }
+        }
+    }
+}

Libs/YY.TechJournalExportAssistant.ClickHouse/YY.TechJournalExportAssistant.ClickHouse.csproj

@@ -3,7 +3,7 @@
   <PropertyGroup>
     <TargetFramework>netstandard2.1</TargetFramework>
     <GeneratePackageOnBuild>true</GeneratePackageOnBuild>
-    <Version>1.0.0.9</Version>
+    <Version>1.0.1.1</Version>
     <Authors>Permitin Yuriy</Authors>
     <Product>Technological journal's export assistant to ClickHouse</Product>
     <Description>Library for exporting 1C:Enterprise 8.x platform's technological journal files to ClickHouse</Description>

Libs/YY.TechJournalExportAssistant.Core/Helpers/DateTimeExtensions.cs

@@ -0,0 +1,14 @@
+using System;
+
+namespace YY.TechJournalExportAssistant.Core.Helpers
+{
+    public static class DateTimeExtensions
+    {
+        public static DateTime Truncate(this DateTime dateTime, TimeSpan timeSpan)
+        {
+            if (timeSpan == TimeSpan.Zero) return dateTime;
+            if (dateTime == DateTime.MinValue || dateTime == DateTime.MaxValue) return dateTime;
+            return dateTime.AddTicks(-(dateTime.Ticks % timeSpan.Ticks));
+        }
+    }
+}

Libs/YY.TechJournalExportAssistant.Core/ITechJournalOnTarget.cs

@@ -1,5 +1,5 @@
 using System.Collections.Generic;
-using System.IO;
+using YY.TechJournalExportAssistant.Core.SharedBuffer;
 using YY.TechJournalReaderAssistant;
 using YY.TechJournalReaderAssistant.Models;
 
@@ -7,11 +7,14 @@ namespace YY.TechJournalExportAssistant.Core
 {
     public interface ITechJournalOnTarget
     {
-        TechJournalPosition GetLastPosition();
-        void SaveLogPosition(FileInfo logFileInfo, TechJournalPosition position);
+        TechJournalPosition GetLastPosition(string directoryName);
+        void SaveLogPosition(TechJournalPosition position);
+        void SaveLogPositions(List<TechJournalPosition> positions);
         int GetPortionSize();
         void SetInformationSystem(TechJournalLogBase techJournalLog);
         void Save(EventData eventData, string fileName);
         void Save(IList<EventData> eventData, string fileName);
+        void Save(IDictionary<string, List<EventData>> rowsData);
+        IDictionary<string, TechJournalPosition> GetCurrentLogPositions(TechJournalSettings settings);
     }
 }

Libs/YY.TechJournalExportAssistant.Core/SharedBuffer/EventArgs/OnErrorExportSharedBufferEventArgs.cs

@@ -0,0 +1,14 @@
+using System;
+
+namespace YY.TechJournalExportAssistant.Core.SharedBuffer.EventArgs
+{
+    public sealed class OnErrorExportSharedBufferEventArgs : System.EventArgs
+    {
+        public OnErrorExportSharedBufferEventArgs(Exception exception)
+        {
+            Exception = exception;
+        }
+
+        public Exception Exception { get; }
+    }    
+}

Libs/YY.TechJournalExportAssistant.Core/SharedBuffer/EventArgs/OnSendLogFromSharedBufferEventArgs.cs

@@ -0,0 +1,15 @@
+using System.Collections.Generic;
+
+namespace YY.TechJournalExportAssistant.Core.SharedBuffer.EventArgs
+{
+    public sealed class OnSendLogFromSharedBufferEventArgs : System.EventArgs
+    {
+        public IReadOnlyDictionary<LogBufferItemKey, LogBufferItem> DataFromBuffer { get; }
+
+        public OnSendLogFromSharedBufferEventArgs(
+            IReadOnlyDictionary<LogBufferItemKey, LogBufferItem> dataFromBuffer)
+        {
+            DataFromBuffer = dataFromBuffer;
+        }
+    }    
+}

Libs/YY.TechJournalExportAssistant.Core/SharedBuffer/EventKey.cs

@@ -0,0 +1,11 @@
+using System;
+using System.IO;
+
+namespace YY.TechJournalExportAssistant.Core.SharedBuffer
+{
+    public class EventKey
+    {
+        public Guid Id { get; set; }
+        public FileInfo File { get; set; }
+    }
+}

Libs/YY.TechJournalExportAssistant.Core/SharedBuffer/ITechJournalOnTargetBuilder.cs

@@ -0,0 +1,12 @@
+using System.Collections.Generic;
+using YY.TechJournalReaderAssistant;
+
+namespace YY.TechJournalExportAssistant.Core.SharedBuffer
+{
+    public interface ITechJournalOnTargetBuilder
+    {
+        ITechJournalOnTarget CreateTarget(TechJournalSettings settings, KeyValuePair<LogBufferItemKey, LogBufferItem> logBufferItem);
+        IDictionary<string, TechJournalPosition> GetCurrentLogPositions(TechJournalSettings settings, TechJournalLogBase techJournalLog);
+        void SaveRowsData(TechJournalSettings settings, Dictionary<LogBufferItemKey, LogBufferItem> dataFromBuffer);
+    }
+}

Libs/YY.TechJournalExportAssistant.Core/SharedBuffer/LogBufferItem.cs

@@ -0,0 +1,45 @@
+using System;
+using System.Collections.Concurrent;
+using YY.TechJournalReaderAssistant;
+using YY.TechJournalReaderAssistant.Models;
+
+namespace YY.TechJournalExportAssistant.Core.SharedBuffer
+{
+    /// <summary>
+    /// Элемент буфера логов
+    /// </summary>
+    public class LogBufferItem
+    {
+        /// <summary>
+        /// Дата создания
+        /// </summary>
+        public DateTime Created { get; set; }
+
+        /// <summary>
+        /// Дата последнего обновления
+        /// </summary>
+        public DateTime LastUpdate { get; set; }
+
+        /// <summary>
+        /// Количество записей лога в буфере
+        /// </summary>
+        public long ItemsCount => LogRows.Count;
+
+        /// <summary>
+        /// Актуальная позиция чтения файла лога
+        /// </summary>
+        public TechJournalPosition LogPosition { get; set; }
+
+        /// <summary>
+        /// Записи логов
+        /// </summary>
+        public ConcurrentDictionary<EventKey, EventData> LogRows { get; set; }
+        
+        public LogBufferItem()
+        {
+            Created = DateTime.Now;
+            LastUpdate = DateTime.MinValue;
+            LogRows = new ConcurrentDictionary<EventKey, EventData>();
+        }
+    }
+}

Libs/YY.TechJournalExportAssistant.Core/SharedBuffer/LogBufferItemKey.cs

@@ -0,0 +1,21 @@
+using System;
+
+namespace YY.TechJournalExportAssistant.Core.SharedBuffer
+{
+    public class LogBufferItemKey
+    {
+        public TechJournalSettings.LogSourceSettings Settings { get; }
+        public DateTime Period { get; }
+        public string LogFile { get; }
+
+        public LogBufferItemKey(
+            TechJournalSettings.LogSourceSettings setting,
+            DateTime period,
+            string logFile)
+        {
+            Settings = setting;
+            Period = period;
+            LogFile = logFile;
+        }
+    }
+}

Libs/YY.TechJournalExportAssistant.Core/SharedBuffer/LogBuffers.cs

@@ -0,0 +1,127 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using YY.TechJournalReaderAssistant;
+using YY.TechJournalReaderAssistant.Models;
+
+namespace YY.TechJournalExportAssistant.Core.SharedBuffer
+{
+    public class LogBuffers
+    {
+        public readonly ConcurrentDictionary<LogBufferItemKey, LogBufferItem> LogBuffer;
+        
+        public ConcurrentDictionary<TechJournalSettings.LogSourceSettings, ConcurrentDictionary<string, TechJournalPosition>> LogPositions { get; }
+        
+        public LogBuffers()
+        {
+            LogBuffer = new ConcurrentDictionary<LogBufferItemKey, LogBufferItem>();
+            LogPositions = new ConcurrentDictionary<TechJournalSettings.LogSourceSettings, ConcurrentDictionary<string, TechJournalPosition>>();
+        }
+
+        /// <summary>
+        /// Общее количество записей логов в буфере
+        /// </summary>
+        public long TotalItemsCount
+        {
+            get
+            {
+                return LogBuffer
+                    .Select(e => e.Value.ItemsCount)
+                    .Sum();
+            }
+        }
+
+        public DateTime Created
+        {
+            get
+            {
+                DateTime bufferCreated = DateTime.MinValue;
+
+                foreach (var logBufferItem in LogBuffer)
+                {
+                    if (bufferCreated == DateTime.MinValue)
+                    {
+                        bufferCreated = logBufferItem.Value.Created;
+                    }
+                    else if (bufferCreated < logBufferItem.Value.Created)
+                    {
+                        bufferCreated = logBufferItem.Value.Created;
+                    }
+                }
+
+                return bufferCreated;
+            }
+        }
+
+        public void SaveLogsAndPosition(
+            TechJournalSettings.LogSourceSettings logSettings,
+            TechJournalPosition position,
+            IList<EventData> rowsData
+            )
+        {
+            lock (logSettings.LockObject)
+            {
+                var logFileInfo = new FileInfo(position.CurrentFileData);
+                SaveLogs(logSettings, position, rowsData, logFileInfo);
+            }
+        }
+        
+        private void SaveLogs(
+            TechJournalSettings.LogSourceSettings logSettings,
+            TechJournalPosition position,
+            IList<EventData> rowsData,
+            FileInfo logFileInfo)
+        {
+            var newBufferItem = new LogBufferItem();
+            newBufferItem.LastUpdate = DateTime.Now;
+            newBufferItem.LogPosition = position;
+            newBufferItem.Created = DateTime.Now;
+            foreach (var rowData in rowsData)
+            {
+                newBufferItem.LogRows.TryAdd(new EventKey()
+                {
+                    Id = Guid.NewGuid(),
+                    File = logFileInfo
+                }, rowData);
+            }
+
+            LogBuffer.TryAdd(new LogBufferItemKey(logSettings, DateTime.Now, logFileInfo.FullName), 
+                newBufferItem);
+
+            LogPositions.AddOrUpdate(logSettings,
+                (settings) =>
+                {
+                    var newPositions = new ConcurrentDictionary<string, TechJournalPosition>();
+                    if (logFileInfo.Directory != null)
+                        newPositions.AddOrUpdate(logFileInfo.Directory.Name,
+                            (dirName) => position, 
+                            (dirName, oldPosition) => position);
+                    return newPositions;
+                },
+                (settings, logBufferItem) =>
+                {
+                    if (logFileInfo.Directory != null)
+                        logBufferItem.AddOrUpdate(logFileInfo.Directory.Name,
+                            (dirName) => position,
+                            (dirName, oldPosition) => position);
+                    return logBufferItem;
+                });
+        }
+        
+        public TechJournalPosition GetLastPosition(
+            TechJournalSettings.LogSourceSettings logSettings, 
+            TechJournalLogBase techJournalLog,
+            string directoryName)
+        {
+            TechJournalPosition position = null;
+            if (LogPositions.TryGetValue(logSettings, out ConcurrentDictionary<string, TechJournalPosition> settingPositions))
+            {
+                settingPositions.TryGetValue(directoryName, out position);
+            }
+
+            return position;
+        }
+    }
+}

Libs/YY.TechJournalExportAssistant.Core/SharedBuffer/TechJournalExport.cs

@@ -0,0 +1,262 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using YY.TechJournalExportAssistant.Core.SharedBuffer.EventArgs;
+using YY.TechJournalReaderAssistant;
+
+namespace YY.TechJournalExportAssistant.Core.SharedBuffer
+{
+    public class TechJournalExport
+    {
+        #region Private Members
+
+        private readonly TechJournalSettings _settings;
+        private readonly LogBuffers _logBuffers;
+        private readonly ITechJournalOnTargetBuilder _techJournalTargetBuilder;
+
+        #endregion
+
+        #region Constructors
+
+        public TechJournalExport(TechJournalSettings settings, ITechJournalOnTargetBuilder techJournalTargetBuilder)
+        {
+            _settings = settings;
+            _logBuffers = new LogBuffers();
+            _techJournalTargetBuilder = techJournalTargetBuilder;
+
+            foreach (var logSourceSettings in _settings.LogSources)
+            {
+                _logBuffers.LogPositions.TryAdd(logSourceSettings,
+                    new ConcurrentDictionary<string, TechJournalPosition>());
+
+                var logPositions = techJournalTargetBuilder.GetCurrentLogPositions(settings, logSourceSettings.TechJournalLog);
+                foreach (var logPosition in logPositions)
+                {
+                    FileInfo logFileInfo = new FileInfo(logPosition.Value.CurrentFileData);
+                    _logBuffers.LogPositions.AddOrUpdate(logSourceSettings,
+                        (settingsKey) =>
+                        {
+                            var newPositions = new ConcurrentDictionary<string, TechJournalPosition>();
+                            if (logFileInfo.Directory != null)
+                                newPositions.AddOrUpdate(logFileInfo.Directory.Name,
+                                    (dirName) => logPosition.Value,
+                                    (dirName, oldPosition) => logPosition.Value);
+                            return newPositions;
+                        },
+                        (settingsKey, logBufferItemOld) =>
+                        {
+                            if (logFileInfo.Directory != null)
+                                logBufferItemOld.AddOrUpdate(logFileInfo.Directory.Name,
+                                    (dirName) => logPosition.Value,
+                                    (dirName, oldPosition) => logPosition.Value);
+                            return logBufferItemOld;
+                        });
+                }
+            }
+        }
+
+        #endregion
+
+        #region Public Methods
+
+        public async Task StartExport()
+        {
+            await StartExport(CancellationToken.None);
+        }
+
+        public async Task StartExport(CancellationToken cancellationToken)
+        {
+            List<Task> exportJobs = new List<Task>();
+            foreach (var logSource in _settings.LogSources)
+            {
+                Task logExportTask = Task.Run(() => LogExportJob(logSource, cancellationToken), cancellationToken);
+                exportJobs.Add(logExportTask);
+                await Task.Delay(1000, cancellationToken);
+            }
+
+            await Task.Delay(10000, cancellationToken);
+
+            exportJobs.Add(Task.Run(() => SendLogFromBuffer(cancellationToken), cancellationToken));
+
+            Task.WaitAll(exportJobs.ToArray(), cancellationToken);
+        }
+
+        #endregion
+
+        #region Private Methods
+
+        private async Task LogExportJob(
+            TechJournalSettings.LogSourceSettings settings,
+            CancellationToken cancellationToken)
+        {
+            string techJournalPath = settings.SourcePath;
+            TimeZoneInfo timeZone = settings.TimeZone;
+
+            while (true)
+            {
+                if (cancellationToken.IsCancellationRequested)
+                    break;
+                bool bufferBlocked = (_logBuffers.TotalItemsCount >= _settings.Export.Buffer.MaxBufferSizeItemsCount);
+
+                try
+                {
+                    if (!bufferBlocked)
+                    {
+                        TechJournalManager tjManager = new TechJournalManager(techJournalPath);
+                        foreach (var tjDirectory in tjManager.Directories)
+                        {
+                            if (cancellationToken.IsCancellationRequested)
+                                break;
+                            bufferBlocked = (_logBuffers.TotalItemsCount >= _settings.Export.Buffer.MaxBufferSizeItemsCount);
+                            if (bufferBlocked)
+                                break;
+                            if (!tjDirectory.DirectoryData.Exists)
+                                continue;
+
+                            using (TechJournalExportMaster exporter = new TechJournalExportMaster())
+                            {
+                                exporter.SetTechJournalPath(tjDirectory.DirectoryData.FullName, timeZone);
+
+                                TechJournalOnBuffer target =
+                                    new TechJournalOnBuffer(_logBuffers, settings.Portion);
+                                target.SetLogSettings(settings);
+                                target.SetInformationSystem(new TechJournalLogBase()
+                                {
+                                    Name = settings.Name,
+                                    Description = settings.Description
+                                });
+                                exporter.SetTarget(target);
+                                exporter.OnErrorExportData += OnErrorExportDataToBuffer;
+
+                                while (exporter.NewDataAvailable())
+                                {
+                                    if (cancellationToken.IsCancellationRequested)
+                                        break;
+
+                                    exporter.SendData();
+
+                                    if (cancellationToken.IsCancellationRequested)
+                                        break;
+                                    bufferBlocked = (_logBuffers.TotalItemsCount >= _settings.Export.Buffer.MaxBufferSizeItemsCount);
+                                    if (bufferBlocked)
+                                        break;
+                                }
+                            }
+                        }
+                    }
+
+                    if (bufferBlocked)
+                    {
+                        await Task.Delay(100, cancellationToken);
+                    }
+                    else if (_settings.WatchMode.Use)
+                    {
+                        await Task.Delay(_settings.WatchMode.Periodicity, cancellationToken);
+                    }
+                    else
+                    {
+                        break;
+                    }
+                }
+                catch (Exception e)
+                {
+                    RaiseOnError(new OnErrorExportSharedBufferEventArgs(e));
+                    await Task.Delay(60000, cancellationToken);
+                }
+            }
+        }
+
+        private async Task SendLogFromBuffer(CancellationToken cancellationToken)
+        {
+            while (true)
+            {
+                if (cancellationToken.IsCancellationRequested)
+                    break;
+
+                bool needExport = false;
+
+                if (_logBuffers.TotalItemsCount >= _settings.Export.Buffer.MaxItemCountSize)
+                {
+                    needExport = true;
+                }
+                else
+                {
+                    DateTime bufferCreated = _logBuffers.Created;
+                    var createTimeLeftMs = (DateTime.Now - bufferCreated).TotalMilliseconds;
+                    if (bufferCreated != DateTime.MinValue && createTimeLeftMs >= _settings.Export.Buffer.MaxSaveDurationMs)
+                    {
+                        needExport = true;
+                    }
+                }
+
+                if (needExport)
+                {
+                    if (cancellationToken.IsCancellationRequested)
+                        break;
+
+                    try
+                    {
+                        var itemsToUpload = _logBuffers.LogBuffer
+                            .Select(i => i.Key)
+                            .OrderBy(i => i.Period)
+                            .ToList();
+
+                        var dataToUpload = _logBuffers.LogBuffer
+                            .Where(i => itemsToUpload.Contains(i.Key))
+                            .ToDictionary(k => k.Key, v => v.Value);
+
+                        OnSend(new OnSendLogFromSharedBufferEventArgs(dataToUpload));
+
+                        _techJournalTargetBuilder.SaveRowsData(_settings, dataToUpload);
+
+                        foreach (var itemToUpload in itemsToUpload)
+                        {
+                            _logBuffers.LogBuffer.TryRemove(itemToUpload, out _);
+                        }
+                    }
+                    catch (Exception e)
+                    {
+                        RaiseOnError(new OnErrorExportSharedBufferEventArgs(e));
+                        await Task.Delay(1000, cancellationToken);
+                    }
+                }
+
+                await Task.Delay(100, cancellationToken);
+
+                if (!_settings.WatchMode.Use
+                    && _logBuffers.TotalItemsCount == 0)
+                {
+                    break;
+                }
+            }
+        }
+
+        #endregion
+
+        #region Events
+
+        public delegate void OnSendLogFromSharedBufferEventArgsHandler(OnSendLogFromSharedBufferEventArgs args);
+        public delegate void OnErrorExportSharedBufferEventArgsHandler(OnErrorExportSharedBufferEventArgs args);
+        public event OnSendLogFromSharedBufferEventArgsHandler OnSendLogEvent;
+        public event OnErrorExportSharedBufferEventArgsHandler OnErrorEvent;
+
+        protected void OnSend(OnSendLogFromSharedBufferEventArgs args)
+        {
+            OnSendLogEvent?.Invoke(args);
+        }
+        protected void RaiseOnError(OnErrorExportSharedBufferEventArgs args)
+        {
+            OnErrorEvent?.Invoke(args);
+        }
+        private void OnErrorExportDataToBuffer(OnErrorExportDataEventArgs e)
+        {
+            RaiseOnError(new OnErrorExportSharedBufferEventArgs(e.Exception));
+        }
+
+        #endregion
+    }
+}

Libs/YY.TechJournalExportAssistant.Core/SharedBuffer/TechJournalOnBuffer.cs

@@ -0,0 +1,92 @@
+using System.Collections.Generic;
+using System.Linq;
+using YY.TechJournalReaderAssistant;
+using EventData = YY.TechJournalReaderAssistant.Models.EventData;
+
+namespace YY.TechJournalExportAssistant.Core.SharedBuffer
+{
+    public class TechJournalOnBuffer : TechJournalOnTarget
+    {
+        #region Private Member Variables
+
+        private const int _defaultPortion = 1000;
+        private readonly int _portion;
+        private TechJournalLogBase _techJournalLog;
+        private TechJournalPosition _lastTechJournalFilePosition;
+        private TechJournalSettings.LogSourceSettings _logSettings;
+        private readonly LogBuffers _logBuffers;
+        private IList<EventData> _rowsDataForPosition;
+
+
+        #endregion
+
+        #region Constructor
+
+        public TechJournalOnBuffer() : this(null, _defaultPortion)
+        {
+
+        }
+        public TechJournalOnBuffer(int portion) : this(null, portion)
+        {
+            _portion = portion;
+        }
+        public TechJournalOnBuffer(LogBuffers buffers, int portion)
+        {
+            _portion = portion;
+            _rowsDataForPosition = new List<EventData>();
+            _logBuffers = buffers;
+        }
+
+        #endregion
+
+        #region Public Methods
+
+        public void SetLogSettings(TechJournalSettings.LogSourceSettings logSettings)
+        {
+            _logSettings = logSettings;
+        }
+        public void SetLastPosition(TechJournalPosition position)
+        {
+            _lastTechJournalFilePosition = position;
+        }
+        public override TechJournalPosition GetLastPosition(string directoryName)
+        {
+            if (_lastTechJournalFilePosition != null)
+                return _lastTechJournalFilePosition;
+
+            TechJournalPosition position = _logBuffers.GetLastPosition(_logSettings, _techJournalLog, directoryName);
+
+            _lastTechJournalFilePosition = position;
+            return position;
+        }
+        public override void SaveLogPosition(TechJournalPosition position)
+        {
+            _lastTechJournalFilePosition = position;
+            _logBuffers.SaveLogsAndPosition(_logSettings, position, _rowsDataForPosition);
+            _rowsDataForPosition.Clear();
+        }
+        public override int GetPortionSize()
+        {
+            return _portion;
+        }
+        public override void Save(EventData rowData, string fileName)
+        {
+            Save(new List<EventData>()
+            {
+                rowData
+            }, fileName);
+        }
+
+        public override void Save(IList<EventData> rowsData, string fileName)
+        {
+            _rowsDataForPosition.Clear();
+            _rowsDataForPosition = rowsData.ToList();
+        }
+        public override void SetInformationSystem(TechJournalLogBase techJournalLog)
+        {
+            _techJournalLog = techJournalLog;
+        }
+
+        #endregion
+    }
+}

Libs/YY.TechJournalExportAssistant.Core/TechJournalExportMaster.cs

@@ -126,7 +126,9 @@ namespace YY.TechJournalExportAssistant.Core
 
         private TechJournalPosition GetFixedCurrentPosition()
         {
-            TechJournalPosition lastPosition = _target.GetLastPosition();
+            DirectoryInfo logDirectoryInfo = new DirectoryInfo(_techJournalLogPath);
+            string directoryName = logDirectoryInfo.Name;
+            TechJournalPosition lastPosition = _target.GetLastPosition(directoryName);
 
             if (lastPosition != null)
             {
@@ -161,15 +163,13 @@ namespace YY.TechJournalExportAssistant.Core
             RiseBeforeExportData(out var cancel);
             if (!cancel)
             {
-                _target.Save(_dataToSend, currentFile.Name);
+                _target.Save(_dataToSend, currentFile.FullName);
                 RiseAfterExportData(reader.GetCurrentPosition());
             }
 
             if (reader.CurrentFile != null)
             {
-                _target.SaveLogPosition(
-                    currentFile,
-                    reader.GetCurrentPosition());
+                _target.SaveLogPosition(reader.GetCurrentPosition());
             }
             _dataToSend.Clear();
         }
@@ -217,13 +217,11 @@ namespace YY.TechJournalExportAssistant.Core
         }
         private void TechJournalReader_AfterReadFile(TechJournalReader sender, AfterReadFileEventArgs args)
         {
-            FileInfo _lastEventLogDataFileInfo = new FileInfo(args.FileName);
-
             if (_dataToSend.Count >= 0)
                 SendDataCurrentPortion(sender);
 
             TechJournalPosition position = sender.GetCurrentPosition();
-            _target.SaveLogPosition(_lastEventLogDataFileInfo, position);
+            _target.SaveLogPosition(position);
         }
         private void TechJournalReader_OnErrorEvent(TechJournalReader sender, OnErrorEventArgs args)
         {

Libs/YY.TechJournalExportAssistant.Core/TechJournalLogBase.cs

@@ -3,7 +3,6 @@
     public class TechJournalLogBase
     {
         public virtual string Name { get; set; }
-        public virtual string DirectoryName { get; set; }
         public virtual string Description { get; set; }
 
         public override string ToString()

Libs/YY.TechJournalExportAssistant.Core/TechJournalOnTarget.cs

@@ -1,6 +1,5 @@
 using System;
 using System.Collections.Generic;
-using System.IO;
 using YY.TechJournalReaderAssistant;
 using YY.TechJournalReaderAssistant.Models;
 
@@ -10,7 +9,12 @@ namespace YY.TechJournalExportAssistant.Core
     {
         #region Public Methods
 
-        public virtual TechJournalPosition GetLastPosition()
+        public virtual TechJournalPosition GetLastPosition(string directoryName)
+        {
+            throw new NotImplementedException();
+        }
+
+        public virtual void SaveLogPositions(List<TechJournalPosition> positions)
         {
             throw new NotImplementedException();
         }
@@ -30,7 +34,7 @@ namespace YY.TechJournalExportAssistant.Core
             throw new NotImplementedException();
         }
 
-        public virtual void SaveLogPosition(FileInfo logFileInfo, TechJournalPosition position)
+        public virtual void SaveLogPosition(TechJournalPosition position)
         {
             throw new NotImplementedException();
         }
@@ -39,7 +43,17 @@ namespace YY.TechJournalExportAssistant.Core
         {
             throw new NotImplementedException();
         }
-        
+
+        public virtual void Save(IDictionary<string, List<EventData>> rowsData)
+        {
+            throw new NotImplementedException();
+        }
+
+        public virtual IDictionary<string, TechJournalPosition> GetCurrentLogPositions(TechJournalSettings settings)
+        {
+            throw new NotImplementedException();
+        }
+
         #endregion
     }
 }

Libs/YY.TechJournalExportAssistant.Core/TechJournalSettings.cs

@@ -0,0 +1,143 @@
+using System;
+using System.Collections.Generic;
+using Microsoft.Extensions.Configuration;
+
+namespace YY.TechJournalExportAssistant.Core
+{
+    public class TechJournalSettings
+    {
+        public static TechJournalSettings Create(string configFile)
+        {
+            IConfiguration Configuration = new ConfigurationBuilder()
+                .AddJsonFile(configFile, optional: true, reloadOnChange: true)
+                .Build();
+
+            return Create(Configuration);
+        }
+        public static TechJournalSettings Create(IConfiguration configuration)
+        {
+            return new TechJournalSettings(configuration);
+        }
+
+        public string ConnectionString { get; }
+        public WatchModeSettings WatchMode { get; }
+        public ExportSettings Export { get; }
+        public IReadOnlyList<LogSourceSettings> LogSources { get; }
+
+        private TechJournalSettings(IConfiguration configuration)
+        {
+            ConnectionString = configuration.GetConnectionString("TechJournalDatabase");
+
+            var exportParametersSection = configuration.GetSection("Export");
+            var bufferSection = exportParametersSection.GetSection("Buffer");
+            var maxItemCountSize = bufferSection.GetValue("MaxItemCountSize", 10000);
+            var maxSaveDurationMs = bufferSection.GetValue("MaxSaveDurationMs", 60000);
+            var maxBufferSizeItemsCount = bufferSection.GetValue("MaxBufferSizeItemsCount", 100000);
+            Export = new ExportSettings(
+                new ExportSettings.BufferSettings(maxItemCountSize, maxSaveDurationMs, maxBufferSizeItemsCount));
+
+            var watchModeSection = configuration.GetSection("WatchMode");
+            var useWatchMode = watchModeSection.GetValue("Use", false);
+            var periodicityWatchMode = watchModeSection.GetValue("Periodicity", 60000);
+            WatchMode = new WatchModeSettings(useWatchMode, periodicityWatchMode);
+
+            var logSourcesSection = configuration.GetSection("LogSources");
+            var logSources = logSourcesSection.GetChildren();
+            List<LogSourceSettings> logSourceSettings = new List<LogSourceSettings>();
+            foreach (var logSource in logSources)
+            {
+                var techJournalLogName = logSource.GetValue("Name", string.Empty);
+                var techJournalLogDescription = logSource.GetValue("Description", string.Empty);
+                string sourcePath = logSource.GetValue("SourcePath", string.Empty);
+                int portion = logSource.GetValue("Portion", 10000);
+                string timeZoneName = logSource.GetValue("TimeZone", string.Empty);
+                TimeZoneInfo timeZone;
+                if (string.IsNullOrEmpty(timeZoneName))
+                    timeZone = TimeZoneInfo.Local;
+                else
+                    timeZone = TimeZoneInfo.FindSystemTimeZoneById(timeZoneName);
+
+                logSourceSettings.Add(new LogSourceSettings(
+                    techJournalLogName,
+                    techJournalLogDescription,
+                    sourcePath,
+                    portion,
+                    timeZone));
+            }
+            LogSources = logSourceSettings;
+        }
+
+        public class WatchModeSettings
+        {
+            public bool Use { get; }
+            public int Periodicity { get; }
+
+            public WatchModeSettings(bool use, int periodicity)
+            {
+                Use = use;
+                Periodicity = periodicity;
+            }
+        }
+
+        public class LogSourceSettings
+        {
+            private TechJournalLogBase _techJournalLog;
+            public string Name { get; }
+            public string Description { get; }
+            public TechJournalLogBase TechJournalLog
+            {
+                get
+                {
+                    if (_techJournalLog == null)
+                    {
+                        _techJournalLog = new TechJournalLogBase()
+                        {
+                            Name = Name,
+                            Description = Description
+                        };
+                    }
+
+                    return _techJournalLog;
+                }
+            }
+            public string SourcePath { get; }
+            public int Portion { get; }
+            public TimeZoneInfo TimeZone { get; }
+            public object LockObject { get; }
+
+            public LogSourceSettings(string name, string description, string sourcePath, int portion, TimeZoneInfo timeZone)
+            {
+                Name = name;
+                Description = description;
+                Portion = portion;
+                SourcePath = sourcePath;
+                TimeZone = timeZone;
+                LockObject = new object();
+            }
+        }
+
+        public class ExportSettings
+        {
+            public BufferSettings Buffer { get; }
+
+            public ExportSettings(BufferSettings bufferSettings)
+            {
+                Buffer = bufferSettings;
+            }
+
+            public class BufferSettings
+            {
+                public long MaxItemCountSize { get; }
+                public long MaxSaveDurationMs { get; }
+                public long MaxBufferSizeItemsCount { get; }
+
+                public BufferSettings(long maxItemCountSize, long maxSaveDurationMs, long maxBufferSizeItemsCount)
+                {
+                    MaxItemCountSize = maxItemCountSize;
+                    MaxSaveDurationMs = maxSaveDurationMs;
+                    MaxBufferSizeItemsCount = maxBufferSizeItemsCount;
+                }
+            }
+        }
+    }
+}

Libs/YY.TechJournalExportAssistant.Core/YY.TechJournalExportAssistant.Core.csproj

@@ -3,7 +3,7 @@
   <PropertyGroup>
     <TargetFramework>netstandard2.1</TargetFramework>
     <GeneratePackageOnBuild>true</GeneratePackageOnBuild>
-    <Version>1.0.0.8</Version>
+    <Version>1.0.1.1</Version>
     <Authors>Permitin Yuriy</Authors>
     <Product>Technological journal's export assistant. Core library</Product>
     <Description>Core library for exporting 1C:Enterprise 8.x platform's technological journal files to different storage</Description>
@@ -15,6 +15,9 @@
   </PropertyGroup>
 
   <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Configuration" Version="5.0.0" />
+    <PackageReference Include="Microsoft.Extensions.Configuration.Binder" Version="5.0.0" />
+    <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="5.0.0" />
     <PackageReference Include="YY.TechJournalReaderAssistant" Version="1.0.0.19" />
   </ItemGroup>
 

README.md

@@ -30,7 +30,8 @@
   * YY.TechJournalExportAssistant.Core - ядро библиотеки с основным функционалом чтения и передачи данных.
   * YY.TechJournalExportAssistant.ClickHouse - функционал для экспорта данных в базу ClickHouse.
 * Примеры приложений
-  * YY.YY.TechJournalExportAssistantConsoleApp - пример приложения для экспорта данных в базу ClickHouse.
+  * YY.TechJournalExportAssistantConsoleApp - пример приложения для экспорта данных в базу ClickHouse.
+  * YY.TechJournalExportAssistantWithSharedBufferConsoleApp - пример приложения для экспорта в базу ClickHouse из множества источников логов с использованием общего буфера.
 
 ## Требования и совместимость
 
@@ -41,11 +42,11 @@
 
 В большинстве случаев работоспособность подтверждается и на более старых версиях ПО, но меньше тестируется. Основная разработка ведется для Microsoft Windows, но некоторый функционал проверялся под *.nix.*
 
-## Пример использования
+### Простой экспорт
 
 Репозиторий содержим пример консольного приложения для экспорта данных в базу ClickHouse - **YY.YY.TechJournalExportAssistantConsoleApp**.
 
-### Конфигурация
+#### Конфигурация
 
 Первое, с чего следует начать - это конфигурационный файл приложения "appsettings.json". Это JSON-файл со строкой подключения к базе данных, сведениями об технологическом журнале и параметрами его обработки. Располагается в корне каталога приложения.
 
@@ -80,7 +81,7 @@
 
 Настройки "UseWatchMode" и "WatchPeriod" не относятся к библиотеке. Эти параметры добавлены лишь для примера консольного приложения и используется в нем же.
 
-### Пример использования
+#### Пример использования
 
 На следующем листинге показан пример использования библиотеки.
 
@@ -144,7 +145,6 @@ class Program
                     target.SetInformationSystem(new TechJournalLogBase()
                     {
                         Name = techJournalName,
-                        DirectoryName = tjDirectory.DirectoryData.Name,
                         Description = techJournalDescription
                     });
                     exporter.SetTarget(target);
@@ -230,6 +230,129 @@ class Program
 
 С их помощью можно проанализировать какие данные будут выгружены и отказаться от выгрузки с помощью поля "Cancel" в параметре события "BeforeExportDataEventArgs" в событии "Перед экспортом данных". В событии "После экспорта данных" можно проанализировать выгруженные данные.
 
+### Экспорт из множества источников логов
+
+В некоторых задачах источников логов для экспорта может быть очень много. При использовании предыдущего подхода может возникнуть проблема, если запросов к хранилищу будет очень много. Это не только приводит к проблемам производительности, но и может привести к остановке работы хранилища логов. Например, ClickHouse не любит большое количество операций записи данных. В этом случае как-раз и поможет общий буфер для хранения данных до их передачи в хранилище. Именно для этих целей и был создан представленный ниже функционал.
+
+#### Настройки для экспорта
+
+В отличии от предыдущего примера, конфигурация при экспорте из множества источников логов будет выглядеть следующим образом.
+
+```json
+{
+  "ConnectionStrings": {
+    "TechJournalDatabase": "Host=127.0.0.1;Port=8123;Username=default;password=;Database=AmazingTechJournalWithBufferDatabase;"
+  },
+  "WatchMode": {
+    "Use": true,
+    "Periodicity": 10000
+  },
+  "Export": {
+    "Buffer": {
+      "MaxItemCountSize": 500000,
+      "MaxSaveDurationMs": 60000,
+      "MaxBufferSizeItemsCount":  1000000
+    } 
+  },
+  "LogSources": [
+    {
+      "Name": "TechJournal1C 1",
+      "Description": "Технологический журнал. Очень разный. 1",
+      "SourcePath": "Q:\\TJ1",
+      "Portion": 10000,
+      "TimeZone": ""
+    },
+    {
+      "Name": "TechJournal1C 2",
+      "Description": "Технологический журнал. Очень разный. 2",
+      "SourcePath": "Q:\\TJ2",
+      "Portion": 10000,
+      "TimeZone": ""
+    },
+    {
+      "Name": "TechJournal1C 3",
+      "Description": "Технологический журнал. Очень разный. 3",
+      "SourcePath": "Q:\\TJ3",
+      "Portion": 10000,
+      "TimeZone": ""
+    }
+  ]
+}
+```
+
+Секция **"ConnectionStrings"** содержит строку подключения **"TechJournalDatabase"** к базе данных для экспорта. База будет создана автоматически при первом запуске приложения. Также можно создать ее вручную, главное, чтобы структура была соответствующей. Имя строки подключения **"TechJournalDatabase"** - это значение по умолчанию. Контекст приложения будет использовать ее автоматически, если это не переопределено разработчиком явно.
+
+Секция **"WatchMode"** содержит настройки режима "наблюдения", в котором библиотеки будут отслеживать появление новых записей в файлах логов. Параметр "Use" включает или отключает этот режим работы, а "Periodicity" указываем в миллисекундах периодичность, с которой выполняется проверка появления новых данных.
+
+Секция **"Export.Buffer"** содержит настройки работы буфера:
+    * **MaxBufferSizeItemsCount** - максимальное количество записей в буфере. По достижению этого размера запись в буфер останавливается.
+    * **MaxItemCountSize** - количество записей, по достижению которого выполняется экспорт данных из буфера и очистка буфера от ранее выгруженных данных.
+    * **MaxSaveDurationMs** - количество миллисекунд хранения записей в буфере. По истечению этого времени записи из буфера будут отправлены в хранилище в любом случае, независимо от количества записей в буфере.
+
+Секция **"LogSources"** содержит список параметров обработки технологических журналов, для каждого из которых указываются параметры:
+
+    * **Name** - имя источника логов.
+    * **Description** - описание источника логов.
+    * **SourcePath** - путь к каталогу с файлами технологического журнала. Необходимо указывать каталог аналогично тому, как он был указан в файле настройки технологического журнала (т.е. в нем должны быть каталоги по процессам и т.д.).
+    * **Portion** - количество записей, передаваемых в одной порции в хранилище.
+    * **TimeZone** - часовой пояс логов для корректной обработки дат.
+
+Почти все настройки аналогич простому способу экспорта, кроме настроек буфера. От них зависит сколько памяти будет выделено для работы буфера и как часто будет выполняться запрос экспорта данных. Параметры нужно подбирать индивидуально, но можете для начала использовать стандартные настройки из примера и менять по обстоятельству.
+
+#### Пример использования буфера
+
+На следующем листинге инициализируем настройки экспорта из файла конфигурации (см. выше), настраиваем обработчики событий экспорта и запускаем сам экспорт.
+
+```csharp
+using Microsoft.Extensions.Configuration;
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using YY.TechJournalExportAssistant.ClickHouse;
+using YY.TechJournalExportAssistant.Core;
+using YY.TechJournalExportAssistant.Core.SharedBuffer;
+using YY.TechJournalExportAssistant.Core.SharedBuffer.EventArgs;
+
+namespace YY.TechJournalExportAssistantWithSharedBufferConsoleApp
+{
+    class Program
+    {
+        public static IConfiguration Configuration = new ConfigurationBuilder()
+            .AddJsonFile("appsettings.json", optional: false, reloadOnChange: false)
+            .Build();
+        
+        static async Task Main(string[] args)
+        {
+            TechJournalSettings settings = TechJournalSettings.Create(Configuration);
+            
+            TechJournalExport exportMaster = new TechJournalExport(settings, new TechJournalOnClickHouseTargetBuilder());
+            exportMaster.OnErrorEvent += OnError;
+            exportMaster.OnSendLogEvent += OnSend;
+            await exportMaster.StartExport();
+
+            Console.WriteLine("Good luck & bye!");
+        }
+
+        private static void OnError(OnErrorExportSharedBufferEventArgs args)
+        {
+            Console.WriteLine($"Ошибка при экспорте логов: {args.Exception}");
+        }
+
+        private static void OnSend(OnSendLogFromSharedBufferEventArgs args)
+        {
+            Console.WriteLine($"Отправка данных в хранилище:\n" +
+                              $"Записей: {args.DataFromBuffer.Values.SelectMany(i => i.LogRows).Select(i => i.Value).Count()}\n" +
+                              $"Актуальных позиций чтения: {args.DataFromBuffer.Values.Select(i => i.LogPosition).Count() }");
+        }
+    }
+}
+
+```
+
+В каком-то плане работа с экспортом через буфер выглядит даже проще, чем первый пример. Но за служебных классом "TechJournalExport" кроется многопоточная обработка чтения логов и отдельный поток экспорта данных в хранилище. Подойдите к настройкам буфера в этом случае разумно, чтобы использовать ресурсы сервера эффективно.
+
+Также не забываем, что чтение файлов логов читается в один поток в рамках одной настройки логов ТЖ (одного каталога с логами). А вот несколько каталогов логов уже будут обрабатываться в отдельных потоках.
+
 ## Cценарии использования
 
 Библиотека может быть использования для создания приложений для экспорта технологического журнала платформы 1С:Предприяние 8.ч в нестандартные хранилища, которые упрощают анализ данных и позволяют организовать эффективный мониторинг.
@@ -239,10 +362,8 @@ class Program
 Планы в части разработки:
 
 * Добавить возможность экспорта данных в PostgreSQL
-* Улучшить обработку ошибок по уровням возникновения (критические и нет)
 * Улучшение производительности и добавление bencmark'ов
-* Расширение unit-тестов библиотеки
 
 ## Лицензия
 
-MIT - делайте все, что посчитаете нужным. Никакой гарантии и никаких ограничений по использованию.
+MIT - делайте все, что посчитаете нужным. Никакой гарантии и никаких ограничений по использованию.

Tests/YY.TechJournalExportAssistant.Tests/TechJournalExportAssistantToClickHouseTests.cs

@@ -1,25 +1,32 @@
 using Microsoft.Extensions.Configuration;
 using System;
+using System.Collections.Generic;
 using System.IO;
 using System.IO.Compression;
+using System.Text;
 using System.Threading;
 using ClickHouse.Client.ADO;
+using Newtonsoft.Json;
 using Xunit;
 using YY.TechJournalExportAssistant.ClickHouse;
 using YY.TechJournalExportAssistant.ClickHouse.Helpers;
 using YY.TechJournalExportAssistant.Core;
+using YY.TechJournalExportAssistant.Core.SharedBuffer;
 using YY.TechJournalReaderAssistant;
 
 namespace YY.TechJournalExportAssistant.Tests
 {
     public class TechJournalExportAssistantToClickHouseTests
     {
-        private readonly string _configFilePath;
         private readonly string _logDataPath;
+        private readonly IConfiguration Configuration;
 
         public TechJournalExportAssistantToClickHouseTests()
         {
-            _configFilePath = GetConfigFile();
+            var configFilePath = GetConfigFile();
+            Configuration = new ConfigurationBuilder()
+                .AddJsonFile(configFilePath, optional: true, reloadOnChange: true)
+                .Build();
 
             string unitTestDirectory = Directory.GetCurrentDirectory();
 
@@ -32,10 +39,6 @@ namespace YY.TechJournalExportAssistant.Tests
         [Fact]
         public void ExportDataTest()
         {
-            IConfiguration Configuration = new ConfigurationBuilder()
-                .AddJsonFile(_configFilePath, optional: true, reloadOnChange: true)
-                .Build();
-
             string connectionString = Configuration.GetConnectionString("TechJournalDatabase");
 
             IConfigurationSection techJournalSection = Configuration.GetSection("TechJournal");
@@ -83,7 +86,6 @@ namespace YY.TechJournalExportAssistant.Tests
                         target.SetInformationSystem(new TechJournalLogBase()
                         {
                             Name = techJournalName,
-                            DirectoryName = tjDirectory.DirectoryData.Name,
                             Description = techJournalDescription
                         });
                         exporter.SetTarget(target);
@@ -124,6 +126,85 @@ namespace YY.TechJournalExportAssistant.Tests
             Assert.Equal(eventNumberReader, eventNumberFromClickHouse);
         }
 
+        [Fact]
+        public async void ExportDataWithSharedBufferTest()
+        {
+            string connectionString = Configuration.GetConnectionString("TechJournalDatabase");
+
+            IConfigurationSection techJournalSection = Configuration.GetSection("TechJournal");
+            int watchPeriodSeconds = techJournalSection.GetValue("WatchPeriod", 60);
+            int watchPeriodSecondsMs = watchPeriodSeconds * 1000;
+            bool useWatchMode = techJournalSection.GetValue("UseWatchMode", false);
+            int portion = techJournalSection.GetValue("Portion", 1000);
+            string timeZoneName = techJournalSection.GetValue("TimeZone", string.Empty);
+
+            IConfigurationSection techJournalLogSection = Configuration.GetSection("TechJournalLog");
+            string techJournalName = techJournalLogSection.GetValue("Name", string.Empty);
+            string techJournalDescription = techJournalLogSection.GetValue("Description", string.Empty);
+
+            var LogSources = new List<dynamic>();
+            for (int i = 1; i <= 3; i++)
+            {
+                LogSources.Add(new
+                {
+                    Name = $"{techJournalName} {i}",
+                    Description = $"{techJournalDescription} {i}",
+                    SourcePath = _logDataPath,
+                    Portion = portion,
+                    TimeZone = timeZoneName
+                });
+            }
+
+            var configurationObject = new
+            {
+                ConnectionStrings = new
+                {
+                    TechJournalDatabase = connectionString.Replace("Database=", "Database=SharedBuffer")
+                },
+                WatchMode = new
+                {
+                    Use = useWatchMode,
+                    Periodicity = watchPeriodSecondsMs
+                },
+                Export = new
+                {
+                    Buffer = new
+                    {
+                        MaxItemCountSize = 10,
+                        MaxSaveDurationMs = 5000,
+                        MaxBufferSizeItemsCount = 100
+                    }
+                },
+                LogSources
+            };
+
+            var configurationAsJson = JsonConvert.SerializeObject(configurationObject);
+            IConfiguration configurationForTest = new ConfigurationBuilder().AddJsonStream(new MemoryStream(Encoding.UTF8.GetBytes(configurationAsJson))).Build();
+            TechJournalSettings settings = TechJournalSettings.Create(configurationForTest);
+
+            ClickHouseHelpers.DropDatabaseIfExist(settings.ConnectionString);
+
+            TechJournalExport exportMaster = new TechJournalExport(settings, new TechJournalOnClickHouseTargetBuilder());
+            await exportMaster.StartExport();
+
+            int eventNumberFromClickHouse = 0;
+            using (ClickHouseConnection cn = new ClickHouseConnection(settings.ConnectionString))
+            {
+                using (ClickHouseCommand cmd = new ClickHouseCommand(cn))
+                {
+                    cmd.CommandText =
+                        @"SELECT
+	                    COUNT(*)
+                    FROM EventData ed";
+                    var reader = await cmd.ExecuteReaderAsync();
+                    if (await reader.ReadAsync())
+                        eventNumberFromClickHouse = Convert.ToInt32(reader.GetValue(0));
+                }
+            }
+
+            Assert.Equal(9339, eventNumberFromClickHouse);
+        }
+
         private string GetConfigFile()
         {
             // TODO

Tests/YY.TechJournalExportAssistant.Tests/YY.TechJournalExportAssistant.Tests.csproj

@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
-    <TargetFramework>netcoreapp3.1</TargetFramework>
+    <TargetFramework>net5.0</TargetFramework>
 
     <IsPackable>false</IsPackable>
   </PropertyGroup>

YY.TechJournalExportAssistant.sln

@@ -38,6 +38,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Scripts", "Scripts", "{7E5B
 		Scripts\install-clickhouse.sh = Scripts\install-clickhouse.sh
 	EndProjectSection
 EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "YY.TechJournalExportAssistantWithSharedBufferConsoleApp", "Apps\YY.TechJournalExportAssistantWithSharedBufferConsoleApp\YY.TechJournalExportAssistantWithSharedBufferConsoleApp.csproj", "{75DDC845-838D-4F33-914B-2D833179A463}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -60,6 +62,10 @@ Global
 		{62CDAE77-D687-423E-B88C-25967ECC2353}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{62CDAE77-D687-423E-B88C-25967ECC2353}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{62CDAE77-D687-423E-B88C-25967ECC2353}.Release|Any CPU.Build.0 = Release|Any CPU
+		{75DDC845-838D-4F33-914B-2D833179A463}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{75DDC845-838D-4F33-914B-2D833179A463}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{75DDC845-838D-4F33-914B-2D833179A463}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{75DDC845-838D-4F33-914B-2D833179A463}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -71,6 +77,7 @@ Global
 		{62CDAE77-D687-423E-B88C-25967ECC2353} = {274A8CD4-5F80-4164-B0F6-1F2ACBB8E00B}
 		{56E62BAA-09E0-4119-AA48-0F3B6E80C1F7} = {300AD5BD-47CB-4AA8-B823-9B061C7654D1}
 		{7E5B0AC0-4A78-4BC2-9C54-E461317125AB} = {300AD5BD-47CB-4AA8-B823-9B061C7654D1}
+		{75DDC845-838D-4F33-914B-2D833179A463} = {81F37230-D7FE-4190-A939-B5AE8A71494B}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {B3DABD43-6E3B-4871-B426-2C4BDD6018CC}