blob: 6d8c2c001d535af6f0992b3a46c62b3ba61fc966 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
|
using Enmarcha.Backend.Types;
using Enmarcha.Backend.Types.Arrivals;
namespace Enmarcha.Backend.Services;
public class ArrivalsContext
{
/// <summary>
/// The full GTFS ID of the stop (e.g., "vitrasa:1400")
/// </summary>
public required string StopId { get; set; }
/// <summary>
/// The public code of the stop (e.g., "1400")
/// </summary>
public required string StopCode { get; set; }
/// <summary>
/// Whether to return a reduced number of arrivals (e.g., 4 instead of 10)
/// </summary>
public bool IsReduced { get; set; }
public Position? StopLocation { get; set; }
public required List<Arrival> Arrivals { get; set; }
public List<BusStopUsagePoint>? Usage { get; set; }
public required DateTime NowLocal { get; set; }
}
public interface IArrivalsProcessor
{
/// <summary>
/// Processes the arrivals in the context. Processors are executed in the order they are registered.
/// </summary>
Task ProcessAsync(ArrivalsContext context);
}
/// <summary>
/// Orchestrates the enrichment of arrival data through a series of processors.
/// This follows a pipeline pattern where each step (processor) adds or modifies data
/// in the shared ArrivalsContext.
/// </summary>
public class ArrivalsPipeline
{
private readonly IEnumerable<IArrivalsProcessor> _processors;
public ArrivalsPipeline(IEnumerable<IArrivalsProcessor> processors)
{
_processors = processors;
}
/// <summary>
/// Executes all registered processors sequentially.
/// </summary>
public async Task ExecuteAsync(ArrivalsContext context)
{
using var activity = Telemetry.Source.StartActivity("ArrivalsPipeline");
foreach (var processor in _processors)
{
using var processorActivity = Telemetry.Source.StartActivity($"Processor:{processor.GetType().Name}");
await processor.ProcessAsync(context);
}
}
}
|
