Compare commits

...

3 Commits

Author SHA1 Message Date
trifonovt a9b62d807d Optimize runtime event processing assembly and projection paths 2026-06-08 17:20:28 +02:00
trifonovt 3bfe05b64e Add runtime processing performance tracing 2026-06-08 15:00:54 +02:00
trifonovt 1d64bd92c7 Remove legacy tachograph runtime wrappers 2026-06-08 14:21:45 +02:00
37 changed files with 1584 additions and 570 deletions

View File

@ -2,7 +2,6 @@ package at.procon.eventhub.processing.api;
import at.procon.eventhub.processing.dto.UnifiedRuntimeDerivedProjectionResultDto;
import at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest;
import at.procon.eventhub.processing.dto.UnifiedRuntimeTachographEsperScopeResultDto;
import at.procon.eventhub.processing.eventprocessing.RuntimeEventProcessingService;
import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingApiRequest;
import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingResultDto;
@ -19,10 +18,6 @@ import at.procon.eventhub.processing.model.UnifiedRuntimeEventBundle;
import at.procon.eventhub.processing.service.UnifiedRuntimeDerivedProjectionService;
import at.procon.eventhub.processing.service.UnifiedRuntimeDriverTimelineService;
import at.procon.eventhub.processing.service.UnifiedRuntimeEventAssemblyService;
import at.procon.eventhub.processing.service.RuntimeDriverWorkingTimeScopeProcessingService;
import at.procon.eventhub.tachographfilesession.processing.validation.RuntimeTachographParityValidationApiRequest;
import at.procon.eventhub.tachographfilesession.processing.validation.RuntimeTachographParityValidationResultDto;
import at.procon.eventhub.tachographfilesession.processing.validation.RuntimeTachographParityValidationService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
@ -39,10 +34,8 @@ public class UnifiedRuntimeProcessingController {
private final UnifiedRuntimeEventAssemblyService eventAssemblyService;
private final UnifiedRuntimeDriverTimelineService runtimeDriverTimelineService;
private final UnifiedRuntimeDerivedProjectionService runtimeDerivedProjectionService;
private final RuntimeDriverWorkingTimeScopeProcessingService tachographEsperScopeProcessingService;
private final RuntimeEventProcessingService runtimeEventProcessingService;
private final RuntimeProcessingExecutionService runtimeProcessingExecutionService;
private final RuntimeTachographParityValidationService tachographParityValidationService;
private final RuntimeMixedSourceEvidenceValidationService mixedSourceEvidenceValidationService;
public UnifiedRuntimeProcessingController(
@ -50,18 +43,16 @@ public class UnifiedRuntimeProcessingController {
UnifiedRuntimeDriverTimelineService runtimeDriverTimelineService,
UnifiedRuntimeDerivedProjectionService runtimeDerivedProjectionService
) {
this(eventAssemblyService, runtimeDriverTimelineService, runtimeDerivedProjectionService, null, null, null, null, null);
this(eventAssemblyService, runtimeDriverTimelineService, runtimeDerivedProjectionService, null, null, null);
}
public UnifiedRuntimeProcessingController(
UnifiedRuntimeEventAssemblyService eventAssemblyService,
UnifiedRuntimeDriverTimelineService runtimeDriverTimelineService,
UnifiedRuntimeDerivedProjectionService runtimeDerivedProjectionService,
RuntimeDriverWorkingTimeScopeProcessingService tachographEsperScopeProcessingService,
RuntimeEventProcessingService runtimeEventProcessingService
) {
this(eventAssemblyService, runtimeDriverTimelineService, runtimeDerivedProjectionService,
tachographEsperScopeProcessingService, runtimeEventProcessingService, null, null, null);
this(eventAssemblyService, runtimeDriverTimelineService, runtimeDerivedProjectionService, runtimeEventProcessingService, null, null);
}
@Autowired
@ -69,35 +60,18 @@ public class UnifiedRuntimeProcessingController {
UnifiedRuntimeEventAssemblyService eventAssemblyService,
UnifiedRuntimeDriverTimelineService runtimeDriverTimelineService,
UnifiedRuntimeDerivedProjectionService runtimeDerivedProjectionService,
RuntimeDriverWorkingTimeScopeProcessingService tachographEsperScopeProcessingService,
RuntimeEventProcessingService runtimeEventProcessingService,
RuntimeTachographParityValidationService tachographParityValidationService,
RuntimeMixedSourceEvidenceValidationService mixedSourceEvidenceValidationService,
RuntimeProcessingExecutionService runtimeProcessingExecutionService
) {
this.eventAssemblyService = eventAssemblyService;
this.runtimeDriverTimelineService = runtimeDriverTimelineService;
this.runtimeDerivedProjectionService = runtimeDerivedProjectionService;
this.tachographEsperScopeProcessingService = tachographEsperScopeProcessingService;
this.runtimeEventProcessingService = runtimeEventProcessingService;
this.runtimeProcessingExecutionService = runtimeProcessingExecutionService;
this.tachographParityValidationService = tachographParityValidationService;
this.mixedSourceEvidenceValidationService = mixedSourceEvidenceValidationService;
}
public UnifiedRuntimeProcessingController(
UnifiedRuntimeEventAssemblyService eventAssemblyService,
UnifiedRuntimeDriverTimelineService runtimeDriverTimelineService,
UnifiedRuntimeDerivedProjectionService runtimeDerivedProjectionService,
RuntimeDriverWorkingTimeScopeProcessingService tachographEsperScopeProcessingService,
RuntimeEventProcessingService runtimeEventProcessingService,
RuntimeTachographParityValidationService tachographParityValidationService
) {
this(eventAssemblyService, runtimeDriverTimelineService, runtimeDerivedProjectionService,
tachographEsperScopeProcessingService, runtimeEventProcessingService,
tachographParityValidationService, null, null);
}
@PostMapping("/driver-events")
public ResponseEntity<UnifiedRuntimeEventBundle> loadDriverEvents(
@RequestBody UnifiedRuntimeProcessingApiRequest request
@ -157,17 +131,6 @@ public class UnifiedRuntimeProcessingController {
}
@PostMapping("/event-processing/validation/tachograph-parity")
public ResponseEntity<RuntimeTachographParityValidationResultDto> validateTachographParity(
@RequestBody RuntimeTachographParityValidationApiRequest request
) {
if (tachographParityValidationService == null) {
throw new IllegalStateException("Runtime tachograph parity validation service is not configured.");
}
return ResponseEntity.ok(tachographParityValidationService.validate(request));
}
@PostMapping("/event-processing/validation/mixed-source-evidence")
public ResponseEntity<RuntimeMixedSourceEvidenceValidationResultDto> validateMixedSourceEvidence(
@RequestBody RuntimeMixedSourceEvidenceValidationApiRequest request
@ -177,22 +140,4 @@ public class UnifiedRuntimeProcessingController {
}
return ResponseEntity.ok(mixedSourceEvidenceValidationService.validate(request));
}
@PostMapping("/tachograph/esper-processing")
public ResponseEntity<UnifiedRuntimeTachographEsperScopeResultDto> runTachographEsperProcessing(
@RequestBody UnifiedRuntimeProcessingApiRequest request
) {
if (runtimeEventProcessingService != null) {
RuntimeEventProcessingResultDto genericResult = runtimeEventProcessingService.process(
RuntimeEventProcessingApiRequest.tachographDriverEsper(request)
);
return ResponseEntity.ok(UnifiedRuntimeTachographEsperScopeResultDto.fromGenericRuntimeEventProcessingResult(genericResult));
}
if (tachographEsperScopeProcessingService == null) {
throw new IllegalStateException("Tachograph Esper scope processing service is not configured.");
}
return ResponseEntity.ok(UnifiedRuntimeTachographEsperScopeResultDto.fromDriverWorkingTime(
tachographEsperScopeProcessingService.processScope(request)
));
}
}

View File

@ -19,6 +19,8 @@ import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Objects;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
/**
@ -32,6 +34,8 @@ import org.springframework.stereotype.Service;
@Service
public class DriverWorkingTimeProcessingCore {
private static final Logger LOG = LoggerFactory.getLogger(DriverWorkingTimeProcessingCore.class);
private final DriverWorkingTimeDerivedProjectionEngine derivedProjectionEngine;
public DriverWorkingTimeProcessingCore(
@ -42,6 +46,7 @@ public class DriverWorkingTimeProcessingCore {
public DriverWorkingTimeProcessingResultDto process(DriverWorkingTimeProcessingInput input) {
Objects.requireNonNull(input, "input must not be null");
long startedAtNanos = System.nanoTime();
OffsetDateTime loadedFrom = input.loadedFrom();
OffsetDateTime loadedTo = input.loadedTo();
String driverKey = input.driverKey();
@ -51,6 +56,7 @@ public class DriverWorkingTimeProcessingCore {
throw new IllegalArgumentException("occurredTo must not be before occurredFrom.");
}
long activityClipStartedAtNanos = System.nanoTime();
List<DriverWorkingTimeActivityInterval> activityIntervals = clipActivityIntervals(
input.activityIntervals(),
requestedFrom,
@ -63,9 +69,13 @@ public class DriverWorkingTimeProcessingCore {
requestedFrom,
requestedTo
);
long activityClipMs = elapsedMillis(activityClipStartedAtNanos);
long derivedProjectionBuildStartedAtNanos = System.nanoTime();
DriverWorkingTimeDerivedProjectionBundle derivedProjectionBundle = derivedProjectionEngine.build(input);
long derivedProjectionBuildMs = elapsedMillis(derivedProjectionBuildStartedAtNanos);
long derivedIntervalClipStartedAtNanos = System.nanoTime();
List<DriverWorkingTimeDrivingInterruptionInterval> drivingInterruptionIntervals =
clipDrivingInterruptionIntervals(
derivedProjectionBundle.drivingInterruptionIntervals(),
@ -131,8 +141,9 @@ public class DriverWorkingTimeProcessingCore {
requestedFrom,
requestedTo
);
long derivedIntervalClipMs = elapsedMillis(derivedIntervalClipStartedAtNanos);
return new DriverWorkingTimeProcessingResultDto(
DriverWorkingTimeProcessingResultDto result = new DriverWorkingTimeProcessingResultDto(
input.sessionId(),
driverKey,
input.sourceKind(),
@ -168,6 +179,19 @@ public class DriverWorkingTimeProcessingCore {
supportGeoEvents,
combinedNotes(input.notes())
);
LOG.info(
"Driver working-time core processed driver {} in {} ms (activityClipMs: {}, derivedProjectionBuildMs: {}, derivedIntervalClipMs: {}, activityIntervals: {}, drivingIntervals: {}, vehicleUsageIntervals: {}, supportGeoEvents: {})",
driverKey,
elapsedMillis(startedAtNanos),
activityClipMs,
derivedProjectionBuildMs,
derivedIntervalClipMs,
result.activityIntervalCount(),
result.drivingIntervalCount(),
result.vehicleUsageIntervalCount(),
result.supportGeoEventCount()
);
return result;
}
public DriverWorkingTimeProcessingResultDto processDriverWorkingTime(DriverWorkingTimeProcessingInput input) {
@ -649,4 +673,8 @@ public class DriverWorkingTimeProcessingCore {
}
return left.isBefore(right) ? left : right;
}
private long elapsedMillis(long startedAtNanos) {
return Math.max(0L, (System.nanoTime() - startedAtNanos) / 1_000_000L);
}
}

View File

@ -26,15 +26,22 @@ import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Collections;
import java.util.Deque;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.stereotype.Service;
@ -43,11 +50,41 @@ import org.springframework.util.StreamUtils;
@Service
public class DriverWorkingTimeReusableProjectionBuilder {
private static final Logger LOG = LoggerFactory.getLogger(DriverWorkingTimeReusableProjectionBuilder.class);
private static final AtomicLong RUNTIME_COUNTER = new AtomicLong();
private static final String DRIVING_DERIVED_PROJECTION_BUNDLE_EPL_TEMPLATE =
loadResource("esper/driver-working-time-derived-projections.epl");
private static final Map<String, Object> ACTIVITY_INTERVAL_INPUT_DEFINITION = activityIntervalInputDefinitionStatic();
private static final Map<String, Object> VEHICLE_USAGE_INTERVAL_INPUT_DEFINITION = vehicleUsageIntervalInputDefinitionStatic();
private static final Map<String, Object> SUPPORT_GEO_EVIDENCE_INPUT_DEFINITION = supportGeoEvidenceInputDefinitionStatic();
private static final int MAX_IDLE_RUNTIMES_PER_DEFINITION = 2;
private static final List<String> REUSABLE_RUNTIME_STATE_CLEANUP_QUERIES = List.of(
"delete from PreviousRestCandidateCoverageInterval",
"delete from OpenPotentialInVehicleTripState",
"delete from SupportGeoEvidenceWindow",
"delete from DailyWeeklyRestCandidateBeginGeoEvidenceCandidateWindow",
"delete from DailyWeeklyRestCandidateBeginGeoEvidenceBestScoreWindow",
"delete from DailyWeeklyRestCandidateBeginGeoEvidenceResolvedWindow",
"delete from DailyWeeklyRestCandidateEndGeoEvidenceCandidateWindow",
"delete from DailyWeeklyRestCandidateEndGeoEvidenceBestScoreWindow",
"delete from DailyWeeklyRestCandidateEndGeoEvidenceResolvedWindow",
"delete from DailyWeeklyRestCandidateBeginBoundaryOdometerCandidateWindow",
"delete from DailyWeeklyRestCandidateBeginBoundaryOdometerBestScoreWindow",
"delete from DailyWeeklyRestCandidateBeginBoundaryOdometerResolvedWindow",
"delete from DailyWeeklyRestCandidateEndBoundaryOdometerCandidateWindow",
"delete from DailyWeeklyRestCandidateEndBoundaryOdometerBestScoreWindow",
"delete from DailyWeeklyRestCandidateEndBoundaryOdometerResolvedWindow",
"delete from DailyWeeklyRestCandidateCoverageCardResolvedIntervalWindow",
"delete from DailyWeeklyRestCandidateCoverageEmittedKeyWindow",
"delete from PreviousSignificantDrivingInterval",
"context PerDriver delete from PreviousVehicleUsageInterval"
);
private final EventHubProperties properties;
private final ConcurrentMap<PreparedProjectionDefinitionCacheKey, PreparedProjectionDefinition> preparedDefinitions =
new ConcurrentHashMap<>();
private final ConcurrentMap<PreparedProjectionDefinitionCacheKey, ReusableProjectionRuntimePool> reusableRuntimePools =
new ConcurrentHashMap<>();
@Autowired
public DriverWorkingTimeReusableProjectionBuilder(
@ -78,6 +115,7 @@ public class DriverWorkingTimeReusableProjectionBuilder {
int significantDrivingMinutes,
int minimumRestPeriodMinutes
) {
long startedAtNanos = System.nanoTime();
if ((activityInputEvents == null || activityInputEvents.isEmpty())
&& (vehicleUsageInputEvents == null || vehicleUsageInputEvents.isEmpty())) {
return emptyDerivedProjectionBundle();
@ -93,22 +131,9 @@ public class DriverWorkingTimeReusableProjectionBuilder {
List<DriverWorkingTimeRestCoverageInterval> potentialInVehicleOvernightStayIntervals = new ArrayList<>();
List<DriverWorkingTimePotentialInVehicleTripInterval> potentialInVehicleTripIntervals = new ArrayList<>();
executeWithRuntime(
configuration -> {
configuration.getCommon().addEventType(
"TachographActivityIntervalInputEvent",
activityIntervalInputDefinition()
);
configuration.getCommon().addEventType(
"TachographVehicleUsageIntervalInputEvent",
vehicleUsageIntervalInputDefinition()
);
configuration.getCommon().addEventType(
"TachographSupportGeoEvidenceInputEvent",
supportGeoEvidenceInputDefinition()
);
},
renderDrivingDerivedProjectionBundleEpl(significantDrivingMinutes, minimumRestPeriodMinutes),
DerivedProjectionRuntimeExecutionMetrics runtimeMetrics = executeWithRuntime(
significantDrivingMinutes,
minimumRestPeriodMinutes,
Map.of(
"drivingInterruptionIntervals",
newData -> collectDrivingInterruptionIntervalModels(newData, drivingInterruptionIntervals),
@ -129,34 +154,26 @@ public class DriverWorkingTimeReusableProjectionBuilder {
"potentialInVehicleTripIntervals",
newData -> collectPotentialInVehicleTripIntervalModels(newData, potentialInVehicleTripIntervals)
),
runtime -> {
sender -> {
if (supportGeoInputEvents != null) {
for (Map<String, Object> supportGeoEvidence : supportGeoInputEvents) {
runtime.getEventService().sendEventMap(
supportGeoEvidence,
"TachographSupportGeoEvidenceInputEvent"
);
sender.sendSupportGeoEvent(supportGeoEvidence);
}
}
if (vehicleUsageInputEvents != null) {
for (Map<String, Object> interval : vehicleUsageInputEvents) {
runtime.getEventService().sendEventMap(
interval,
"TachographVehicleUsageIntervalInputEvent"
);
sender.sendVehicleUsageEvent(interval);
}
}
if (activityInputEvents != null) {
for (Map<String, Object> interval : activityInputEvents) {
runtime.getEventService().sendEventMap(
interval,
"TachographActivityIntervalInputEvent"
);
sender.sendActivityEvent(interval);
}
}
}
);
long sortOutputStartedAtNanos = System.nanoTime();
List<DriverWorkingTimeRestCoverageInterval> sortedDailyWeeklyRestCandidateCoverageIntervals =
sortDailyWeeklyRestCandidateCoverageIntervalsCommon(dailyWeeklyRestCandidateCoverageIntervals);
List<DriverWorkingTimeRestCoverageInterval> sortedUnclassifiedDailyWeeklyRestCandidateCoverageIntervals =
@ -165,6 +182,27 @@ public class DriverWorkingTimeReusableProjectionBuilder {
sortPotentialHomeOvernightStayIntervalsCommon(potentialHomeOvernightStayIntervals);
List<DriverWorkingTimeRestCoverageInterval> sortedPotentialInVehicleOvernightStayIntervals =
sortPotentialInVehicleOvernightStayIntervalsCommon(potentialInVehicleOvernightStayIntervals);
long sortOutputMs = elapsedMillis(sortOutputStartedAtNanos);
LOG.info(
"Driver working-time derived projection bundle built in {} ms (definitionCacheHit: {}, definitionPreparationMs: {}, runtimePoolHit: {}, runtimeInitMs: {}, deployMs: {}, listenerRegistrationMs: {}, runtimeResetMs: {}, sendSupportGeoMs: {}, sendVehicleUsageMs: {}, sendActivityMs: {}, destroyMs: {}, sortOutputMs: {}, activityInputEvents: {}, vehicleUsageInputEvents: {}, supportGeoInputEvents: {})",
elapsedMillis(startedAtNanos),
runtimeMetrics.definitionCacheHit(),
runtimeMetrics.definitionPreparationMs(),
runtimeMetrics.runtimePoolHit(),
runtimeMetrics.runtimeInitMs(),
runtimeMetrics.deployMs(),
runtimeMetrics.listenerRegistrationMs(),
runtimeMetrics.runtimeResetMs(),
runtimeMetrics.sendSupportGeoMs(),
runtimeMetrics.sendVehicleUsageMs(),
runtimeMetrics.sendActivityMs(),
runtimeMetrics.destroyMs(),
sortOutputMs,
safeList(activityInputEvents).size(),
safeList(vehicleUsageInputEvents).size(),
safeList(supportGeoInputEvents).size()
);
return new DriverWorkingTimeDerivedProjectionBundle(
sortDrivingInterruptionIntervalsCommon(drivingInterruptionIntervals),
@ -237,39 +275,180 @@ public class DriverWorkingTimeReusableProjectionBuilder {
);
}
private void executeWithRuntime(
Consumer<Configuration> configurationSetup,
String epl,
private DerivedProjectionRuntimeExecutionMetrics executeWithRuntime(
int significantDrivingMinutes,
int minimumRestPeriodMinutes,
Map<String, Consumer<EventBean[]>> listeners,
Consumer<EPRuntime> sender
Consumer<DerivedProjectionEventSender> sender
) {
long definitionPreparationMs = 0L;
boolean definitionCacheHit = false;
boolean runtimePoolHit = false;
long runtimeInitMs = 0L;
long deployMs = 0L;
long listenerRegistrationMs = 0L;
long runtimeResetMs = 0L;
long sendSupportGeoMs = 0L;
long sendVehicleUsageMs = 0L;
long sendActivityMs = 0L;
long destroyMs = 0L;
ReusableProjectionRuntimePool runtimePool = null;
ReusableProjectionRuntime reusableRuntime = null;
boolean discardRuntime = false;
try {
PreparedProjectionDefinitionCacheKey cacheKey = new PreparedProjectionDefinitionCacheKey(
significantDrivingMinutes,
minimumRestPeriodMinutes,
Math.max(1, properties.getRuntimeProcessing().getRestCandidateGeoLookbackMinutes()),
Math.max(1, properties.getRuntimeProcessing().getRestCandidateGeoLookaheadMinutes()),
Math.max(0, properties.getRuntimeProcessing().getRestCandidateGeoStationaryMaxMeters()),
Math.max(
properties.getRuntimeProcessing().getRestCandidateGeoStationaryMaxMeters(),
properties.getRuntimeProcessing().getRestCandidateGeoMinorMovementMaxMeters()
)
);
PreparedProjectionDefinition preparedDefinition = preparedDefinitions.get(cacheKey);
definitionCacheHit = preparedDefinition != null;
if (!definitionCacheHit) {
long definitionPreparationStartedAtNanos = System.nanoTime();
PreparedProjectionDefinition candidate = prepareProjectionDefinition(
significantDrivingMinutes,
minimumRestPeriodMinutes
);
PreparedProjectionDefinition existing = preparedDefinitions.putIfAbsent(cacheKey, candidate);
preparedDefinition = existing == null ? candidate : existing;
definitionCacheHit = existing != null;
definitionPreparationMs = definitionCacheHit ? 0L : elapsedMillis(definitionPreparationStartedAtNanos);
}
runtimePool = reusableRuntimePools.computeIfAbsent(cacheKey, ignored -> new ReusableProjectionRuntimePool());
reusableRuntime = runtimePool.acquire(preparedDefinition, listeners);
runtimePoolHit = reusableRuntime.poolHit();
runtimeInitMs = reusableRuntime.runtimeInitMs();
deployMs = reusableRuntime.deployMs();
listenerRegistrationMs = reusableRuntime.listenerRegistrationMs();
ReusableProjectionRuntimeExecution execution = reusableRuntime.execute(listeners, sender);
runtimeResetMs = execution.runtimeResetMs();
sendSupportGeoMs = execution.sendSupportGeoMs();
sendVehicleUsageMs = execution.sendVehicleUsageMs();
sendActivityMs = execution.sendActivityMs();
} catch (EPCompileException | EPDeployException e) {
discardRuntime = true;
throw new IllegalStateException("Cannot compile/deploy reusable driver working-time projection EPL bundle", e);
} catch (RuntimeException e) {
discardRuntime = true;
throw e;
} finally {
if (reusableRuntime != null) {
if (discardRuntime || runtimePool == null) {
destroyMs = reusableRuntime.destroy();
} else {
destroyMs = runtimePool.release(reusableRuntime);
}
}
}
return new DerivedProjectionRuntimeExecutionMetrics(
definitionCacheHit,
definitionPreparationMs,
runtimePoolHit,
runtimeInitMs,
deployMs,
listenerRegistrationMs,
runtimeResetMs,
sendSupportGeoMs,
sendVehicleUsageMs,
sendActivityMs,
destroyMs
);
}
private ReusableProjectionRuntime createReusableRuntime(
PreparedProjectionDefinition preparedDefinition,
Map<String, Consumer<EventBean[]>> listeners
) throws EPDeployException, EPCompileException {
EPRuntime runtime = null;
try {
Configuration configuration = new Configuration();
configurationSetup.accept(configuration);
long runtimeInitStartedAtNanos = System.nanoTime();
Configuration configuration = createRuntimeConfiguration();
String runtimeUri = "eventhub-driver-working-time-reusable-projection-" + RUNTIME_COUNTER.incrementAndGet();
runtime = EPRuntimeProvider.getRuntime(runtimeUri, configuration);
long runtimeInitMs = elapsedMillis(runtimeInitStartedAtNanos);
CompilerArguments arguments = new CompilerArguments(configuration);
EPCompiled compiled = EPCompilerProvider.getCompiler().compile(epl, arguments);
EPDeployment deployment = runtime.getDeploymentService().deploy(compiled);
for (Map.Entry<String, Consumer<EventBean[]>> entry : listeners.entrySet()) {
long deployStartedAtNanos = System.nanoTime();
EPDeployment deployment = runtime.getDeploymentService().deploy(preparedDefinition.compiled());
long deployMs = elapsedMillis(deployStartedAtNanos);
ReusableProjectionRuntime reusableRuntime = new ReusableProjectionRuntime(runtime, runtimeInitMs, deployMs, false);
long listenerRegistrationStartedAtNanos = System.nanoTime();
for (String statementName : listeners.keySet()) {
runtime.getDeploymentService()
.getStatement(deployment.getDeploymentId(), entry.getKey())
.addListener((newData, oldData, statement, rt) -> entry.getValue().accept(newData));
.getStatement(deployment.getDeploymentId(), statementName)
.addListener((newData, oldData, statement, rt) -> reusableRuntime.onStatementEvents(statement.getName(), newData));
}
sender.accept(runtime);
} catch (EPCompileException | EPDeployException e) {
throw new IllegalStateException("Cannot compile/deploy reusable driver working-time projection EPL bundle", e);
} finally {
if (runtime != null) {
reusableRuntime.listenerRegistrationMs(elapsedMillis(listenerRegistrationStartedAtNanos));
reusableRuntime.cleanupQueries(compileCleanupQueries(runtime));
return reusableRuntime;
} catch (EPDeployException | EPCompileException | RuntimeException ex) {
if (runtime != null && !runtime.isDestroyed()) {
runtime.destroy();
}
throw ex;
}
}
private List<EPCompiled> compileCleanupQueries(EPRuntime runtime) throws EPCompileException {
CompilerArguments arguments = new CompilerArguments(runtime.getConfigurationDeepCopy());
arguments.getPath().add(runtime.getRuntimePath());
List<EPCompiled> compiledQueries = new ArrayList<>(REUSABLE_RUNTIME_STATE_CLEANUP_QUERIES.size());
for (String cleanupQuery : REUSABLE_RUNTIME_STATE_CLEANUP_QUERIES) {
compiledQueries.add(EPCompilerProvider.getCompiler().compileQuery(cleanupQuery, arguments));
}
return List.copyOf(compiledQueries);
}
private Map<String, Object> activityIntervalInputDefinition() {
return ACTIVITY_INTERVAL_INPUT_DEFINITION;
}
private Map<String, Object> vehicleUsageIntervalInputDefinition() {
return VEHICLE_USAGE_INTERVAL_INPUT_DEFINITION;
}
private Map<String, Object> supportGeoEvidenceInputDefinition() {
return SUPPORT_GEO_EVIDENCE_INPUT_DEFINITION;
}
private PreparedProjectionDefinition prepareProjectionDefinition(
int significantDrivingMinutes,
int minimumRestPeriodMinutes
) throws EPCompileException {
Configuration configuration = createRuntimeConfiguration();
String epl = renderDrivingDerivedProjectionBundleEpl(significantDrivingMinutes, minimumRestPeriodMinutes);
CompilerArguments arguments = new CompilerArguments(configuration);
EPCompiled compiled = EPCompilerProvider.getCompiler().compile(epl, arguments);
return new PreparedProjectionDefinition(compiled);
}
private Configuration createRuntimeConfiguration() {
Configuration configuration = new Configuration();
configuration.getCommon().addEventType(
"TachographActivityIntervalInputEvent",
activityIntervalInputDefinition()
);
configuration.getCommon().addEventType(
"TachographVehicleUsageIntervalInputEvent",
vehicleUsageIntervalInputDefinition()
);
configuration.getCommon().addEventType(
"TachographSupportGeoEvidenceInputEvent",
supportGeoEvidenceInputDefinition()
);
return configuration;
}
private static Map<String, Object> activityIntervalInputDefinitionStatic() {
Map<String, Object> definition = new LinkedHashMap<>();
definition.put("sessionId", UUID.class);
definition.put("driverKey", String.class);
@ -292,10 +471,10 @@ public class DriverWorkingTimeReusableProjectionBuilder {
definition.put("synthetic", boolean.class);
definition.put("clippedToRequestedPeriod", boolean.class);
definition.put("level", String.class);
return definition;
return Collections.unmodifiableMap(definition);
}
private Map<String, Object> vehicleUsageIntervalInputDefinition() {
private static Map<String, Object> vehicleUsageIntervalInputDefinitionStatic() {
Map<String, Object> definition = new LinkedHashMap<>();
definition.put("sessionId", UUID.class);
definition.put("driverKey", String.class);
@ -313,10 +492,10 @@ public class DriverWorkingTimeReusableProjectionBuilder {
definition.put("vehicleKey", String.class);
definition.put("sourceKind", String.class);
definition.put("sourceIntervalIds", java.util.List.class);
return definition;
return Collections.unmodifiableMap(definition);
}
private Map<String, Object> supportGeoEvidenceInputDefinition() {
private static Map<String, Object> supportGeoEvidenceInputDefinitionStatic() {
Map<String, Object> definition = new LinkedHashMap<>();
definition.put("sessionId", UUID.class);
definition.put("driverKey", String.class);
@ -330,7 +509,7 @@ public class DriverWorkingTimeReusableProjectionBuilder {
definition.put("longitude", Double.class);
definition.put("odometerKm", Long.class);
definition.put("priority", int.class);
return definition;
return Collections.unmodifiableMap(definition);
}
private Map<String, Object> toActivityIntervalInputMap(
@ -758,6 +937,10 @@ public class DriverWorkingTimeReusableProjectionBuilder {
return value instanceof Long number ? number : null;
}
private long elapsedMillis(long startedAtNanos) {
return Math.max(0L, (System.nanoTime() - startedAtNanos) / 1_000_000L);
}
private static String loadResource(String path) {
try {
ClassPathResource resource = new ClassPathResource(path);
@ -766,4 +949,230 @@ public class DriverWorkingTimeReusableProjectionBuilder {
throw new IllegalStateException("Cannot load EPL resource: " + path, e);
}
}
private record PreparedProjectionDefinition(
EPCompiled compiled
) {
}
private record PreparedProjectionDefinitionCacheKey(
int significantDrivingMinutes,
int minimumRestPeriodMinutes,
int restGeoLookbackMinutes,
int restGeoLookaheadMinutes,
int restGeoStationaryMaxMeters,
int restGeoMinorMovementMaxMeters
) {
}
private record DerivedProjectionRuntimeExecutionMetrics(
boolean definitionCacheHit,
long definitionPreparationMs,
boolean runtimePoolHit,
long runtimeInitMs,
long deployMs,
long listenerRegistrationMs,
long runtimeResetMs,
long sendSupportGeoMs,
long sendVehicleUsageMs,
long sendActivityMs,
long destroyMs
) {
}
private final class ReusableProjectionRuntimePool {
private final Deque<ReusableProjectionRuntime> idleRuntimes = new ArrayDeque<>();
private synchronized ReusableProjectionRuntime acquire(
PreparedProjectionDefinition preparedDefinition,
Map<String, Consumer<EventBean[]>> listeners
) throws EPDeployException, EPCompileException {
ReusableProjectionRuntime runtime = idleRuntimes.pollFirst();
if (runtime != null) {
return runtime.reused();
}
return createReusableRuntime(preparedDefinition, listeners);
}
private long release(ReusableProjectionRuntime runtime) {
if (runtime == null) {
return 0L;
}
synchronized (this) {
if (idleRuntimes.size() < MAX_IDLE_RUNTIMES_PER_DEFINITION) {
runtime.poolHit(false);
idleRuntimes.addLast(runtime);
return 0L;
}
}
return runtime.destroy();
}
}
private static final class ReusableProjectionRuntime {
private final EPRuntime runtime;
private final long runtimeInitMs;
private final long deployMs;
private volatile long listenerRegistrationMs;
private volatile List<EPCompiled> cleanupQueries = List.of();
private volatile boolean poolHit;
private ExecutionListeners currentExecutionListeners;
private ReusableProjectionRuntime(
EPRuntime runtime,
long runtimeInitMs,
long deployMs,
boolean poolHit
) {
this.runtime = runtime;
this.runtimeInitMs = runtimeInitMs;
this.deployMs = deployMs;
this.poolHit = poolHit;
}
private synchronized ReusableProjectionRuntimeExecution execute(
Map<String, Consumer<EventBean[]>> listeners,
Consumer<DerivedProjectionEventSender> sender
) {
currentExecutionListeners = new ExecutionListeners(listeners);
long runtimeResetStartedAtNanos = System.nanoTime();
for (EPCompiled cleanupQuery : cleanupQueries) {
runtime.getFireAndForgetService().executeQuery(cleanupQuery);
}
long runtimeResetMs = elapsedMillisStatic(runtimeResetStartedAtNanos);
try {
DerivedProjectionEventSender timedSender = new DerivedProjectionEventSender(runtime);
sender.accept(timedSender);
return new ReusableProjectionRuntimeExecution(
runtimeResetMs,
timedSender.sendSupportGeoMs(),
timedSender.sendVehicleUsageMs(),
timedSender.sendActivityMs()
);
} finally {
currentExecutionListeners = null;
}
}
private void onStatementEvents(String statementName, EventBean[] newData) {
ExecutionListeners listeners = currentExecutionListeners;
if (listeners == null || newData == null || statementName == null) {
return;
}
Consumer<EventBean[]> consumer = listeners.listeners().get(statementName);
if (consumer != null) {
consumer.accept(newData);
}
}
private long destroy() {
if (runtime == null || runtime.isDestroyed()) {
return 0L;
}
long destroyStartedAtNanos = System.nanoTime();
runtime.destroy();
return elapsedMillisStatic(destroyStartedAtNanos);
}
private ReusableProjectionRuntime reused() {
poolHit = true;
return this;
}
private void cleanupQueries(List<EPCompiled> value) {
cleanupQueries = value == null ? List.of() : value;
}
private void listenerRegistrationMs(long value) {
listenerRegistrationMs = value;
}
private void poolHit(boolean value) {
poolHit = value;
}
private boolean poolHit() {
return poolHit;
}
private long runtimeInitMs() {
return poolHit ? 0L : runtimeInitMs;
}
private long deployMs() {
return poolHit ? 0L : deployMs;
}
private long listenerRegistrationMs() {
return poolHit ? 0L : listenerRegistrationMs;
}
}
private record ExecutionListeners(
Map<String, Consumer<EventBean[]>> listeners
) {
private ExecutionListeners {
listeners = listeners == null ? Map.of() : Map.copyOf(listeners);
}
}
private record ReusableProjectionRuntimeExecution(
long runtimeResetMs,
long sendSupportGeoMs,
long sendVehicleUsageMs,
long sendActivityMs
) {
}
private static final class DerivedProjectionEventSender {
private final EPRuntime delegate;
private long sendSupportGeoMs;
private long sendVehicleUsageMs;
private long sendActivityMs;
private DerivedProjectionEventSender(EPRuntime delegate) {
this.delegate = delegate;
}
private void sendSupportGeoEvent(Map<String, Object> event) {
long startedAtNanos = System.nanoTime();
delegate.getEventService().sendEventMap(event, "TachographSupportGeoEvidenceInputEvent");
sendSupportGeoMs += elapsedMillisStatic(startedAtNanos);
}
private void sendVehicleUsageEvent(Map<String, Object> event) {
long startedAtNanos = System.nanoTime();
delegate.getEventService().sendEventMap(event, "TachographVehicleUsageIntervalInputEvent");
sendVehicleUsageMs += elapsedMillisStatic(startedAtNanos);
}
private void sendActivityEvent(Map<String, Object> event) {
long startedAtNanos = System.nanoTime();
delegate.getEventService().sendEventMap(event, "TachographActivityIntervalInputEvent");
sendActivityMs += elapsedMillisStatic(startedAtNanos);
}
private long sendSupportGeoMs() {
return sendSupportGeoMs;
}
private long sendVehicleUsageMs() {
return sendVehicleUsageMs;
}
private long sendActivityMs() {
return sendActivityMs;
}
private static long elapsedMillisStatic(long startedAtNanos) {
return Math.max(0L, (System.nanoTime() - startedAtNanos) / 1_000_000L);
}
}
private static long elapsedMillisStatic(long startedAtNanos) {
return Math.max(0L, (System.nanoTime() - startedAtNanos) / 1_000_000L);
}
}

View File

@ -1,93 +0,0 @@
package at.procon.eventhub.processing.dto;
import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingPartitionResultDto;
import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingResultDto;
import at.procon.eventhub.processing.model.UnifiedDiscoveredVehicleRef;
import at.procon.eventhub.processing.model.UnifiedRuntimeProcessingRequest;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
public record UnifiedRuntimeTachographEsperScopeResultDto(
UnifiedRuntimeProcessingRequest request,
int inputEventCount,
int selectedDriverCount,
int discoveredVehicleCount,
List<UnifiedDiscoveredVehicleRef> discoveredVehicles,
Map<String, UnifiedRuntimeDerivedProjectionResultDto> driverResults,
Map<String, RuntimeDriverPartitionDebugDto> partitionDebugByDriver,
List<String> notes,
List<String> warnings
) {
public UnifiedRuntimeTachographEsperScopeResultDto {
discoveredVehicles = discoveredVehicles == null ? List.of() : List.copyOf(discoveredVehicles);
driverResults = driverResults == null ? Map.of() : Collections.unmodifiableMap(new LinkedHashMap<>(driverResults));
partitionDebugByDriver = partitionDebugByDriver == null ? Map.of() : Collections.unmodifiableMap(new LinkedHashMap<>(partitionDebugByDriver));
notes = notes == null ? List.of() : List.copyOf(notes);
warnings = warnings == null ? List.of() : List.copyOf(warnings);
}
public static UnifiedRuntimeTachographEsperScopeResultDto fromDriverWorkingTime(
UnifiedRuntimeDriverWorkingTimeScopeResultDto result
) {
if (result == null) {
return null;
}
return new UnifiedRuntimeTachographEsperScopeResultDto(
result.request(),
result.inputEventCount(),
result.selectedDriverCount(),
result.discoveredVehicleCount(),
result.discoveredVehicles(),
result.driverResults(),
result.partitionDebugByDriver(),
result.notes(),
result.warnings()
);
}
public static UnifiedRuntimeTachographEsperScopeResultDto fromGenericRuntimeEventProcessingResult(
RuntimeEventProcessingResultDto genericResult
) {
if (genericResult == null) {
throw new IllegalArgumentException("genericResult must not be null");
}
LinkedHashMap<String, UnifiedRuntimeDerivedProjectionResultDto> driverResults = new LinkedHashMap<>();
for (Map.Entry<String, RuntimeEventProcessingPartitionResultDto> entry : genericResult.partitionResults().entrySet()) {
Object value = entry.getValue().result();
if (value instanceof UnifiedRuntimeDerivedProjectionResultDto projectionResult) {
driverResults.put(entry.getKey(), projectionResult);
} else {
throw new IllegalArgumentException("Cannot convert generic partition result for key "
+ entry.getKey() + " to UnifiedRuntimeDerivedProjectionResultDto: "
+ (value == null ? "null" : value.getClass().getName()));
}
}
return new UnifiedRuntimeTachographEsperScopeResultDto(
genericResult.request(),
genericResult.inputEventCount(),
genericResult.selectedPartitionCount(),
genericResult.discoveredVehicleCount(),
genericResult.discoveredVehicles(),
driverResults,
extractPartitionDebug(genericResult),
genericResult.notes(),
genericResult.warnings()
);
}
private static Map<String, RuntimeDriverPartitionDebugDto> extractPartitionDebug(
RuntimeEventProcessingResultDto genericResult
) {
LinkedHashMap<String, RuntimeDriverPartitionDebugDto> debugByDriver = new LinkedHashMap<>();
for (Map.Entry<String, RuntimeEventProcessingPartitionResultDto> entry : genericResult.partitionResults().entrySet()) {
Object debug = entry.getValue().metadata().get("partitionDebug");
if (debug instanceof RuntimeDriverPartitionDebugDto partitionDebug) {
debugByDriver.put(entry.getKey(), partitionDebug);
}
}
return debugByDriver;
}
}

View File

@ -3,13 +3,17 @@ package at.procon.eventhub.processing.eventprocessing;
import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingApiRequest;
import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingResultDto;
import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingProfileDescriptorDto;
import java.util.List;
import at.procon.eventhub.processing.eventprocessing.profile.RuntimeEventProcessingProfileRegistry;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
@Service
public class RuntimeEventProcessingService {
private static final Logger LOG = LoggerFactory.getLogger(RuntimeEventProcessingService.class);
private final RuntimeEventProcessingProfileRegistry profileRegistry;
public RuntimeEventProcessingService(RuntimeEventProcessingProfileRegistry profileRegistry) {
@ -17,10 +21,34 @@ public class RuntimeEventProcessingService {
}
public RuntimeEventProcessingResultDto process(RuntimeEventProcessingApiRequest request) {
return profileRegistry.require(request.profileKey()).process(request);
long startedAtNanos = System.nanoTime();
try {
RuntimeEventProcessingResultDto result = profileRegistry.require(request.profileKey()).process(request);
LOG.info(
"Runtime event processing profile {} completed in {} ms (partitions: {}, inputEvents: {}, warnings: {})",
request.profileKey(),
elapsedMillis(startedAtNanos),
result.selectedPartitionCount(),
result.inputEventCount(),
result.warnings().size()
);
return result;
} catch (RuntimeException ex) {
LOG.error(
"Runtime event processing profile {} failed after {} ms",
request == null ? null : request.profileKey(),
elapsedMillis(startedAtNanos),
ex
);
throw ex;
}
}
public List<RuntimeEventProcessingProfileDescriptorDto> listProfiles() {
return profileRegistry.profileDescriptors();
}
private long elapsedMillis(long startedAtNanos) {
return Math.max(0L, (System.nanoTime() - startedAtNanos) / 1_000_000L);
}
}

View File

@ -1,6 +1,7 @@
package at.procon.eventhub.processing.eventprocessing.dto;
import at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest;
import at.procon.eventhub.processing.eventprocessing.plan.DriverWorkingTimeRuntimeProcessingPlan;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
@ -27,9 +28,9 @@ public record RuntimeEventProcessingApiRequest(
: Collections.unmodifiableMap(new LinkedHashMap<>(parameters));
}
public static RuntimeEventProcessingApiRequest tachographDriverEsper(UnifiedRuntimeProcessingApiRequest scope) {
public static RuntimeEventProcessingApiRequest driverWorkingTime(UnifiedRuntimeProcessingApiRequest scope) {
return new RuntimeEventProcessingApiRequest(
"tachograph-driver-esper-v1",
DriverWorkingTimeRuntimeProcessingPlan.PLAN_KEY,
scope,
new RuntimeEventPartitioningApiRequest(
at.procon.eventhub.processing.eventprocessing.partition.RuntimeEventPartitioningStrategy.DRIVER,
@ -39,7 +40,7 @@ public record RuntimeEventProcessingApiRequest(
scope != null ? scope.includeAllDrivers() : null,
scope != null ? scope.vehicleKeys() : null,
scope != null ? scope.includeAllVehicles() : null,
null,
scope != null ? scope.expandVehicleEvents() : null,
scope != null ? scope.vehicleExpansionPaddingMinutes() : null,
null
),

View File

@ -15,12 +15,19 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class DriverWorkingTimeDerivedProjectionsModule implements RuntimeProcessingModule {
private static final Logger LOG = LoggerFactory.getLogger(DriverWorkingTimeDerivedProjectionsModule.class);
private static final String DRIVER_PROCESSING_TOTAL_MS_METADATA_KEY = "driverProcessingTotalMs";
private static final String SLOWEST_DRIVER_KEY_METADATA_KEY = "slowestDriverKey";
private static final String SLOWEST_DRIVER_PROCESSING_MS_METADATA_KEY = "slowestDriverProcessingMs";
private final DriverWorkingTimeProcessingCore workingTimeProcessingCore;
private final RuntimeDriverWorkingTimeScopeProcessingService legacyScopeProcessingService;
@ -60,20 +67,31 @@ public class DriverWorkingTimeDerivedProjectionsModule implements RuntimeProcess
if (legacyScopeProcessingService != null) {
return executeLegacy(context);
}
long startedAtNanos = System.nanoTime();
UnifiedRuntimeEventBundle broadBundle = runtimeEventBundle(context);
UnifiedRuntimeProcessingApiRequest scopeRequest = scopeRequest(context);
Map<String, DriverWorkingTimePreparedInput> preparedInputs = preparedInputs(context);
LinkedHashMap<String, UnifiedRuntimeDerivedProjectionResultDto> driverResults = new LinkedHashMap<>();
List<String> warnings = new ArrayList<>();
long driverProcessingTotalMs = 0L;
String slowestDriverKey = null;
long slowestDriverProcessingMs = -1L;
for (Map.Entry<String, DriverWorkingTimePreparedInput> entry : preparedInputs.entrySet()) {
DriverWorkingTimePreparedInput preparedInput = entry.getValue();
long driverStartedAtNanos = System.nanoTime();
DriverWorkingTimeProcessingResultDto projection =
workingTimeProcessingCore.process(preparedInput.processingInput())
.withIncludedIntervals(
scopeRequest.includeActivityIntervalsOrDefault(),
scopeRequest.includeDrivingIntervalsOrDefault()
);
long driverProcessingMs = elapsedMillis(driverStartedAtNanos);
driverProcessingTotalMs += driverProcessingMs;
if (driverProcessingMs > slowestDriverProcessingMs) {
slowestDriverProcessingMs = driverProcessingMs;
slowestDriverKey = preparedInput.driverKey();
}
warnings.addAll(preparedInput.partition().warnings());
UnifiedRuntimeProcessingRequest driverRequest = broadBundle.request().withDriverKey(preparedInput.driverKey());
driverResults.put(preparedInput.driverKey(), new UnifiedRuntimeDerivedProjectionResultDto(
@ -110,6 +128,18 @@ public class DriverWorkingTimeDerivedProjectionsModule implements RuntimeProcess
metadata.put("selectedDriverCount", result.selectedDriverCount());
metadata.put("discoveredVehicleCount", result.discoveredVehicleCount());
metadata.put("driverResultCount", result.driverResults().size());
metadata.put(DRIVER_PROCESSING_TOTAL_MS_METADATA_KEY, driverProcessingTotalMs);
metadata.put(SLOWEST_DRIVER_KEY_METADATA_KEY, slowestDriverKey);
metadata.put(SLOWEST_DRIVER_PROCESSING_MS_METADATA_KEY, Math.max(0L, slowestDriverProcessingMs));
LOG.info(
"Driving-derived projections processed {} drivers in {} ms (driverProcessingTotalMs: {}, slowestDriverKey: {}, slowestDriverProcessingMs: {}, warnings: {})",
result.driverResults().size(),
elapsedMillis(startedAtNanos),
driverProcessingTotalMs,
slowestDriverKey,
Math.max(0L, slowestDriverProcessingMs),
result.warnings().size()
);
return new RuntimeProcessingModuleResult(
moduleKey(),
RuntimeProcessingModuleStatus.SUCCESS,
@ -187,4 +217,8 @@ public class DriverWorkingTimeDerivedProjectionsModule implements RuntimeProcess
}
return fallback;
}
private long elapsedMillis(long startedAtNanos) {
return Math.max(0L, (System.nanoTime() - startedAtNanos) / 1_000_000L);
}
}

View File

@ -39,6 +39,8 @@ public class RuntimeEventAssemblyModule implements RuntimeProcessingModule {
UnifiedRuntimeProcessingApiRequest scopeRequest = scopeRequest(context);
UnifiedRuntimeEventBundle bundle = eventAssemblyService.assembleDriverScopedEvents(scopeRequest.toRuntimeRequest());
Map<String, Object> metadata = new LinkedHashMap<>();
metadata.put("expandVehicleEvents", scopeRequest.expandVehicleEvents() == null || scopeRequest.expandVehicleEvents());
metadata.put("vehicleExpansionPaddingMinutes", scopeRequest.vehicleExpansionPaddingMinutes() == null ? 0 : scopeRequest.vehicleExpansionPaddingMinutes());
metadata.put("driverSeedEventCount", bundle.driverSeedEvents().size());
metadata.put("expandedVehicleEventCount", bundle.expandedVehicleEvents().size());
metadata.put("mergedEventCount", bundle.mergedEvents().size());

View File

@ -4,11 +4,17 @@ import at.procon.eventhub.processing.eventprocessing.plan.RuntimeProcessingExecu
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
@Service
public class RuntimeProcessingPipelineExecutor {
public static final String EXECUTION_DURATION_MS_METADATA_KEY = "executionDurationMs";
private static final Logger LOG = LoggerFactory.getLogger(RuntimeProcessingPipelineExecutor.class);
private final RuntimeProcessingModuleRegistry moduleRegistry;
public RuntimeProcessingPipelineExecutor(RuntimeProcessingModuleRegistry moduleRegistry) {
@ -29,8 +35,17 @@ public class RuntimeProcessingPipelineExecutor {
continue;
}
RuntimeProcessingModule module = moduleRegistry.require(moduleKey.trim());
RuntimeProcessingModuleResult result = module.execute(context);
long startedAtNanos = System.nanoTime();
RuntimeProcessingModuleResult result;
try {
result = withExecutionDuration(module.execute(context), startedAtNanos);
} catch (RuntimeException ex) {
long durationMs = elapsedMillis(startedAtNanos);
LOG.error("Runtime processing module {} failed after {} ms", module.moduleKey(), durationMs, ex);
throw ex;
}
results.put(module.moduleKey(), result);
logModuleCompletion(result);
context = new RuntimeProcessingModuleContext(
request,
context.events(),
@ -43,4 +58,45 @@ public class RuntimeProcessingPipelineExecutor {
}
return Map.copyOf(results);
}
private RuntimeProcessingModuleResult withExecutionDuration(
RuntimeProcessingModuleResult result,
long startedAtNanos
) {
long durationMs = elapsedMillis(startedAtNanos);
RuntimeProcessingModuleResult safeResult = result == null
? new RuntimeProcessingModuleResult(
"UNKNOWN",
RuntimeProcessingModuleStatus.FAILED,
null,
Map.of(),
List.of("Runtime processing module returned null result.")
)
: result;
LinkedHashMap<String, Object> metadata = new LinkedHashMap<>(safeResult.metadata());
metadata.put(EXECUTION_DURATION_MS_METADATA_KEY, durationMs);
return new RuntimeProcessingModuleResult(
safeResult.moduleKey(),
safeResult.status(),
safeResult.output(),
metadata,
safeResult.warnings()
);
}
private void logModuleCompletion(RuntimeProcessingModuleResult result) {
Object durationValue = result.metadata().get(EXECUTION_DURATION_MS_METADATA_KEY);
long durationMs = durationValue instanceof Number number ? number.longValue() : -1L;
LOG.info(
"Runtime processing module {} completed with status {} in {} ms (warnings: {})",
result.moduleKey(),
result.status(),
durationMs,
result.warnings().size()
);
}
private long elapsedMillis(long startedAtNanos) {
return Math.max(0L, (System.nanoTime() - startedAtNanos) / 1_000_000L);
}
}

View File

@ -85,8 +85,16 @@ public class VehicleEvidenceAttachmentModule implements RuntimeProcessingModule
directDriverEvents,
broadBundle.mergedEvents(),
driverVehicleUsageIntervals,
scopeRequest.expandVehicleEvents() == null || scopeRequest.expandVehicleEvents(),
scopeRequest.vehicleExpansionPaddingMinutes() == null ? 0 : scopeRequest.vehicleExpansionPaddingMinutes(),
booleanAttribute(
context,
at.procon.eventhub.processing.eventprocessing.plan.DriverWorkingTimeRuntimeProcessingPlan.ATTACH_VEHICLE_ONLY_EVENTS_ATTRIBUTE,
scopeRequest.expandVehicleEvents() == null || scopeRequest.expandVehicleEvents()
),
integerAttribute(
context,
at.procon.eventhub.processing.eventprocessing.plan.DriverWorkingTimeRuntimeProcessingPlan.VEHICLE_EVIDENCE_PADDING_MINUTES_ATTRIBUTE,
scopeRequest.vehicleExpansionPaddingMinutes() == null ? 0 : scopeRequest.vehicleExpansionPaddingMinutes()
),
includePartitionDebug
);
for (EventHubEventDto attachedEvent : attachmentResult.attachedVehicleEvidenceEvents()) {
@ -291,4 +299,23 @@ public class VehicleEvidenceAttachmentModule implements RuntimeProcessingModule
}
return fallback;
}
private int integerAttribute(RuntimeProcessingModuleContext context, String key, int fallback) {
Object value = context.attributes().get(key);
if (value instanceof Number number) {
return Math.max(0, number.intValue());
}
if (value == null) {
return Math.max(0, fallback);
}
String text = value.toString();
if (text.isBlank()) {
return Math.max(0, fallback);
}
try {
return Math.max(0, Integer.parseInt(text.trim()));
} catch (NumberFormatException ignored) {
return Math.max(0, fallback);
}
}
}

View File

@ -13,10 +13,15 @@ import com.espertech.esper.runtime.client.EPRuntimeProvider;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.ClassPathResource;
import org.springframework.stereotype.Service;
import org.springframework.util.StreamUtils;
@ -24,24 +29,63 @@ import org.springframework.util.StreamUtils;
@Service
public class RuntimeEplModuleExecutor {
public static final String DEFINITION_CACHE_HIT_METADATA_KEY = "eplDefinitionCacheHit";
public static final String DEFINITION_PREPARATION_MS_METADATA_KEY = "eplDefinitionPreparationMs";
public static final String RUNTIME_INIT_MS_METADATA_KEY = "eplRuntimeInitMs";
public static final String DEPLOY_MS_METADATA_KEY = "eplDeployMs";
public static final String LISTENER_REGISTRATION_MS_METADATA_KEY = "eplListenerRegistrationMs";
public static final String EVENT_SEND_MS_METADATA_KEY = "eplEventSendMs";
public static final String DESTROY_MS_METADATA_KEY = "eplDestroyMs";
private static final Logger LOG = LoggerFactory.getLogger(RuntimeEplModuleExecutor.class);
private static final AtomicLong RUNTIME_COUNTER = new AtomicLong();
private final ConcurrentMap<PreparedDefinitionCacheKey, PreparedDefinition> preparedDefinitions = new ConcurrentHashMap<>();
private final ConcurrentMap<String, String> eplResourceContents = new ConcurrentHashMap<>();
public RuntimeEplModuleExecutionResult execute(RuntimeEplModuleDefinition definition) {
EPRuntime runtime = null;
Map<String, List<Map<String, Object>>> outputs = null;
LinkedHashMap<String, Object> metadata = null;
PreparedDefinition preparedDefinition = null;
boolean definitionCacheHit = false;
long definitionPreparationMs = 0L;
long runtimeInitMs = 0L;
long deployMs = 0L;
long listenerRegistrationMs = 0L;
long eventSendMs = 0L;
long destroyMs = 0L;
long executionStartedAtNanos = System.nanoTime();
try {
Configuration configuration = new Configuration();
definition.eventTypes().forEach((eventTypeName, eventTypeDefinition) ->
configuration.getCommon().addEventType(eventTypeName, eventTypeDefinition));
PreparedDefinitionCacheKey cacheKey = PreparedDefinitionCacheKey.from(definition);
preparedDefinition = preparedDefinitions.get(cacheKey);
definitionCacheHit = preparedDefinition != null;
if (!definitionCacheHit) {
long definitionPreparationStartedAtNanos = System.nanoTime();
PreparedDefinition candidate = prepareDefinition(definition);
PreparedDefinition existing = preparedDefinitions.putIfAbsent(cacheKey, candidate);
preparedDefinition = existing == null ? candidate : existing;
definitionCacheHit = existing != null;
definitionPreparationMs = definitionCacheHit
? 0L
: elapsedMillis(definitionPreparationStartedAtNanos);
}
long runtimeInitStartedAtNanos = System.nanoTime();
Configuration configuration = createConfiguration(preparedDefinition.eventTypes());
String runtimeUri = "eventhub-runtime-epl-module-"
+ definition.moduleKey().replaceAll("[^A-Za-z0-9_-]", "-")
+ "-" + RUNTIME_COUNTER.incrementAndGet();
runtime = EPRuntimeProvider.getRuntime(runtimeUri, configuration);
runtimeInitMs = elapsedMillis(runtimeInitStartedAtNanos);
EPCompiled compiled = EPCompilerProvider.getCompiler().compile(renderEpl(definition.eplResources()), new CompilerArguments(configuration));
EPDeployment deployment = runtime.getDeploymentService().deploy(compiled);
long deployStartedAtNanos = System.nanoTime();
EPDeployment deployment = runtime.getDeploymentService().deploy(preparedDefinition.compiled());
deployMs = elapsedMillis(deployStartedAtNanos);
Map<String, List<Map<String, Object>>> outputs = new LinkedHashMap<>();
long listenerRegistrationStartedAtNanos = System.nanoTime();
outputs = new LinkedHashMap<>();
for (String statementName : definition.outputStatementNames()) {
outputs.put(statementName, new ArrayList<>());
var statement = runtime.getDeploymentService().getStatement(deployment.getDeploymentId(), statementName);
@ -49,18 +93,28 @@ public class RuntimeEplModuleExecutor {
throw new IllegalStateException("EPL module " + definition.moduleKey()
+ " did not deploy expected statement '" + statementName + "'.");
}
statement.addListener((newData, oldData, stmt, rt) -> collect(newData, outputs.get(statementName)));
List<Map<String, Object>> statementOutputs = outputs.get(statementName);
statement.addListener((newData, oldData, stmt, rt) -> collect(newData, statementOutputs));
}
listenerRegistrationMs = elapsedMillis(listenerRegistrationStartedAtNanos);
long eventSendStartedAtNanos = System.nanoTime();
for (RuntimeEplInputEventStream inputStream : definition.inputStreams()) {
for (Map<String, Object> event : inputStream.events()) {
runtime.getEventService().sendEventMap(event, inputStream.eventTypeName());
}
}
eventSendMs = elapsedMillis(eventSendStartedAtNanos);
Map<String, Object> metadata = new LinkedHashMap<>();
metadata = new LinkedHashMap<>();
metadata.put("engine", "EPL");
metadata.put("eplResources", definition.eplResources());
metadata.put(DEFINITION_CACHE_HIT_METADATA_KEY, definitionCacheHit);
metadata.put(DEFINITION_PREPARATION_MS_METADATA_KEY, definitionPreparationMs);
metadata.put(RUNTIME_INIT_MS_METADATA_KEY, runtimeInitMs);
metadata.put(DEPLOY_MS_METADATA_KEY, deployMs);
metadata.put(LISTENER_REGISTRATION_MS_METADATA_KEY, listenerRegistrationMs);
metadata.put(EVENT_SEND_MS_METADATA_KEY, eventSendMs);
Map<String, Integer> inputCounts = new LinkedHashMap<>();
for (RuntimeEplInputEventStream inputStream : definition.inputStreams()) {
inputCounts.merge(inputStream.eventTypeName(), inputStream.events().size(), Integer::sum);
@ -69,14 +123,48 @@ public class RuntimeEplModuleExecutor {
Map<String, Integer> outputCounts = new LinkedHashMap<>();
outputs.forEach((statement, events) -> outputCounts.put(statement, events.size()));
metadata.put("outputCounts", outputCounts);
return new RuntimeEplModuleExecutionResult(outputs, metadata);
} catch (EPCompileException | EPDeployException e) {
throw new IllegalStateException("Cannot compile/deploy runtime EPL module " + definition.moduleKey(), e);
} finally {
if (runtime != null) {
long destroyStartedAtNanos = System.nanoTime();
runtime.destroy();
destroyMs = elapsedMillis(destroyStartedAtNanos);
}
if (metadata != null) {
metadata.put(DESTROY_MS_METADATA_KEY, destroyMs);
LOG.info(
"Runtime EPL module {} completed in {} ms (definitionCacheHit: {}, definitionPreparationMs: {}, runtimeInitMs: {}, deployMs: {}, listenerRegistrationMs: {}, eventSendMs: {}, destroyMs: {})",
definition.moduleKey(),
elapsedMillis(executionStartedAtNanos),
definitionCacheHit,
definitionPreparationMs,
runtimeInitMs,
deployMs,
listenerRegistrationMs,
eventSendMs,
destroyMs
);
}
}
return new RuntimeEplModuleExecutionResult(
outputs == null ? Map.of() : outputs,
metadata == null ? Map.of() : metadata
);
}
private PreparedDefinition prepareDefinition(RuntimeEplModuleDefinition definition) throws EPCompileException {
Configuration configuration = createConfiguration(definition.eventTypes());
String epl = renderEpl(definition.eplResources());
EPCompiled compiled = EPCompilerProvider.getCompiler().compile(epl, new CompilerArguments(configuration));
return new PreparedDefinition(definition.eventTypes(), compiled);
}
private Configuration createConfiguration(Map<String, Map<String, Object>> eventTypes) {
Configuration configuration = new Configuration();
eventTypes.forEach((eventTypeName, eventTypeDefinition) ->
configuration.getCommon().addEventType(eventTypeName, eventTypeDefinition));
return configuration;
}
private String renderEpl(List<String> resources) {
@ -88,7 +176,7 @@ public class RuntimeEplModuleExecutor {
if (!builder.isEmpty()) {
builder.append("\n\n");
}
builder.append(loadResource(resource.trim()));
builder.append(eplResourceContents.computeIfAbsent(resource.trim(), RuntimeEplModuleExecutor::loadResource));
}
return builder.toString();
}
@ -117,4 +205,40 @@ public class RuntimeEplModuleExecutor {
target.add(values);
}
}
private long elapsedMillis(long startedAtNanos) {
return Math.max(0L, (System.nanoTime() - startedAtNanos) / 1_000_000L);
}
private record PreparedDefinition(
Map<String, Map<String, Object>> eventTypes,
EPCompiled compiled
) {
private PreparedDefinition {
eventTypes = eventTypes == null ? Map.of() : immutableNestedMap(eventTypes);
}
}
private record PreparedDefinitionCacheKey(
String moduleKey,
Map<String, Map<String, Object>> eventTypes,
List<String> eplResources
) {
private static PreparedDefinitionCacheKey from(RuntimeEplModuleDefinition definition) {
return new PreparedDefinitionCacheKey(
definition.moduleKey(),
immutableNestedMap(definition.eventTypes()),
List.copyOf(definition.eplResources())
);
}
}
private static Map<String, Map<String, Object>> immutableNestedMap(Map<String, Map<String, Object>> source) {
LinkedHashMap<String, Map<String, Object>> copy = new LinkedHashMap<>();
source.forEach((key, value) -> copy.put(
key,
value == null ? Map.of() : Collections.unmodifiableMap(new LinkedHashMap<>(value))
));
return Collections.unmodifiableMap(copy);
}
}

View File

@ -30,7 +30,8 @@ import org.springframework.stereotype.Component;
public class DriverWorkingTimeRuntimeProcessingPlan implements RuntimeProcessingPlan {
public static final String PLAN_KEY = "driver-working-time-v1";
public static final String LEGACY_PROFILE_ALIAS = "tachograph-driver-esper-v1";
public static final String ATTACH_VEHICLE_ONLY_EVENTS_ATTRIBUTE = "attachVehicleOnlyEvents";
public static final String VEHICLE_EVIDENCE_PADDING_MINUTES_ATTRIBUTE = "vehicleEvidencePaddingMinutes";
private final RuntimeProcessingPipelineExecutor pipelineExecutor;
private final boolean includeRuntimeEventAssemblyModule;
@ -76,7 +77,7 @@ public class DriverWorkingTimeRuntimeProcessingPlan implements RuntimeProcessing
@Override
public Set<String> aliases() {
return Set.of(LEGACY_PROFILE_ALIAS);
return Set.of();
}
@Override
@ -181,6 +182,16 @@ public class DriverWorkingTimeRuntimeProcessingPlan implements RuntimeProcessing
"includePartitionDebug",
request.partitioning() != null && request.partitioning().includeDebugOrDefault()
);
boolean attachVehicleOnlyEvents = resolveAttachVehicleOnlyEvents(
request.sourceSelection(),
request.partitioning(),
request.parameters()
);
int vehicleEvidencePaddingMinutes = resolveVehicleEvidencePaddingMinutes(
request.sourceSelection(),
request.partitioning(),
request.parameters()
);
UnifiedRuntimeProcessingApiRequest scopeRequest = applyExecutionRequest(
request.sourceSelection(),
request.partitioning(),
@ -190,6 +201,8 @@ public class DriverWorkingTimeRuntimeProcessingPlan implements RuntimeProcessing
Map<String, Object> attributes = new LinkedHashMap<>();
attributes.put("runtimeScopeApiRequest", scopeRequest);
attributes.put("includePartitionDebug", includePartitionDebug);
attributes.put(ATTACH_VEHICLE_ONLY_EVENTS_ATTRIBUTE, attachVehicleOnlyEvents);
attributes.put(VEHICLE_EVIDENCE_PADDING_MINUTES_ATTRIBUTE, vehicleEvidencePaddingMinutes);
RuntimeProcessingModuleContext initialContext = new RuntimeProcessingModuleContext(
request,
List.of(),
@ -374,12 +387,6 @@ public class DriverWorkingTimeRuntimeProcessingPlan implements RuntimeProcessing
sourceSelection.includeActivityIntervalsOrDefault());
boolean includeDrivingIntervals = booleanParameter(parameters, "includeDrivingIntervals",
sourceSelection.includeDrivingIntervalsOrDefault());
boolean attachVehicleOnlyEvents = booleanParameter(parameters, "attachVehicleOnlyEvents",
partitioning == null ? sourceSelection.expandVehicleEvents() == null || sourceSelection.expandVehicleEvents() : partitioning.attachVehicleEvidenceOrDefault());
Integer vehicleEvidencePaddingMinutes = nonNegativeIntegerParameter(parameters, "vehicleEvidencePaddingMinutes",
partitioning == null
? sourceSelection.vehicleExpansionPaddingMinutes()
: partitioning.vehicleEvidencePaddingMinutesOrDefault(sourceSelection.vehicleExpansionPaddingMinutes() == null ? 0 : sourceSelection.vehicleExpansionPaddingMinutes()));
return new UnifiedRuntimeProcessingApiRequest(
sourceSelection.sessionId(),
@ -399,8 +406,8 @@ public class DriverWorkingTimeRuntimeProcessingPlan implements RuntimeProcessing
sourceSelection.driverCardNumber(),
sourceSelection.occurredFrom(),
sourceSelection.occurredTo(),
attachVehicleOnlyEvents,
vehicleEvidencePaddingMinutes,
sourceSelection.expandVehicleEvents(),
sourceSelection.vehicleExpansionPaddingMinutes(),
sourceSelection.includeIntersectingIntervals(),
significantDrivingMinutes,
minimumRestPeriodMinutes,
@ -410,6 +417,37 @@ public class DriverWorkingTimeRuntimeProcessingPlan implements RuntimeProcessing
);
}
private boolean resolveAttachVehicleOnlyEvents(
UnifiedRuntimeProcessingApiRequest sourceSelection,
RuntimeEventPartitioningApiRequest partitioning,
Map<String, Object> parameters
) {
return booleanParameter(
parameters,
"attachVehicleOnlyEvents",
partitioning == null
? sourceSelection.expandVehicleEvents() == null || sourceSelection.expandVehicleEvents()
: partitioning.attachVehicleEvidenceOrDefault()
);
}
private int resolveVehicleEvidencePaddingMinutes(
UnifiedRuntimeProcessingApiRequest sourceSelection,
RuntimeEventPartitioningApiRequest partitioning,
Map<String, Object> parameters
) {
Integer resolved = nonNegativeIntegerParameter(
parameters,
"vehicleEvidencePaddingMinutes",
partitioning == null
? sourceSelection.vehicleExpansionPaddingMinutes()
: partitioning.vehicleEvidencePaddingMinutesOrDefault(
sourceSelection.vehicleExpansionPaddingMinutes() == null ? 0 : sourceSelection.vehicleExpansionPaddingMinutes()
)
);
return resolved == null ? 0 : resolved;
}
private List<String> requestedOrDefaultModules(List<String> requestedModules) {
if (requestedModules != null && !requestedModules.isEmpty()) {
LinkedHashMap<String, String> requested = new LinkedHashMap<>();

View File

@ -48,7 +48,7 @@ public record RuntimeProcessingExecutionApiRequest(
sourceSelection != null ? sourceSelection.includeAllDrivers() : null,
sourceSelection != null ? sourceSelection.vehicleKeys() : null,
sourceSelection != null ? sourceSelection.includeAllVehicles() : null,
null,
sourceSelection != null ? sourceSelection.expandVehicleEvents() : null,
sourceSelection != null ? sourceSelection.vehicleExpansionPaddingMinutes() : null,
null
),

View File

@ -1,11 +1,17 @@
package at.procon.eventhub.processing.eventprocessing.plan;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import at.procon.eventhub.processing.eventprocessing.module.RuntimeProcessingModuleResult;
import at.procon.eventhub.processing.eventprocessing.module.RuntimeProcessingPipelineExecutor;
import org.springframework.stereotype.Service;
@Service
public class RuntimeProcessingExecutionService {
private static final Logger LOG = LoggerFactory.getLogger(RuntimeProcessingExecutionService.class);
private final RuntimeProcessingPlanRegistry planRegistry;
public RuntimeProcessingExecutionService(RuntimeProcessingPlanRegistry planRegistry) {
@ -13,12 +19,42 @@ public class RuntimeProcessingExecutionService {
}
public RuntimeProcessingExecutionResultDto execute(RuntimeProcessingExecutionApiRequest request) {
long startedAtNanos = System.nanoTime();
RuntimeProcessingPlan plan = planRegistry.require(request.processingPlanKey());
try {
plan.validatePartitioning(request.partitioning());
return plan.execute(request);
RuntimeProcessingExecutionResultDto result = plan.execute(request);
RuntimeProcessingModuleResult slowestModule = result.moduleResults().values().stream()
.filter(moduleResult -> moduleResult.metadata().get(RuntimeProcessingPipelineExecutor.EXECUTION_DURATION_MS_METADATA_KEY) instanceof Number)
.max(java.util.Comparator.comparingLong(moduleResult ->
((Number) moduleResult.metadata().get(RuntimeProcessingPipelineExecutor.EXECUTION_DURATION_MS_METADATA_KEY)).longValue()))
.orElse(null);
LOG.info(
"Runtime processing plan {} completed in {} ms (modules: {}, slowestModule: {}, slowestModuleMs: {}, warnings: {})",
request.processingPlanKey(),
elapsedMillis(startedAtNanos),
result.executedModules().size(),
slowestModule == null ? null : slowestModule.moduleKey(),
slowestModule == null ? null : slowestModule.metadata().get(RuntimeProcessingPipelineExecutor.EXECUTION_DURATION_MS_METADATA_KEY),
result.warnings().size()
);
return result;
} catch (RuntimeException ex) {
LOG.error(
"Runtime processing plan {} failed after {} ms",
request == null ? null : request.processingPlanKey(),
elapsedMillis(startedAtNanos),
ex
);
throw ex;
}
}
public List<RuntimeProcessingPlanDescriptorDto> listPlans() {
return planRegistry.planDescriptors();
}
private long elapsedMillis(long startedAtNanos) {
return Math.max(0L, (System.nanoTime() - startedAtNanos) / 1_000_000L);
}
}

View File

@ -1,38 +1,28 @@
package at.procon.eventhub.tachographfilesession.processing.profile;
package at.procon.eventhub.processing.eventprocessing.profile;
import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingApiRequest;
import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingResultDto;
import at.procon.eventhub.processing.eventprocessing.partition.RuntimeEventPartitioningStrategy;
import at.procon.eventhub.processing.eventprocessing.plan.DriverWorkingTimeRuntimeProcessingPlan;
import at.procon.eventhub.processing.eventprocessing.plan.RuntimeProcessingExecutionApiRequest;
import at.procon.eventhub.processing.eventprocessing.plan.RuntimeProcessingExecutionResultDto;
import at.procon.eventhub.processing.eventprocessing.plan.DriverWorkingTimeRuntimeProcessingPlan;
import at.procon.eventhub.processing.eventprocessing.profile.RuntimeEventProcessingProfile;
import at.procon.eventhub.processing.service.RuntimeDriverWorkingTimeScopeProcessingService;
import java.util.List;
import java.util.Set;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class TachographDriverEsperRuntimeEventProcessingProfile implements RuntimeEventProcessingProfile {
public class DriverWorkingTimeRuntimeEventProcessingProfile implements RuntimeEventProcessingProfile {
/**
* Legacy compatibility key. New clients should use processingPlanKey=driver-working-time-v1
* via /api/eventhub/runtime-processing/executions.
*/
public static final String PROFILE_KEY = DriverWorkingTimeRuntimeProcessingPlan.LEGACY_PROFILE_ALIAS;
public static final String PROFILE_KEY = DriverWorkingTimeRuntimeProcessingPlan.PLAN_KEY;
private final DriverWorkingTimeRuntimeProcessingPlan plan;
@Autowired
public TachographDriverEsperRuntimeEventProcessingProfile(DriverWorkingTimeRuntimeProcessingPlan plan) {
public DriverWorkingTimeRuntimeEventProcessingProfile(DriverWorkingTimeRuntimeProcessingPlan plan) {
this.plan = plan;
}
public TachographDriverEsperRuntimeEventProcessingProfile(RuntimeDriverWorkingTimeScopeProcessingService scopeProcessingService) {
this(new DriverWorkingTimeRuntimeProcessingPlan(scopeProcessingService));
}
@Override
public String profileKey() {
return PROFILE_KEY;
@ -45,14 +35,13 @@ public class TachographDriverEsperRuntimeEventProcessingProfile implements Runti
@Override
public String displayName() {
return "Tachograph Driver Esper Processing";
return plan.displayName();
}
@Override
public String description() {
return "Compatibility adapter for legacy profileKey=" + PROFILE_KEY
+ ". New clients should use processingPlanKey=" + plan.processingPlanKey() + ". "
+ plan.description();
return "Compatibility runtime-event-processing adapter for processingPlanKey="
+ plan.processingPlanKey() + ". " + plan.description();
}
@Override

View File

@ -14,11 +14,15 @@ import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Objects;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
@Service
public class UnifiedRuntimeEventAssemblyService {
private static final Logger LOG = LoggerFactory.getLogger(UnifiedRuntimeEventAssemblyService.class);
private final List<RuntimeDriverEventLoader> driverEventLoaders;
private final List<RuntimeVehicleEventLoader> vehicleEventLoaders;
@ -34,10 +38,13 @@ public class UnifiedRuntimeEventAssemblyService {
List<UnifiedRuntimeSourceInput> sourceInputs = request.normalizedSourceInputs();
List<EventHubEventDto> driverSeedEvents = loadDriverSeedEvents(request);
List<UnifiedDiscoveredVehicleRef> discoveredVehicles = discoverVehicles(driverSeedEvents);
List<EventHubEventDto> expandedVehicleEvents = request.expandVehicleEvents()
boolean expandVehicleEvents = request.expandVehicleEvents();
List<EventHubEventDto> expandedVehicleEvents = expandVehicleEvents
? loadExpandedVehicleEvents(request, discoveredVehicles)
: List.of();
List<EventHubEventDto> mergedEvents = deduplicateAndSort(driverSeedEvents, expandedVehicleEvents);
List<EventHubEventDto> mergedEvents = expandVehicleEvents
? deduplicateAndSort(driverSeedEvents, expandedVehicleEvents)
: driverSeedEvents;
List<String> notes = new ArrayList<>();
boolean includesEventHub = sourceInputs.stream()
@ -63,12 +70,21 @@ public class UnifiedRuntimeEventAssemblyService {
notes.add("Tachograph file-session events were loaded from session " + sourceInput.sessionId() + ".");
}
}
if (request.expandVehicleEvents()) {
if (expandVehicleEvents) {
notes.add("Vehicle expansion loaded additional events for vehicles discovered in the driver seed set.");
notes.add("Vehicle expansion padding minutes: " + request.vehicleExpansionPaddingMinutes() + ".");
} else {
notes.add("Vehicle expansion was disabled for this runtime request.");
}
LOG.info(
"Runtime event assembly completed (expandVehicleEvents: {}, sourceInputs: {}, driverSeedEvents: {}, discoveredVehicles: {}, expandedVehicleEvents: {}, mergedEvents: {})",
expandVehicleEvents,
sourceInputs.size(),
driverSeedEvents.size(),
discoveredVehicles.size(),
expandedVehicleEvents.size(),
mergedEvents.size()
);
return new UnifiedRuntimeEventBundle(
request,
driverSeedEvents,

View File

@ -1,17 +0,0 @@
package at.procon.eventhub.tachographfilesession.processing.plan;
import at.procon.eventhub.processing.eventprocessing.plan.DriverWorkingTimeRuntimeProcessingPlan;
import at.procon.eventhub.processing.service.RuntimeDriverWorkingTimeScopeProcessingService;
/**
* @deprecated Compatibility adapter. Use {@link DriverWorkingTimeRuntimeProcessingPlan}.
*/
@Deprecated(forRemoval = false)
public class TachographDriverWorkingTimeRuntimeProcessingPlan extends DriverWorkingTimeRuntimeProcessingPlan {
public TachographDriverWorkingTimeRuntimeProcessingPlan(
RuntimeDriverWorkingTimeScopeProcessingService driverWorkingTimeScopeProcessingService
) {
super(driverWorkingTimeScopeProcessingService);
}
}

View File

@ -9,7 +9,7 @@ import at.procon.eventhub.tachographfilesession.model.DriverExtractionSession;
import at.procon.eventhub.tachographfilesession.model.TachographFileSession;
import at.procon.eventhub.tachographfilesession.model.TachographTimelineEventBundle;
import at.procon.eventhub.tachographfilesession.service.DriverNotFoundInSessionException;
import at.procon.eventhub.tachographfilesession.service.DriverTimelineEventBuilder;
import at.procon.eventhub.tachographfilesession.service.RawSourceDriverTimelineEventBuilder;
import at.procon.eventhub.tachographfilesession.service.TachographFileSessionNotFoundException;
import at.procon.eventhub.tachographfilesession.service.TachographFileSessionRepository;
import java.util.List;
@ -19,14 +19,14 @@ import org.springframework.stereotype.Component;
public class TachographFileSessionUnifiedDriverEventSource implements UnifiedDriverEventSource {
private final TachographFileSessionRepository repository;
private final DriverTimelineEventBuilder eventBuilder;
private final RawSourceDriverTimelineEventBuilder rawSourceEventBuilder;
public TachographFileSessionUnifiedDriverEventSource(
TachographFileSessionRepository repository,
DriverTimelineEventBuilder eventBuilder
RawSourceDriverTimelineEventBuilder rawSourceEventBuilder
) {
this.repository = repository;
this.eventBuilder = eventBuilder;
this.rawSourceEventBuilder = rawSourceEventBuilder;
}
@Override
@ -56,7 +56,7 @@ public class TachographFileSessionUnifiedDriverEventSource implements UnifiedDri
DriverExtractionSession driver,
UnifiedDriverEventsRequest request
) {
TachographTimelineEventBundle bundle = eventBuilder.buildEventBundle(session, driver);
TachographTimelineEventBundle bundle = rawSourceEventBuilder.buildRawEventBundle(session, driver);
return TachographTimelineEventBundle.fromRuntimeBundle(
RuntimeIntervalEventWindowSelector.filterBundle(
bundle == null ? null : bundle.toRuntimeBundle(),

View File

@ -10,7 +10,7 @@ import at.procon.eventhub.reference.TachographNationRegistry;
import at.procon.eventhub.tachographfilesession.model.DriverExtractionSession;
import at.procon.eventhub.tachographfilesession.model.TachographFileSession;
import at.procon.eventhub.tachographfilesession.model.TachographTimelineEventBundle;
import at.procon.eventhub.tachographfilesession.service.DriverTimelineEventBuilder;
import at.procon.eventhub.tachographfilesession.service.RawSourceDriverTimelineEventBuilder;
import at.procon.eventhub.tachographfilesession.service.TachographFileSessionNotFoundException;
import at.procon.eventhub.tachographfilesession.service.TachographFileSessionRepository;
import java.util.List;
@ -20,14 +20,14 @@ import org.springframework.stereotype.Component;
public class TachographFileSessionUnifiedVehicleEventSource implements UnifiedVehicleEventSource {
private final TachographFileSessionRepository repository;
private final DriverTimelineEventBuilder eventBuilder;
private final RawSourceDriverTimelineEventBuilder rawSourceEventBuilder;
public TachographFileSessionUnifiedVehicleEventSource(
TachographFileSessionRepository repository,
DriverTimelineEventBuilder eventBuilder
RawSourceDriverTimelineEventBuilder rawSourceEventBuilder
) {
this.repository = repository;
this.eventBuilder = eventBuilder;
this.rawSourceEventBuilder = rawSourceEventBuilder;
}
@Override
@ -52,7 +52,7 @@ public class TachographFileSessionUnifiedVehicleEventSource implements UnifiedVe
DriverExtractionSession driver,
UnifiedVehicleEventsRequest request
) {
TachographTimelineEventBundle bundle = eventBuilder.buildEventBundle(session, driver);
TachographTimelineEventBundle bundle = rawSourceEventBuilder.buildRawEventBundle(session, driver);
return TachographTimelineEventBundle.fromRuntimeBundle(
RuntimeIntervalEventWindowSelector.filterBundle(
bundle == null ? null : bundle.toRuntimeBundle(),

View File

@ -1,49 +0,0 @@
package at.procon.eventhub.tachographfilesession.processing.service;
import at.procon.eventhub.processing.dto.UnifiedRuntimeDriverWorkingTimeScopeResultDto;
import at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest;
import at.procon.eventhub.processing.dto.UnifiedRuntimeTachographEsperScopeResultDto;
import at.procon.eventhub.processing.service.RuntimeDriverWorkingTimeScopeProcessingService;
/**
* @deprecated Compatibility adapter. Use {@link RuntimeDriverWorkingTimeScopeProcessingService};
* tachograph files/databases are only one possible source for the common driver working-time runtime plan.
*/
@Deprecated(forRemoval = false)
public class UnifiedRuntimeTachographEsperScopeProcessingService {
private final RuntimeDriverWorkingTimeScopeProcessingService delegate;
public UnifiedRuntimeTachographEsperScopeProcessingService(
RuntimeDriverWorkingTimeScopeProcessingService delegate
) {
this.delegate = delegate;
}
public UnifiedRuntimeTachographEsperScopeResultDto processScope(UnifiedRuntimeProcessingApiRequest apiRequest) {
return toLegacy(delegate.processScope(apiRequest));
}
public UnifiedRuntimeTachographEsperScopeResultDto processScope(
UnifiedRuntimeProcessingApiRequest apiRequest,
boolean includePartitionDebug
) {
return toLegacy(delegate.processScope(apiRequest, includePartitionDebug));
}
private UnifiedRuntimeTachographEsperScopeResultDto toLegacy(
UnifiedRuntimeDriverWorkingTimeScopeResultDto result
) {
return new UnifiedRuntimeTachographEsperScopeResultDto(
result.request(),
result.inputEventCount(),
result.selectedDriverCount(),
result.discoveredVehicleCount(),
result.discoveredVehicles(),
result.driverResults(),
result.partitionDebugByDriver(),
result.notes(),
result.warnings()
);
}
}

View File

@ -9,7 +9,7 @@ import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingA
import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingPartitionResultDto;
import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingResultDto;
import at.procon.eventhub.processing.eventprocessing.partition.RuntimeEventPartitioningStrategy;
import at.procon.eventhub.tachographfilesession.processing.profile.TachographDriverEsperRuntimeEventProcessingProfile;
import at.procon.eventhub.processing.eventprocessing.profile.DriverWorkingTimeRuntimeEventProcessingProfile;
import at.procon.eventhub.processing.model.UnifiedEventSourceFamily;
import at.procon.eventhub.processing.model.UnifiedRuntimeEventBackend;
import at.procon.eventhub.tachographfilesession.dto.TachographEsperDriverProcessingResultDto;
@ -85,13 +85,13 @@ public class RuntimeTachographParityValidationService {
: "DIFFERENT";
List<String> notes = new ArrayList<>(runtimeResult.notes());
notes.add("Validation compares the legacy tachograph file-session esper-events path with the generic runtime event-processing profile '"
+ TachographDriverEsperRuntimeEventProcessingProfile.PROFILE_KEY + "'.");
+ DriverWorkingTimeRuntimeEventProcessingProfile.PROFILE_KEY + "'.");
if (sessionIds.size() > 1) {
notes.add("For multiple sessions, file-session reference counts are summed per driver. Runtime processing may intentionally merge or deduplicate intervals across session boundaries, so category differences need domain review.");
}
return new RuntimeTachographParityValidationResultDto(
TachographDriverEsperRuntimeEventProcessingProfile.PROFILE_KEY,
DriverWorkingTimeRuntimeEventProcessingProfile.PROFILE_KEY,
status,
sessionIds,
driverResults.size(),
@ -207,7 +207,7 @@ public class RuntimeTachographParityValidationService {
parameters.put("vehicleEvidencePaddingMinutes", request.vehicleExpansionPaddingMinutesOrDefault());
parameters.put("includePartitionDebug", request.includeDebugOrDefault());
return new RuntimeEventProcessingApiRequest(
TachographDriverEsperRuntimeEventProcessingProfile.PROFILE_KEY,
DriverWorkingTimeRuntimeEventProcessingProfile.PROFILE_KEY,
scope,
partitioning,
parameters

View File

@ -288,7 +288,7 @@ create schema DailyWeeklyRestCandidateCoverageEmittedKey(
endedAtEpochSecond long
);
create context PerDriver partition by driverKey from TachographVehicleUsageIntervalInputEvent;
@public create context PerDriver partition by driverKey from TachographVehicleUsageIntervalInputEvent;
create schema VuCardAbsentInterval(
sessionId java.util.UUID,
@ -438,27 +438,27 @@ create schema PotentialInVehicleTripInterval(
lastNextDrivingSourceIntervalId string
);
create window PreviousRestCandidateCoverageInterval#unique(driverKey) as DailyWeeklyRestCandidateCoverageInterval;
@public create window PreviousRestCandidateCoverageInterval#unique(driverKey) as DailyWeeklyRestCandidateCoverageInterval;
create window OpenPotentialInVehicleTripState#unique(driverKey) as PotentialInVehicleTripState;
@public create window OpenPotentialInVehicleTripState#unique(driverKey) as PotentialInVehicleTripState;
create window SupportGeoEvidenceWindow#keepall as SupportGeoEvidence;
@public create window SupportGeoEvidenceWindow#keepall as SupportGeoEvidence;
create window DailyWeeklyRestCandidateBeginGeoEvidenceCandidateWindow#keepall as DailyWeeklyRestCandidateBeginGeoEvidenceCandidate;
create window DailyWeeklyRestCandidateBeginGeoEvidenceBestScoreWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateBeginGeoEvidenceBestScore;
create window DailyWeeklyRestCandidateBeginGeoEvidenceResolvedWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateBeginGeoEvidenceResolved;
@public create window DailyWeeklyRestCandidateBeginGeoEvidenceCandidateWindow#keepall as DailyWeeklyRestCandidateBeginGeoEvidenceCandidate;
@public create window DailyWeeklyRestCandidateBeginGeoEvidenceBestScoreWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateBeginGeoEvidenceBestScore;
@public create window DailyWeeklyRestCandidateBeginGeoEvidenceResolvedWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateBeginGeoEvidenceResolved;
create window DailyWeeklyRestCandidateEndGeoEvidenceCandidateWindow#keepall as DailyWeeklyRestCandidateEndGeoEvidenceCandidate;
create window DailyWeeklyRestCandidateEndGeoEvidenceBestScoreWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateEndGeoEvidenceBestScore;
create window DailyWeeklyRestCandidateEndGeoEvidenceResolvedWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateEndGeoEvidenceResolved;
create window DailyWeeklyRestCandidateBeginBoundaryOdometerCandidateWindow#keepall as DailyWeeklyRestCandidateBeginBoundaryOdometerCandidate;
create window DailyWeeklyRestCandidateBeginBoundaryOdometerBestScoreWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateBeginBoundaryOdometerBestScore;
create window DailyWeeklyRestCandidateBeginBoundaryOdometerResolvedWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateBeginBoundaryOdometerResolved;
create window DailyWeeklyRestCandidateEndBoundaryOdometerCandidateWindow#keepall as DailyWeeklyRestCandidateEndBoundaryOdometerCandidate;
create window DailyWeeklyRestCandidateEndBoundaryOdometerBestScoreWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateEndBoundaryOdometerBestScore;
create window DailyWeeklyRestCandidateEndBoundaryOdometerResolvedWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateEndBoundaryOdometerResolved;
create window DailyWeeklyRestCandidateCoverageCardResolvedIntervalWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateCoverageCardResolvedInterval;
create window DailyWeeklyRestCandidateCoverageEmittedKeyWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateCoverageEmittedKey;
@public create window DailyWeeklyRestCandidateEndGeoEvidenceCandidateWindow#keepall as DailyWeeklyRestCandidateEndGeoEvidenceCandidate;
@public create window DailyWeeklyRestCandidateEndGeoEvidenceBestScoreWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateEndGeoEvidenceBestScore;
@public create window DailyWeeklyRestCandidateEndGeoEvidenceResolvedWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateEndGeoEvidenceResolved;
@public create window DailyWeeklyRestCandidateBeginBoundaryOdometerCandidateWindow#keepall as DailyWeeklyRestCandidateBeginBoundaryOdometerCandidate;
@public create window DailyWeeklyRestCandidateBeginBoundaryOdometerBestScoreWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateBeginBoundaryOdometerBestScore;
@public create window DailyWeeklyRestCandidateBeginBoundaryOdometerResolvedWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateBeginBoundaryOdometerResolved;
@public create window DailyWeeklyRestCandidateEndBoundaryOdometerCandidateWindow#keepall as DailyWeeklyRestCandidateEndBoundaryOdometerCandidate;
@public create window DailyWeeklyRestCandidateEndBoundaryOdometerBestScoreWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateEndBoundaryOdometerBestScore;
@public create window DailyWeeklyRestCandidateEndBoundaryOdometerResolvedWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateEndBoundaryOdometerResolved;
@public create window DailyWeeklyRestCandidateCoverageCardResolvedIntervalWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateCoverageCardResolvedInterval;
@public create window DailyWeeklyRestCandidateCoverageEmittedKeyWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateCoverageEmittedKey;
insert into SupportGeoEvidenceWindow
select
@ -488,7 +488,7 @@ select
vehicleKey
from TachographActivityIntervalInputEvent(activityType = 'DRIVE', durationSeconds > ${SIGNIFICANT_DRIVING_THRESHOLD_SECONDS});
create window PreviousSignificantDrivingInterval#unique(driverKey) as SignificantDrivingInterval;
@public create window PreviousSignificantDrivingInterval#unique(driverKey) as SignificantDrivingInterval;
on SignificantDrivingInterval as next
insert into DrivingInterruptionInterval
@ -1483,7 +1483,7 @@ select
endedAtEpochSecond
from DailyWeeklyRestCandidateCoverageInterval;
context PerDriver
@public context PerDriver
create window PreviousVehicleUsageInterval#lastevent as TachographVehicleUsageIntervalInputEvent;
@Priority(30)

View File

@ -35,10 +35,6 @@ import at.procon.eventhub.processing.model.RuntimeDriverTimeline;
import at.procon.eventhub.processing.model.RuntimeVehicleUsageInterval;
import at.procon.eventhub.processing.model.UnifiedRuntimeProcessingRequest;
import at.procon.eventhub.processing.service.UnifiedRuntimeDerivedProjectionService;
import at.procon.eventhub.tachographfilesession.processing.validation.RuntimeTachographDriverParityResultDto;
import at.procon.eventhub.tachographfilesession.processing.validation.RuntimeTachographParityCategoryComparisonDto;
import at.procon.eventhub.tachographfilesession.processing.validation.RuntimeTachographParityValidationResultDto;
import at.procon.eventhub.tachographfilesession.processing.validation.RuntimeTachographParityValidationService;
import at.procon.eventhub.processing.service.UnifiedRuntimeDriverTimelineService;
import at.procon.eventhub.processing.service.UnifiedRuntimeEventAssemblyService;
import at.procon.eventhub.tachographfilesession.dto.TachographEsperDriverProcessingResultDto;
@ -328,8 +324,6 @@ class UnifiedRuntimeProcessingControllerTest {
derivedProjectionService,
null,
null,
null,
null,
executionService
))
.setMessageConverters(new MappingJackson2HttpMessageConverter(objectMapper))
@ -339,7 +333,7 @@ class UnifiedRuntimeProcessingControllerTest {
when(executionService.listPlans())
.thenReturn(List.of(new RuntimeProcessingPlanDescriptorDto(
"driver-working-time-v1",
Set.of("tachograph-driver-esper-v1"),
Set.of(),
"Driver working-time processing",
"Runs common runtime event processing modules.",
RuntimeEventPartitioningStrategy.DRIVER,
@ -351,8 +345,7 @@ class UnifiedRuntimeProcessingControllerTest {
mockMvc.perform(get("/api/eventhub/runtime-processing/executions/plans"))
.andExpect(status().isOk())
.andExpect(jsonPath("$[0].processingPlanKey").value("driver-working-time-v1"))
.andExpect(jsonPath("$[0].aliases[0]").value("tachograph-driver-esper-v1"));
.andExpect(jsonPath("$[0].processingPlanKey").value("driver-working-time-v1"));
}
@Test
@ -367,8 +360,6 @@ class UnifiedRuntimeProcessingControllerTest {
derivedProjectionService,
null,
null,
null,
null,
executionService
))
.setMessageConverters(new MappingJackson2HttpMessageConverter(objectMapper))
@ -434,8 +425,6 @@ class UnifiedRuntimeProcessingControllerTest {
derivedProjectionService,
null,
null,
null,
null,
executionService
))
.setMessageConverters(new MappingJackson2HttpMessageConverter(objectMapper))
@ -517,7 +506,6 @@ class UnifiedRuntimeProcessingControllerTest {
eventAssemblyService,
timelineService,
derivedProjectionService,
null,
runtimeEventProcessingService
))
.setMessageConverters(new MappingJackson2HttpMessageConverter(objectMapper))
@ -526,9 +514,9 @@ class UnifiedRuntimeProcessingControllerTest {
when(runtimeEventProcessingService.listProfiles())
.thenReturn(List.of(new RuntimeEventProcessingProfileDescriptorDto(
"tachograph-driver-esper-v1",
"Tachograph Driver Esper Processing",
"Runs tachograph driver Esper processing over runtime event scopes.",
"driver-working-time-v1",
"Driver working-time processing",
"Runs common runtime event processing modules.",
RuntimeEventPartitioningStrategy.DRIVER,
List.of(RuntimeEventPartitioningStrategy.DRIVER),
Set.of(),
@ -537,8 +525,8 @@ class UnifiedRuntimeProcessingControllerTest {
mockMvc.perform(get("/api/eventhub/runtime-processing/event-processing/profiles"))
.andExpect(status().isOk())
.andExpect(jsonPath("$[0].profileKey").value("tachograph-driver-esper-v1"))
.andExpect(jsonPath("$[0].displayName").value("Tachograph Driver Esper Processing"))
.andExpect(jsonPath("$[0].profileKey").value("driver-working-time-v1"))
.andExpect(jsonPath("$[0].displayName").value("Driver working-time processing"))
.andExpect(jsonPath("$[0].defaultPartitioningStrategy").value("DRIVER"))
.andExpect(jsonPath("$[0].supportedPartitioningStrategies[0]").value("DRIVER"))
.andExpect(jsonPath("$[0].optionalParameters[0]").exists());
@ -554,7 +542,6 @@ class UnifiedRuntimeProcessingControllerTest {
eventAssemblyService,
timelineService,
derivedProjectionService,
null,
runtimeEventProcessingService
))
.setMessageConverters(new MappingJackson2HttpMessageConverter(objectMapper))
@ -572,7 +559,7 @@ class UnifiedRuntimeProcessingControllerTest {
);
when(runtimeEventProcessingService.process(any()))
.thenReturn(new RuntimeEventProcessingResultDto(
"tachograph-driver-esper-v1",
"driver-working-time-v1",
RuntimeEventPartitioningStrategy.DRIVER,
request,
3,
@ -588,7 +575,7 @@ class UnifiedRuntimeProcessingControllerTest {
.contentType("application/json")
.content("""
{
"profileKey": "tachograph-driver-esper-v1",
"profileKey": "driver-working-time-v1",
"scope": {
"sessionId": "%s",
"sourceFamilies": ["TACHOGRAPH_FILE_SESSION"],
@ -607,155 +594,12 @@ class UnifiedRuntimeProcessingControllerTest {
}
""".formatted(sessionId)))
.andExpect(status().isOk())
.andExpect(jsonPath("$.profileKey").value("tachograph-driver-esper-v1"))
.andExpect(jsonPath("$.profileKey").value("driver-working-time-v1"))
.andExpect(jsonPath("$.partitioningStrategy").value("DRIVER"))
.andExpect(jsonPath("$.inputEventCount").value(3))
.andExpect(jsonPath("$.discoveredVehicles[0].vin").value("VIN-1"));
}
@Test
void compatibilityTachographEndpointDelegatesThroughGenericProfileRuntimeApi() throws Exception {
UnifiedRuntimeEventAssemblyService eventAssemblyService = org.mockito.Mockito.mock(UnifiedRuntimeEventAssemblyService.class);
UnifiedRuntimeDriverTimelineService timelineService = org.mockito.Mockito.mock(UnifiedRuntimeDriverTimelineService.class);
UnifiedRuntimeDerivedProjectionService derivedProjectionService = org.mockito.Mockito.mock(UnifiedRuntimeDerivedProjectionService.class);
RuntimeEventProcessingService runtimeEventProcessingService = org.mockito.Mockito.mock(RuntimeEventProcessingService.class);
MockMvc mockMvc = MockMvcBuilders.standaloneSetup(new UnifiedRuntimeProcessingController(
eventAssemblyService,
timelineService,
derivedProjectionService,
null,
runtimeEventProcessingService
))
.setMessageConverters(new MappingJackson2HttpMessageConverter(objectMapper))
.setControllerAdvice(new UnifiedRuntimeProcessingExceptionHandler())
.build();
UUID sessionId = UUID.randomUUID();
UnifiedRuntimeProcessingRequest request = UnifiedRuntimeProcessingRequest.forTachographFileSession(
sessionId,
"12:123",
OffsetDateTime.parse("2026-05-01T08:00:00Z"),
OffsetDateTime.parse("2026-05-01T10:00:00Z"),
true,
0
);
UnifiedRuntimeDerivedProjectionResultDto driverResult = new UnifiedRuntimeDerivedProjectionResultDto(
request,
2,
1,
3,
5,
List.of(new UnifiedDiscoveredVehicleRef("VEH-1", "VIN-1", "12", "REG-1")),
null,
List.of("processed through generic profile")
);
when(runtimeEventProcessingService.process(any()))
.thenReturn(new RuntimeEventProcessingResultDto(
"tachograph-driver-esper-v1",
RuntimeEventPartitioningStrategy.DRIVER,
request,
5,
1,
1,
List.of(new UnifiedDiscoveredVehicleRef("VEH-1", "VIN-1", "12", "REG-1")),
Map.of("12:123", new RuntimeEventProcessingPartitionResultDto(
"DRIVER",
"12:123",
"UnifiedRuntimeDerivedProjectionResultDto",
driverResult,
Map.of("mergedEventCount", 5)
)),
List.of("generic adapter"),
List.of()
));
mockMvc.perform(post("/api/eventhub/runtime-processing/tachograph/esper-processing")
.contentType("application/json")
.content("""
{
"sessionId": "%s",
"sourceFamilies": ["TACHOGRAPH_FILE_SESSION"],
"driverKey": "12:123",
"occurredFrom": "2026-05-01T08:00:00Z",
"occurredTo": "2026-05-01T10:00:00Z",
"expandVehicleEvents": true,
"significantDrivingMinutes": 3,
"minimumRestPeriodMinutes": 720
}
""".formatted(sessionId)))
.andExpect(status().isOk())
.andExpect(jsonPath("$.inputEventCount").value(5))
.andExpect(jsonPath("$.selectedDriverCount").value(1))
.andExpect(jsonPath("$.driverResults['12:123'].mergedEventCount").value(5))
.andExpect(jsonPath("$.notes[0]").value("generic adapter"));
}
@Test
void validatesTachographParityViaRuntimeApi() throws Exception {
UnifiedRuntimeEventAssemblyService eventAssemblyService = org.mockito.Mockito.mock(UnifiedRuntimeEventAssemblyService.class);
UnifiedRuntimeDriverTimelineService timelineService = org.mockito.Mockito.mock(UnifiedRuntimeDriverTimelineService.class);
UnifiedRuntimeDerivedProjectionService derivedProjectionService = org.mockito.Mockito.mock(UnifiedRuntimeDerivedProjectionService.class);
RuntimeEventProcessingService runtimeEventProcessingService = org.mockito.Mockito.mock(RuntimeEventProcessingService.class);
RuntimeTachographParityValidationService parityValidationService = org.mockito.Mockito.mock(RuntimeTachographParityValidationService.class);
MockMvc mockMvc = MockMvcBuilders.standaloneSetup(new UnifiedRuntimeProcessingController(
eventAssemblyService,
timelineService,
derivedProjectionService,
null,
runtimeEventProcessingService,
parityValidationService
))
.setMessageConverters(new MappingJackson2HttpMessageConverter(objectMapper))
.setControllerAdvice(new UnifiedRuntimeProcessingExceptionHandler())
.build();
UUID sessionId = UUID.randomUUID();
when(parityValidationService.validate(any()))
.thenReturn(new RuntimeTachographParityValidationResultDto(
"tachograph-driver-esper-v1",
"EQUAL",
List.of(sessionId),
1,
Map.of("12:123", new RuntimeTachographDriverParityResultDto(
"12:123",
"EQUAL",
"SINGLE_FILE_SESSION",
1,
true,
List.of(new RuntimeTachographParityCategoryComparisonDto(
"activityIntervals",
2,
2,
true
)),
List.of("validated"),
List.of()
)),
List.of("validation complete"),
List.of()
));
mockMvc.perform(post("/api/eventhub/runtime-processing/event-processing/validation/tachograph-parity")
.contentType("application/json")
.content("""
{
"sessionId": "%s",
"driverKey": "12:123",
"occurredFrom": "2026-05-01T08:00:00Z",
"occurredTo": "2026-05-01T10:00:00Z",
"includeDebug": true
}
""".formatted(sessionId)))
.andExpect(status().isOk())
.andExpect(jsonPath("$.profileKey").value("tachograph-driver-esper-v1"))
.andExpect(jsonPath("$.status").value("EQUAL"))
.andExpect(jsonPath("$.driverResults['12:123'].status").value("EQUAL"))
.andExpect(jsonPath("$.driverResults['12:123'].comparisons[0].category").value("activityIntervals"))
.andExpect(jsonPath("$.driverResults['12:123'].comparisons[0].equal").value(true));
}
@Test
void returnsBadRequestForInvalidRuntimeRequest() throws Exception {
UnifiedRuntimeEventAssemblyService eventAssemblyService = org.mockito.Mockito.mock(UnifiedRuntimeEventAssemblyService.class);

View File

@ -0,0 +1,115 @@
package at.procon.eventhub.processing.driverworkingtime.service;
import static org.assertj.core.api.Assertions.assertThat;
import at.procon.eventhub.config.EventHubProperties;
import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeActivityInterval;
import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeDerivedProjectionBundle;
import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeProcessingInput;
import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeVehicleUsageInterval;
import java.time.OffsetDateTime;
import java.util.List;
import java.util.UUID;
import org.junit.jupiter.api.Test;
class DriverWorkingTimeReusableProjectionBuilderTest {
@Test
void reusesWarmRuntimeWithoutLeakingPreviousState() {
DriverWorkingTimeReusableProjectionBuilder builder =
new DriverWorkingTimeReusableProjectionBuilder(new EventHubProperties());
UUID sessionId = UUID.randomUUID();
OffsetDateTime from = OffsetDateTime.parse("2026-05-01T08:00:00Z");
OffsetDateTime firstDriveEnd = OffsetDateTime.parse("2026-05-01T09:00:00Z");
OffsetDateTime secondDriveStart = OffsetDateTime.parse("2026-05-01T10:00:00Z");
OffsetDateTime to = OffsetDateTime.parse("2026-05-01T11:00:00Z");
DriverWorkingTimeProcessingInput input = new DriverWorkingTimeProcessingInput(
sessionId,
"12:123",
"DRIVER_CARD",
from,
to,
from,
to,
15,
30,
List.of(
new DriverWorkingTimeActivityInterval(
sessionId,
"12:123",
"ACT-1",
"DRIVE",
"DRIVER",
"INSERTED",
"SINGLE",
"12:REG-1",
"VIN-1",
"DRIVER_CARD",
"ACT-1",
"ACT-1",
from,
firstDriveEnd,
from.toEpochSecond(),
firstDriveEnd.toEpochSecond(),
firstDriveEnd.toEpochSecond() - from.toEpochSecond(),
List.of("ACT-1"),
false,
false,
"RAW_INTERVAL"
),
new DriverWorkingTimeActivityInterval(
sessionId,
"12:123",
"ACT-2",
"DRIVE",
"DRIVER",
"INSERTED",
"SINGLE",
"12:REG-1",
"VIN-1",
"DRIVER_CARD",
"ACT-2",
"ACT-2",
secondDriveStart,
to,
secondDriveStart.toEpochSecond(),
to.toEpochSecond(),
to.toEpochSecond() - secondDriveStart.toEpochSecond(),
List.of("ACT-2"),
false,
false,
"RAW_INTERVAL"
)
),
List.of(
new DriverWorkingTimeVehicleUsageInterval(
sessionId,
"12:123",
"VU-1",
"VU-1",
"VU-1",
from,
to,
from.toEpochSecond(),
to.toEpochSecond(),
to.toEpochSecond() - from.toEpochSecond(),
100L,
150L,
"12:REG-1",
"VIN-1",
"DRIVER_CARD",
List.of("VU-1")
)
),
List.of(),
List.of()
);
DriverWorkingTimeDerivedProjectionBundle first = builder.buildDerivedProjectionBundle(input);
DriverWorkingTimeDerivedProjectionBundle second = builder.buildDerivedProjectionBundle(input);
assertThat(second).isEqualTo(first);
assertThat(second.drivingInterruptionIntervals()).hasSize(1);
}
}

View File

@ -0,0 +1,49 @@
package at.procon.eventhub.processing.eventprocessing.dto;
import static org.assertj.core.api.Assertions.assertThat;
import at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest;
import at.procon.eventhub.processing.model.UnifiedEventSourceFamily;
import java.time.OffsetDateTime;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import org.junit.jupiter.api.Test;
class RuntimeEventProcessingApiRequestTest {
@Test
void driverWorkingTimePreservesDisabledVehicleExpansion() {
UnifiedRuntimeProcessingApiRequest scope = new UnifiedRuntimeProcessingApiRequest(
UUID.randomUUID(),
List.of(),
null,
null,
Set.of(UnifiedEventSourceFamily.TACHOGRAPH_FILE_SESSION),
null,
null,
"12:123",
Set.of(),
false,
Set.of(),
false,
null,
null,
null,
OffsetDateTime.parse("2026-05-01T00:00:00Z"),
OffsetDateTime.parse("2026-05-02T00:00:00Z"),
false,
0,
true,
null,
null,
null,
null,
List.of()
);
RuntimeEventProcessingApiRequest request = RuntimeEventProcessingApiRequest.driverWorkingTime(scope);
assertThat(request.partitioning().attachVehicleEvidence()).isFalse();
}
}

View File

@ -15,6 +15,7 @@ import at.procon.eventhub.dto.VehicleRefDto;
import at.procon.eventhub.dto.VehicleRegistrationRefDto;
import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeVehicleUsageInterval;
import at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest;
import at.procon.eventhub.processing.eventprocessing.module.epl.RuntimeEplModuleExecutor;
import at.procon.eventhub.processing.eventprocessing.plan.RuntimeProcessingExecutionApiRequest;
import at.procon.eventhub.processing.model.UnifiedEventSourceFamily;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
@ -78,6 +79,64 @@ class DriverVehicleUsageIntervalsModuleTest {
assertThat(intervals.get(0).endedAt()).isEqualTo(OffsetDateTime.parse("2026-05-01T10:10:00Z"));
}
@Test
void reusesPreparedEplDefinitionAcrossExecutions() {
DriverVehicleUsageIntervalsModule module = new DriverVehicleUsageIntervalsModule(new RuntimeEplModuleExecutor());
RuntimeProcessingModuleContext context = new RuntimeProcessingModuleContext(
new RuntimeProcessingExecutionApiRequest(
"driver-working-time-v1",
runtimeScope(),
null,
List.of(),
Map.of()
),
List.of(
vehicleUsageEvent(
"CVU-1",
EventType.CARD_INSERTED,
EventLifecycle.INSERT,
"2026-05-01T07:50:00Z",
"2026-05-01T07:50:00Z",
"2026-05-01T10:10:00Z",
100_000L,
"INSERTED"
),
vehicleUsageEvent(
"CVU-99",
EventType.CARD_WITHDRAWN,
EventLifecycle.WITHDRAW,
"2026-05-01T10:10:00Z",
"2026-05-01T07:50:00Z",
"2026-05-01T10:10:00Z",
140_000L,
"NOT_INSERTED"
)
),
Map.of(),
Map.of()
);
RuntimeProcessingModuleResult first = module.execute(context);
RuntimeProcessingModuleResult second = module.execute(context);
assertThat(first.metadata())
.containsEntry(RuntimeEplModuleExecutor.DEFINITION_CACHE_HIT_METADATA_KEY, false)
.containsKey(RuntimeEplModuleExecutor.DEFINITION_PREPARATION_MS_METADATA_KEY)
.containsKey(RuntimeEplModuleExecutor.RUNTIME_INIT_MS_METADATA_KEY)
.containsKey(RuntimeEplModuleExecutor.DEPLOY_MS_METADATA_KEY)
.containsKey(RuntimeEplModuleExecutor.LISTENER_REGISTRATION_MS_METADATA_KEY)
.containsKey(RuntimeEplModuleExecutor.EVENT_SEND_MS_METADATA_KEY)
.containsKey(RuntimeEplModuleExecutor.DESTROY_MS_METADATA_KEY);
assertThat(second.metadata())
.containsEntry(RuntimeEplModuleExecutor.DEFINITION_CACHE_HIT_METADATA_KEY, true)
.containsEntry(RuntimeEplModuleExecutor.DEFINITION_PREPARATION_MS_METADATA_KEY, 0L)
.containsKey(RuntimeEplModuleExecutor.RUNTIME_INIT_MS_METADATA_KEY)
.containsKey(RuntimeEplModuleExecutor.DEPLOY_MS_METADATA_KEY)
.containsKey(RuntimeEplModuleExecutor.LISTENER_REGISTRATION_MS_METADATA_KEY)
.containsKey(RuntimeEplModuleExecutor.EVENT_SEND_MS_METADATA_KEY)
.containsKey(RuntimeEplModuleExecutor.DESTROY_MS_METADATA_KEY);
}
private UnifiedRuntimeProcessingApiRequest runtimeScope() {
return new UnifiedRuntimeProcessingApiRequest(
UUID.randomUUID(),

View File

@ -211,6 +211,9 @@ class DriverWorkingTimeDerivedProjectionsModuleTest {
(UnifiedRuntimeDriverWorkingTimeScopeResultDto) result.output();
UnifiedRuntimeDerivedProjectionResultDto driverResult = scopeResult.driverResults().get("12:123");
assertThat(result.metadata()).containsKey("driverProcessingTotalMs");
assertThat(result.metadata()).containsEntry("slowestDriverKey", "12:123");
assertThat(result.metadata()).containsKey("slowestDriverProcessingMs");
assertThat(driverResult.projection().activityIntervalCount()).isEqualTo(1);
assertThat(driverResult.projection().drivingIntervalCount()).isEqualTo(1);
assertThat(driverResult.projection().activityIntervals()).isEmpty();

View File

@ -0,0 +1,119 @@
package at.procon.eventhub.processing.eventprocessing.module;
import static org.assertj.core.api.Assertions.assertThat;
import at.procon.eventhub.processing.eventprocessing.plan.RuntimeProcessingExecutionApiRequest;
import at.procon.eventhub.processing.eventprocessing.partition.RuntimeEventPartitioningStrategy;
import java.util.List;
import java.util.Map;
import org.junit.jupiter.api.Test;
class RuntimeProcessingPipelineExecutorTest {
@Test
void addsExecutionDurationMetadataToEachModuleResult() {
RuntimeProcessingPipelineExecutor executor = new RuntimeProcessingPipelineExecutor(
new RuntimeProcessingModuleRegistry(List.of(
new TestModule("first"),
new TestModule("second")
))
);
Map<String, RuntimeProcessingModuleResult> results = executor.execute(
new RuntimeProcessingExecutionApiRequest(
"driver-working-time-v1",
new at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest(
java.util.UUID.randomUUID(),
List.of(),
null,
null,
java.util.Set.of(at.procon.eventhub.processing.model.UnifiedEventSourceFamily.TACHOGRAPH_FILE_SESSION),
null,
null,
"12:123",
java.util.Set.of(),
false,
java.util.Set.of(),
false,
null,
null,
null,
java.time.OffsetDateTime.parse("2026-05-01T00:00:00Z"),
java.time.OffsetDateTime.parse("2026-05-01T01:00:00Z"),
true,
0,
null,
null,
null,
null,
null
),
new at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventPartitioningApiRequest(
RuntimeEventPartitioningStrategy.DRIVER,
null,
false,
null,
false,
null,
false,
null,
null,
null
),
List.of(),
Map.of()
),
List.of("first", "second"),
null
);
assertThat(results).containsKeys("first", "second");
assertThat(results.get("first").metadata()).containsEntry("marker", "first");
assertThat(results.get("first").metadata()).containsKey(RuntimeProcessingPipelineExecutor.EXECUTION_DURATION_MS_METADATA_KEY);
assertThat(results.get("second").metadata()).containsEntry("marker", "second");
assertThat(results.get("second").metadata()).containsKey(RuntimeProcessingPipelineExecutor.EXECUTION_DURATION_MS_METADATA_KEY);
assertThat(((Number) results.get("first").metadata().get(RuntimeProcessingPipelineExecutor.EXECUTION_DURATION_MS_METADATA_KEY)).longValue())
.isGreaterThanOrEqualTo(0L);
assertThat(((Number) results.get("second").metadata().get(RuntimeProcessingPipelineExecutor.EXECUTION_DURATION_MS_METADATA_KEY)).longValue())
.isGreaterThanOrEqualTo(0L);
}
private static final class TestModule implements RuntimeProcessingModule {
private final String key;
private TestModule(String key) {
this.key = key;
}
@Override
public String moduleKey() {
return key;
}
@Override
public at.procon.eventhub.processing.eventprocessing.plan.RuntimeProcessingModuleDescriptorDto descriptor() {
return new at.procon.eventhub.processing.eventprocessing.plan.RuntimeProcessingModuleDescriptorDto(
key,
key,
"test",
"JAVA",
java.util.Set.of()
);
}
@Override
public RuntimeProcessingModuleResult execute(RuntimeProcessingModuleContext context) {
if ("second".equals(key)) {
assertThat(context.previousResults()).containsKey("first");
}
return new RuntimeProcessingModuleResult(
key,
RuntimeProcessingModuleStatus.SUCCESS,
null,
Map.of("marker", key),
List.of()
);
}
}
}

View File

@ -0,0 +1,137 @@
package at.procon.eventhub.processing.eventprocessing.module;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeDriverPartition;
import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeVehicleUsageInterval;
import at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest;
import at.procon.eventhub.processing.eventprocessing.plan.DriverWorkingTimeRuntimeProcessingPlan;
import at.procon.eventhub.processing.eventprocessing.plan.RuntimeProcessingExecutionApiRequest;
import at.procon.eventhub.processing.model.RuntimeDriverVehicleEvidenceAttachmentResult;
import at.procon.eventhub.processing.model.UnifiedEventSourceFamily;
import at.procon.eventhub.processing.model.UnifiedRuntimeEventBundle;
import at.procon.eventhub.processing.model.UnifiedRuntimeProcessingRequest;
import at.procon.eventhub.processing.service.RuntimeDriverVehicleEvidenceAttachmentService;
import java.time.OffsetDateTime;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import org.junit.jupiter.api.Test;
class VehicleEvidenceAttachmentModuleTest {
@Test
void usesAttachmentAttributeInsteadOfScopeExpansionFlag() {
RuntimeDriverVehicleEvidenceAttachmentService service =
org.mockito.Mockito.mock(RuntimeDriverVehicleEvidenceAttachmentService.class);
VehicleEvidenceAttachmentModule module = new VehicleEvidenceAttachmentModule(service);
UnifiedRuntimeProcessingApiRequest scope = new UnifiedRuntimeProcessingApiRequest(
UUID.randomUUID(),
List.of(),
null,
null,
Set.of(UnifiedEventSourceFamily.TACHOGRAPH_FILE_SESSION),
null,
null,
"12:123",
Set.of(),
false,
Set.of(),
false,
null,
null,
null,
OffsetDateTime.parse("2026-05-01T00:00:00Z"),
OffsetDateTime.parse("2026-05-02T00:00:00Z"),
true,
15,
true,
null,
null,
null,
null,
List.of()
);
UnifiedRuntimeProcessingRequest runtimeRequest = scope.toRuntimeRequest();
UnifiedRuntimeEventBundle bundle = new UnifiedRuntimeEventBundle(
runtimeRequest,
List.of(),
List.of(),
List.of(),
List.of(),
List.of()
);
when(service.attachVehicleEvidence(
eq("12:123"),
eq(List.of()),
eq(List.of()),
eq(List.<DriverWorkingTimeVehicleUsageInterval>of()),
eq(false),
eq(3),
anyBoolean()
)).thenReturn(new RuntimeDriverVehicleEvidenceAttachmentResult(
"12:123",
List.of(),
List.of(),
List.of(),
0,
0,
0,
List.of(),
List.of(),
List.of(),
List.of()
));
RuntimeProcessingModuleContext context = new RuntimeProcessingModuleContext(
new RuntimeProcessingExecutionApiRequest("driver-working-time-v1", scope, null, List.of(), Map.of()),
List.of(),
Map.of(
"runtimeScopeApiRequest", scope,
DriverWorkingTimeRuntimeProcessingPlan.ATTACH_VEHICLE_ONLY_EVENTS_ATTRIBUTE, false,
DriverWorkingTimeRuntimeProcessingPlan.VEHICLE_EVIDENCE_PADDING_MINUTES_ATTRIBUTE, 3
),
Map.of(
DriverWorkingTimeModuleKeys.RUNTIME_EVENT_ASSEMBLY,
new RuntimeProcessingModuleResult(
DriverWorkingTimeModuleKeys.RUNTIME_EVENT_ASSEMBLY,
RuntimeProcessingModuleStatus.SUCCESS,
bundle,
Map.of(),
List.of()
),
DriverWorkingTimeModuleKeys.VEHICLE_USAGE_MERGE,
new RuntimeProcessingModuleResult(
DriverWorkingTimeModuleKeys.VEHICLE_USAGE_MERGE,
RuntimeProcessingModuleStatus.SUCCESS,
List.of(),
Map.of(),
List.of()
)
)
);
RuntimeProcessingModuleResult result = module.execute(context);
verify(service).attachVehicleEvidence(
eq("12:123"),
eq(List.of()),
eq(List.of()),
eq(List.<DriverWorkingTimeVehicleUsageInterval>of()),
eq(false),
eq(3),
eq(false)
);
@SuppressWarnings("unchecked")
Map<String, DriverWorkingTimeDriverPartition> partitions =
(Map<String, DriverWorkingTimeDriverPartition>) result.output();
org.assertj.core.api.Assertions.assertThat(partitions).containsKey("12:123");
}
}

View File

@ -0,0 +1,72 @@
package at.procon.eventhub.processing.eventprocessing.plan;
import static org.assertj.core.api.Assertions.assertThat;
import at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest;
import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventPartitioningApiRequest;
import at.procon.eventhub.processing.eventprocessing.partition.RuntimeEventPartitioningStrategy;
import at.procon.eventhub.processing.model.UnifiedEventSourceFamily;
import at.procon.eventhub.processing.service.RuntimeDriverWorkingTimeScopeProcessingService;
import java.time.OffsetDateTime;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import org.junit.jupiter.api.Test;
class DriverWorkingTimeRuntimeProcessingPlanTest {
@Test
void applyExecutionRequestDoesNotRewriteScopeExpansionFromAttachmentFlag() {
DriverWorkingTimeRuntimeProcessingPlan plan = new DriverWorkingTimeRuntimeProcessingPlan(
org.mockito.Mockito.mock(RuntimeDriverWorkingTimeScopeProcessingService.class)
);
UnifiedRuntimeProcessingApiRequest scope = new UnifiedRuntimeProcessingApiRequest(
UUID.randomUUID(),
List.of(),
null,
null,
Set.of(UnifiedEventSourceFamily.TACHOGRAPH_FILE_SESSION),
null,
null,
"12:123",
Set.of(),
false,
Set.of(),
false,
null,
null,
null,
OffsetDateTime.parse("2026-05-01T00:00:00Z"),
OffsetDateTime.parse("2026-05-02T00:00:00Z"),
true,
15,
true,
null,
null,
null,
null,
List.of()
);
UnifiedRuntimeProcessingApiRequest resolved = plan.applyExecutionRequest(
scope,
new RuntimeEventPartitioningApiRequest(
RuntimeEventPartitioningStrategy.DRIVER,
Set.of(),
false,
Set.of(),
false,
Set.of(),
false,
false,
0,
false
),
Map.of("attachVehicleOnlyEvents", false)
);
assertThat(resolved.expandVehicleEvents()).isTrue();
assertThat(resolved.vehicleExpansionPaddingMinutes()).isEqualTo(15);
}
}

View File

@ -0,0 +1,49 @@
package at.procon.eventhub.processing.eventprocessing.plan;
import static org.assertj.core.api.Assertions.assertThat;
import at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest;
import at.procon.eventhub.processing.model.UnifiedEventSourceFamily;
import java.time.OffsetDateTime;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import org.junit.jupiter.api.Test;
class RuntimeProcessingExecutionApiRequestTest {
@Test
void driverWorkingTimePreservesDisabledVehicleExpansion() {
UnifiedRuntimeProcessingApiRequest scope = new UnifiedRuntimeProcessingApiRequest(
UUID.randomUUID(),
List.of(),
null,
null,
Set.of(UnifiedEventSourceFamily.TACHOGRAPH_FILE_SESSION),
null,
null,
"12:123",
Set.of(),
false,
Set.of(),
false,
null,
null,
null,
OffsetDateTime.parse("2026-05-01T00:00:00Z"),
OffsetDateTime.parse("2026-05-02T00:00:00Z"),
false,
0,
true,
null,
null,
null,
null,
List.of()
);
RuntimeProcessingExecutionApiRequest request = RuntimeProcessingExecutionApiRequest.driverWorkingTime(scope);
assertThat(request.partitioning().attachVehicleEvidence()).isFalse();
}
}

View File

@ -1,10 +1,6 @@
package at.procon.eventhub.processing.eventprocessing.profile;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import at.procon.eventhub.processing.dto.UnifiedRuntimeDerivedProjectionResultDto;
import at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest;
@ -12,28 +8,29 @@ import at.procon.eventhub.processing.dto.UnifiedRuntimeDriverWorkingTimeScopeRes
import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventPartitioningApiRequest;
import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingApiRequest;
import at.procon.eventhub.processing.eventprocessing.partition.RuntimeEventPartitioningStrategy;
import at.procon.eventhub.processing.eventprocessing.plan.DriverWorkingTimeRuntimeProcessingPlan;
import at.procon.eventhub.processing.model.UnifiedEventSourceFamily;
import at.procon.eventhub.processing.model.UnifiedRuntimeProcessingRequest;
import at.procon.eventhub.processing.service.RuntimeDriverWorkingTimeScopeProcessingService;
import at.procon.eventhub.tachographfilesession.processing.profile.TachographDriverEsperRuntimeEventProcessingProfile;
import java.time.OffsetDateTime;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
class TachographDriverEsperRuntimeEventProcessingProfileTest {
class DriverWorkingTimeRuntimeEventProcessingProfileTest {
@Test
void exposesDiscoveryMetadata() {
TachographDriverEsperRuntimeEventProcessingProfile profile = new TachographDriverEsperRuntimeEventProcessingProfile(
DriverWorkingTimeRuntimeEventProcessingProfile profile = new DriverWorkingTimeRuntimeEventProcessingProfile(
new DriverWorkingTimeRuntimeProcessingPlan(
org.mockito.Mockito.mock(RuntimeDriverWorkingTimeScopeProcessingService.class)
)
);
assertThat(profile.profileKey()).isEqualTo("tachograph-driver-esper-v1");
assertThat(profile.displayName()).isEqualTo("Tachograph Driver Esper Processing");
assertThat(profile.profileKey()).isEqualTo("driver-working-time-v1");
assertThat(profile.displayName()).isEqualTo("Driver working-time processing");
assertThat(profile.defaultPartitioningStrategy()).isEqualTo(RuntimeEventPartitioningStrategy.DRIVER);
assertThat(profile.supportedPartitioningStrategies()).containsExactly(RuntimeEventPartitioningStrategy.DRIVER);
assertThat(profile.optionalParameters()).containsExactlyInAnyOrder(
@ -48,9 +45,11 @@ class TachographDriverEsperRuntimeEventProcessingProfileTest {
}
@Test
void delegatesToTachographScopeServiceAndMapsPartitionResults() {
void delegatesToDriverWorkingTimePlanAndMapsPartitionResults() {
RuntimeDriverWorkingTimeScopeProcessingService scopeService = org.mockito.Mockito.mock(RuntimeDriverWorkingTimeScopeProcessingService.class);
TachographDriverEsperRuntimeEventProcessingProfile profile = new TachographDriverEsperRuntimeEventProcessingProfile(scopeService);
DriverWorkingTimeRuntimeEventProcessingProfile profile = new DriverWorkingTimeRuntimeEventProcessingProfile(
new DriverWorkingTimeRuntimeProcessingPlan(scopeService)
);
UUID sessionId = UUID.randomUUID();
UnifiedRuntimeProcessingApiRequest scope = new UnifiedRuntimeProcessingApiRequest(
@ -80,7 +79,7 @@ class TachographDriverEsperRuntimeEventProcessingProfileTest {
null
);
RuntimeEventProcessingApiRequest request = new RuntimeEventProcessingApiRequest(
TachographDriverEsperRuntimeEventProcessingProfile.PROFILE_KEY,
DriverWorkingTimeRuntimeEventProcessingProfile.PROFILE_KEY,
scope,
new RuntimeEventPartitioningApiRequest(
RuntimeEventPartitioningStrategy.DRIVER,
@ -137,7 +136,7 @@ class TachographDriverEsperRuntimeEventProcessingProfileTest {
null,
List.of("driver processed")
);
when(scopeService.processScope(any(), anyBoolean()))
org.mockito.Mockito.when(scopeService.processScope(org.mockito.ArgumentMatchers.any(), org.mockito.ArgumentMatchers.anyBoolean()))
.thenReturn(new UnifiedRuntimeDriverWorkingTimeScopeResultDto(
processedRequest,
5,
@ -152,23 +151,10 @@ class TachographDriverEsperRuntimeEventProcessingProfileTest {
var result = profile.process(request);
assertThat(result.profileKey()).isEqualTo(TachographDriverEsperRuntimeEventProcessingProfile.PROFILE_KEY);
assertThat(result.profileKey()).isEqualTo(DriverWorkingTimeRuntimeEventProcessingProfile.PROFILE_KEY);
assertThat(result.partitioningStrategy()).isEqualTo(RuntimeEventPartitioningStrategy.DRIVER);
assertThat(result.partitionResults()).containsOnlyKeys("12:DRIVER-1");
assertThat(result.partitionResults().get("12:DRIVER-1").partitionType()).isEqualTo("DRIVER");
assertThat(result.partitionResults().get("12:DRIVER-1").result()).isSameAs(driverResult);
ArgumentCaptor<UnifiedRuntimeProcessingApiRequest> captor = ArgumentCaptor.forClass(UnifiedRuntimeProcessingApiRequest.class);
ArgumentCaptor<Boolean> debugCaptor = ArgumentCaptor.forClass(Boolean.class);
verify(scopeService).processScope(captor.capture(), debugCaptor.capture());
UnifiedRuntimeProcessingApiRequest delegated = captor.getValue();
assertThat(delegated.driverKeys()).containsExactly("12:DRIVER-1");
assertThat(delegated.significantDrivingMinutes()).isEqualTo(5);
assertThat(delegated.minimumRestPeriodMinutes()).isEqualTo(600);
assertThat(delegated.vehicleExpansionPaddingMinutes()).isEqualTo(20);
assertThat(delegated.expandVehicleEvents()).isTrue();
assertThat(delegated.includeActivityIntervals()).isTrue();
assertThat(delegated.includeDrivingIntervals()).isTrue();
assertThat(debugCaptor.getValue()).isTrue();
}
}

View File

@ -14,7 +14,7 @@ import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingA
import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingPartitionResultDto;
import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingResultDto;
import at.procon.eventhub.processing.eventprocessing.partition.RuntimeEventPartitioningStrategy;
import at.procon.eventhub.tachographfilesession.processing.profile.TachographDriverEsperRuntimeEventProcessingProfile;
import at.procon.eventhub.processing.eventprocessing.profile.DriverWorkingTimeRuntimeEventProcessingProfile;
import at.procon.eventhub.processing.model.UnifiedEventSourceFamily;
import at.procon.eventhub.processing.model.UnifiedRuntimeEventBackend;
import at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest;
@ -58,7 +58,7 @@ class RuntimeMixedSourceEvidenceValidationServiceTest {
)
);
when(processingService.process(any())).thenReturn(new RuntimeEventProcessingResultDto(
TachographDriverEsperRuntimeEventProcessingProfile.PROFILE_KEY,
DriverWorkingTimeRuntimeEventProcessingProfile.PROFILE_KEY,
RuntimeEventPartitioningStrategy.DRIVER,
null,
3,
@ -78,7 +78,7 @@ class RuntimeMixedSourceEvidenceValidationServiceTest {
RuntimeMixedSourceEvidenceValidationResultDto result = service.validate(new RuntimeMixedSourceEvidenceValidationApiRequest(
new RuntimeEventProcessingApiRequest(
TachographDriverEsperRuntimeEventProcessingProfile.PROFILE_KEY,
DriverWorkingTimeRuntimeEventProcessingProfile.PROFILE_KEY,
new UnifiedRuntimeProcessingApiRequest(
null,
List.of(),

View File

@ -10,7 +10,7 @@ import at.procon.eventhub.processing.eventprocessing.RuntimeEventProcessingServi
import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingPartitionResultDto;
import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingResultDto;
import at.procon.eventhub.processing.eventprocessing.partition.RuntimeEventPartitioningStrategy;
import at.procon.eventhub.tachographfilesession.processing.profile.TachographDriverEsperRuntimeEventProcessingProfile;
import at.procon.eventhub.processing.eventprocessing.profile.DriverWorkingTimeRuntimeEventProcessingProfile;
import at.procon.eventhub.tachographfilesession.processing.validation.RuntimeTachographParityValidationApiRequest;
import at.procon.eventhub.tachographfilesession.processing.validation.RuntimeTachographParityValidationResultDto;
import at.procon.eventhub.tachographfilesession.processing.validation.RuntimeTachographParityValidationService;
@ -73,7 +73,7 @@ class RuntimeTachographParityValidationServiceTest {
);
when(runtimeEventProcessingService.process(any()))
.thenReturn(new RuntimeEventProcessingResultDto(
TachographDriverEsperRuntimeEventProcessingProfile.PROFILE_KEY,
DriverWorkingTimeRuntimeEventProcessingProfile.PROFILE_KEY,
RuntimeEventPartitioningStrategy.DRIVER,
runtimeRequest,
4,

View File

@ -28,6 +28,7 @@ import at.procon.eventhub.tachographfilesession.model.TachographCompositeSession
import at.procon.eventhub.tachographfilesession.service.InMemoryTachographCompositeSessionRepository;
import at.procon.eventhub.tachographfilesession.service.InMemoryTachographFileSessionRepository;
import at.procon.eventhub.tachographfilesession.service.IntervalBackedDriverTimelineEventBuilder;
import at.procon.eventhub.tachographfilesession.service.RawSourceDriverTimelineEventBuilder;
import at.procon.eventhub.tachographfilesession.service.TachographFileSessionRepository;
import at.procon.eventhub.tachographfilesession.service.VehicleKeyFactory;
import com.fasterxml.jackson.databind.ObjectMapper;
@ -53,9 +54,10 @@ class TachographFileSessionRuntimeEventLoaderTest {
new VehicleKeyFactory(),
new EventDetailsFactory(new ObjectMapper())
);
RawSourceDriverTimelineEventBuilder rawSourceEventBuilder = new RawSourceDriverTimelineEventBuilder(eventBuilder);
TachographFileSessionRuntimeEventLoader loader = new TachographFileSessionRuntimeEventLoader(
new UnifiedDriverEventSourceService(List.of(new TachographFileSessionUnifiedDriverEventSource(repository, eventBuilder))),
new UnifiedVehicleEventSourceService(List.of(new TachographFileSessionUnifiedVehicleEventSource(repository, eventBuilder))),
new UnifiedDriverEventSourceService(List.of(new TachographFileSessionUnifiedDriverEventSource(repository, rawSourceEventBuilder))),
new UnifiedVehicleEventSourceService(List.of(new TachographFileSessionUnifiedVehicleEventSource(repository, rawSourceEventBuilder))),
compositeRepository,
new EventAcquisitionRecordKeyService(),
new EventHubEventSorter()
@ -90,9 +92,10 @@ class TachographFileSessionRuntimeEventLoaderTest {
new VehicleKeyFactory(),
new EventDetailsFactory(new ObjectMapper())
);
RawSourceDriverTimelineEventBuilder rawSourceEventBuilder = new RawSourceDriverTimelineEventBuilder(eventBuilder);
TachographFileSessionRuntimeEventLoader loader = new TachographFileSessionRuntimeEventLoader(
new UnifiedDriverEventSourceService(List.of(new TachographFileSessionUnifiedDriverEventSource(repository, eventBuilder))),
new UnifiedVehicleEventSourceService(List.of(new TachographFileSessionUnifiedVehicleEventSource(repository, eventBuilder))),
new UnifiedDriverEventSourceService(List.of(new TachographFileSessionUnifiedDriverEventSource(repository, rawSourceEventBuilder))),
new UnifiedVehicleEventSourceService(List.of(new TachographFileSessionUnifiedVehicleEventSource(repository, rawSourceEventBuilder))),
compositeRepository,
new EventAcquisitionRecordKeyService(),
new EventHubEventSorter()
@ -135,9 +138,10 @@ class TachographFileSessionRuntimeEventLoaderTest {
new VehicleKeyFactory(),
new EventDetailsFactory(new ObjectMapper())
);
RawSourceDriverTimelineEventBuilder rawSourceEventBuilder = new RawSourceDriverTimelineEventBuilder(eventBuilder);
TachographFileSessionRuntimeEventLoader loader = new TachographFileSessionRuntimeEventLoader(
new UnifiedDriverEventSourceService(List.of(new TachographFileSessionUnifiedDriverEventSource(repository, eventBuilder))),
new UnifiedVehicleEventSourceService(List.of(new TachographFileSessionUnifiedVehicleEventSource(repository, eventBuilder))),
new UnifiedDriverEventSourceService(List.of(new TachographFileSessionUnifiedDriverEventSource(repository, rawSourceEventBuilder))),
new UnifiedVehicleEventSourceService(List.of(new TachographFileSessionUnifiedVehicleEventSource(repository, rawSourceEventBuilder))),
compositeRepository,
new EventAcquisitionRecordKeyService(),
new EventHubEventSorter()

View File

@ -23,6 +23,7 @@ import at.procon.eventhub.tachographfilesession.service.DriverKeyFactory;
import at.procon.eventhub.tachographfilesession.service.DriverTimelineBuilder;
import at.procon.eventhub.tachographfilesession.service.InMemoryTachographFileSessionRepository;
import at.procon.eventhub.tachographfilesession.service.IntervalBackedDriverTimelineEventBuilder;
import at.procon.eventhub.tachographfilesession.service.RawSourceDriverTimelineEventBuilder;
import at.procon.eventhub.tachographfilesession.service.TachographFileSessionRepository;
import at.procon.eventhub.tachographfilesession.service.VehicleKeyFactory;
import com.fasterxml.jackson.databind.ObjectMapper;
@ -45,12 +46,12 @@ class UnifiedDriverEventSourceServiceTest {
UnifiedDriverEventSourceService service = new UnifiedDriverEventSourceService(List.of(
new TachographFileSessionUnifiedDriverEventSource(
repository,
new IntervalBackedDriverTimelineEventBuilder(
new RawSourceDriverTimelineEventBuilder(new IntervalBackedDriverTimelineEventBuilder(
timelineBuilder,
new DriverKeyFactory(),
new VehicleKeyFactory(),
new EventDetailsFactory(new ObjectMapper())
)
))
)
));

View File

@ -23,6 +23,7 @@ import at.procon.eventhub.tachographfilesession.service.DriverKeyFactory;
import at.procon.eventhub.tachographfilesession.service.DriverTimelineBuilder;
import at.procon.eventhub.tachographfilesession.service.InMemoryTachographFileSessionRepository;
import at.procon.eventhub.tachographfilesession.service.IntervalBackedDriverTimelineEventBuilder;
import at.procon.eventhub.tachographfilesession.service.RawSourceDriverTimelineEventBuilder;
import at.procon.eventhub.tachographfilesession.service.TachographFileSessionRepository;
import at.procon.eventhub.tachographfilesession.service.VehicleKeyFactory;
import com.fasterxml.jackson.databind.ObjectMapper;
@ -42,15 +43,16 @@ class UnifiedVehicleEventSourceServiceTest {
EventHubProperties properties = new EventHubProperties();
TachographFileSessionRepository repository = new InMemoryTachographFileSessionRepository(properties);
DriverTimelineBuilder timelineBuilder = new DriverTimelineBuilder();
UnifiedVehicleEventSourceService service = new UnifiedVehicleEventSourceService(List.of(
new TachographFileSessionUnifiedVehicleEventSource(
repository,
new IntervalBackedDriverTimelineEventBuilder(
IntervalBackedDriverTimelineEventBuilder eventBuilder = new IntervalBackedDriverTimelineEventBuilder(
timelineBuilder,
new DriverKeyFactory(),
new VehicleKeyFactory(),
new EventDetailsFactory(new ObjectMapper())
)
);
UnifiedVehicleEventSourceService service = new UnifiedVehicleEventSourceService(List.of(
new TachographFileSessionUnifiedVehicleEventSource(
repository,
new RawSourceDriverTimelineEventBuilder(eventBuilder)
)
));