CZY/MES-ETL
1
0
Fork 0
Code Pull Requests Releases Activity

整理代码

Browse Source
This commit is contained in:
lindj 2024-01-22 17:04:49 +08:00
parent fcc0de5b2a
commit a169eecec5
2 changed files with 35 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
ConsoleApp2/HostedServices/TransformService.cs
View File

@ -21,7 +21,7 @@ public class TransformService : ITransformService
private readonly IDistributedCache _cache;
public TransformService(ILogger<TransformService> logger,
public TransformService(ILogger<TransformService> logger,
IOptions<DataTransformOptions> options,
ProcessContext context,
IDistributedCache cache)
@ -35,30 +35,41 @@ public class TransformService : ITransformService
public async Task ExecuteAsync(TasksOptions tasksOptions, DataRecordQueue producerQueue, DataRecordQueue consumerQueue, ProcessContext context, CancellationToken cancellationToken)
{
_logger.LogInformation("***** Data transform service started, thread id: {ThreadId} *****", Environment.CurrentManagedThreadId);
while ((!context.IsInputCompleted || producerQueue.Count > 0))
while ((!context.IsInputCompleted || producerQueue.Count > 0))
{
if (_context.GetExceptions().Count > 0)
{
if (_context.GetExceptions().Count > 0)
{
_logger.LogInformation("***** Csv transform service is canceled *****");
return;
}
if (!producerQueue.TryDequeue(out var record)) continue;
_logger.LogInformation("***** Csv transform service is canceled *****");
return;
}
if (!producerQueue.TryDequeue(out var record)) continue;
//过滤不要的record
if (await _options.Value.RecordFilter?.Invoke(record, _cache) == false) continue;
record.Database = _options.Value.DatabaseFilter?.Invoke(record);
//修改record
_options.Value.RecordModify?.Invoke(record);
//缓存record
await _options.Value.RecordCache?.Invoke(record, _cache);
//替换record
var replaceRecord = await _options.Value.RecordReplace?.Invoke(record, _cache);
if (replaceRecord != null)
//过滤不要的record
if (_options.Value.RecordFilter != null)
{
var result = await _options.Value.RecordFilter.Invoke(record, _cache);
if (result == false) continue;
}
record.Database = _options.Value.DatabaseFilter?.Invoke(record);
//修改record
_options.Value.RecordModify?.Invoke(record);
//缓存record
if (_options.Value.RecordCache != null)
{
await _options.Value.RecordCache.Invoke(record, _cache);
}
//替换record
if (_options.Value.RecordReplace != null)
{
var result = await _options.Value.RecordReplace.Invoke(record, _cache);
if (result != null)
{
record = replaceRecord;
record = result;
}
consumerQueue.Enqueue(record);
_context.AddTransform();
}
consumerQueue.Enqueue(record);
_context.AddTransform();
//数据增加
var addRecords = _options.Value.RecordAdd?.Invoke(record);
if (addRecords != null && addRecords.Count > 0)
@ -70,8 +81,8 @@ public class TransformService : ITransformService
}
}
}
context.CompleteTransform();
context.CompleteTransform();
_logger.LogInformation("***** Data transformation service completed *****");
}
}

2
ConsoleApp2/Program.cs
View File

@ -445,7 +445,7 @@ async Task RunProgram()
host.Services.AddHostedService<MainHostedService>();
host.Services.AddHostedService<TaskMonitorService>();
if(commandOptions.IsMock)host.Services.AddSingleton<IInputService,SimulationInputService>();
if(commandOptions.IsMock)host.Services.AddSingleton<IInputService,InputService>();
else host.Services.AddSingleton<IInputService, InputService>();
host.Services.AddSingleton<ITransformService, TransformService>();
host.Services.AddSingleton<IOutputService, OutputService>();