-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Browse files
Browse the repository at this point in the history
- Loading branch information
1 parent
25d357d
commit b72d505
Showing
15 changed files
with
412 additions
and
44 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
85 changes: 85 additions & 0 deletions
85
sfera-mock/src/main/java/ch/sbb/sferamock/messages/services/EventRepository.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,85 @@ | ||
| package ch.sbb.sferamock.messages.services; | ||
|
|
||
| import ch.sbb.sferamock.adapters.sfera.model.v0201.G2BEventPayload; | ||
| import ch.sbb.sferamock.messages.common.XmlHelper; | ||
| import java.io.File; | ||
| import java.io.FileInputStream; | ||
| import java.io.IOException; | ||
| import java.io.InputStream; | ||
| import java.util.ArrayList; | ||
| import java.util.HashMap; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.regex.Matcher; | ||
| import java.util.regex.Pattern; | ||
| import org.springframework.boot.ApplicationArguments; | ||
| import org.springframework.boot.ApplicationRunner; | ||
| import org.springframework.core.io.support.PathMatchingResourcePatternResolver; | ||
| import org.springframework.stereotype.Service; | ||
|
|
||
| @Service | ||
| public class EventRepository implements ApplicationRunner { | ||
|
|
||
| private static final String XML_RESOURCES_CLASSPATH = "classpath:static_sfera_resources/*/SFERA_Event_*.xml"; | ||
| private static final String XML_REGEX = "/([a-zA-Z0-9]+)_\\w+/SFERA_Event_([a-zA-Z0-9]+)_\\d+\\.xml"; | ||
| private static final String OFFSET_XML_REGEX = "SFERA_Event_[a-zA-Z0-9]+_(\\d+)\\.xml"; | ||
| private final XmlHelper xmlHelper; | ||
|
|
||
| Map<String, List<Event>> events = new HashMap<>(); | ||
|
|
||
| public EventRepository(XmlHelper xmlHelper) { | ||
| this.xmlHelper = xmlHelper; | ||
| } | ||
|
|
||
| @Override | ||
| public void run(ApplicationArguments args) throws Exception { | ||
| importEvents(); | ||
| } | ||
|
|
||
| private void importEvents() throws IOException { | ||
| PathMatchingResourcePatternResolver resolver = new PathMatchingResourcePatternResolver(); | ||
| var resources = resolver.getResources(XML_RESOURCES_CLASSPATH); | ||
| for (var resource : resources) { | ||
| File file = resource.getFile(); | ||
| var operationalNumber = extractOperationalNumber(file.getPath()); | ||
| var offsetMs = extractOffsetMs(file.getName()); | ||
| try (InputStream in = new FileInputStream(file)) { | ||
| String xmlPayload = new String(in.readAllBytes()); | ||
| var eventPayload = xmlHelper.xmlToObject(xmlPayload); | ||
|
|
||
| List<Event> trainEvents = events.get(operationalNumber); | ||
| if (trainEvents == null) { | ||
| trainEvents = new ArrayList<>(); | ||
| } | ||
| trainEvents.add(new Event(offsetMs, (G2BEventPayload) eventPayload)); | ||
| events.put(operationalNumber, trainEvents); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private static String extractOperationalNumber(String filename) { | ||
| Pattern pattern = Pattern.compile(XML_REGEX); | ||
| Matcher matcher = pattern.matcher(filename); | ||
| if (matcher.find()) { | ||
| String directoryOperationalNumber = matcher.group(1); | ||
| String fileOperationalNumber = matcher.group(2); | ||
| if (directoryOperationalNumber != null && directoryOperationalNumber.equals(fileOperationalNumber)) { | ||
| return directoryOperationalNumber; | ||
| } | ||
| } | ||
| throw new RuntimeException("Operational number extraction in Event repository failed for file: " + filename); | ||
| } | ||
|
|
||
| private static int extractOffsetMs(String filename) { | ||
| Pattern pattern = Pattern.compile(OFFSET_XML_REGEX); | ||
| Matcher matcher = pattern.matcher(filename); | ||
| if (matcher.find()) { | ||
| return Integer.parseInt(matcher.group(1)); | ||
| } | ||
| throw new RuntimeException("Offset extraction in Event repository failed for file: " + filename); | ||
| } | ||
|
|
||
| public record Event(int offsetMs, G2BEventPayload payload) { | ||
|
|
||
| } | ||
| } |
71 changes: 71 additions & 0 deletions
71
sfera-mock/src/main/java/ch/sbb/sferamock/messages/services/EventService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| package ch.sbb.sferamock.messages.services; | ||
|
|
||
| import ch.sbb.sferamock.adapters.sfera.model.v0201.G2BEventPayload; | ||
| import ch.sbb.sferamock.messages.model.ClientId; | ||
| import ch.sbb.sferamock.messages.model.RequestContext; | ||
| import ch.sbb.sferamock.messages.services.EventRepository.Event; | ||
| import ch.sbb.sferamock.messages.sfera.EventPublisher; | ||
| import java.time.Instant; | ||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
| import java.util.concurrent.ScheduledFuture; | ||
| import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; | ||
| import org.springframework.stereotype.Service; | ||
|
|
||
| @Service | ||
| public class EventService { | ||
|
|
||
| private final EventRepository eventRepository; | ||
| private final ThreadPoolTaskScheduler taskScheduler; | ||
| private final EventPublisher eventPublisher; | ||
| private final Map<ClientId, List<ScheduledFuture<?>>> scheduledTasks = new ConcurrentHashMap<>(); | ||
|
|
||
| public EventService(EventRepository eventRepository, EventPublisher eventPublisher) { | ||
| this.eventRepository = eventRepository; | ||
| this.eventPublisher = eventPublisher; | ||
| this.taskScheduler = new ThreadPoolTaskScheduler(); | ||
| this.taskScheduler.initialize(); | ||
| } | ||
|
|
||
| public void registerActiveTrain(RequestContext requestContext) { | ||
| if (hasActiveFutures(requestContext.clientId())) { | ||
| return; | ||
| } | ||
| List<Event> events = this.eventRepository.events.get(requestContext.tid().operationalNumber()); | ||
| if (events == null) { | ||
| return; | ||
| } | ||
| List<ScheduledFuture<?>> futures = new ArrayList<>(); | ||
| for (Event event : events) { | ||
| ScheduledFuture<?> future = taskScheduler.schedule(() -> processEvent(event.payload(), requestContext), Instant.now().plusMillis(event.offsetMs())); | ||
| futures.add(future); | ||
| } | ||
| scheduledTasks.put(requestContext.clientId(), futures); | ||
| } | ||
|
|
||
| public void deregisterActiveTrain(ClientId clientId) { | ||
| List<ScheduledFuture<?>> futures = scheduledTasks.remove(clientId); | ||
| if (futures != null) { | ||
| for (ScheduledFuture<?> future : futures) { | ||
| future.cancel(false); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private void processEvent(G2BEventPayload eventPayload, RequestContext requestContext) { | ||
| if (eventPayload.getRelatedTrainInformation() != null) { | ||
| eventPublisher.publishRelatedTrainInformation(eventPayload, requestContext); | ||
| } else if (eventPayload.getJourneyProfile() != null) { | ||
| eventPublisher.publishJourneyProfile(eventPayload, requestContext); | ||
| } else { | ||
| eventPublisher.publishEventPayload(eventPayload, requestContext); | ||
| } | ||
| } | ||
|
|
||
| private boolean hasActiveFutures(ClientId clientId) { | ||
| List<ScheduledFuture<?>> scheduledFutures = scheduledTasks.get(clientId); | ||
| return scheduledFutures != null && scheduledFutures.stream().anyMatch(scheduledFuture -> !scheduledFuture.isDone() && !scheduledFuture.isCancelled()); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
60 changes: 60 additions & 0 deletions
60
sfera-mock/src/main/java/ch/sbb/sferamock/messages/sfera/EventPublisher.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| package ch.sbb.sferamock.messages.sfera; | ||
|
|
||
| import static org.springframework.util.MimeTypeUtils.APPLICATION_XML; | ||
|
|
||
| import ch.sbb.sferamock.adapters.sfera.model.v0201.G2BEventPayload; | ||
| import ch.sbb.sferamock.adapters.sfera.model.v0201.SFERAG2BEventMessage; | ||
| import ch.sbb.sferamock.messages.common.XmlHelper; | ||
| import ch.sbb.sferamock.messages.model.RequestContext; | ||
| import java.util.UUID; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
| import org.springframework.beans.factory.annotation.Value; | ||
| import org.springframework.cloud.stream.function.StreamBridge; | ||
| import org.springframework.messaging.support.MessageBuilder; | ||
| import org.springframework.stereotype.Service; | ||
|
|
||
| @Service | ||
| public class EventPublisher { | ||
|
|
||
| private static final Logger log = LoggerFactory.getLogger(EventPublisher.class); | ||
| private static final String SOLACE_BINDER = "solace"; | ||
| private final XmlHelper xmlHelper; | ||
| private final StreamBridge streamBridge; | ||
| private final SferaMessageCreator sferaMessageCreator; | ||
| @Value("${spring.cloud.stream.bindings.publishEvent-out-0.destination}") | ||
| private String publishDestination; | ||
|
|
||
| public EventPublisher(XmlHelper xmlHelper, StreamBridge streamBridge, SferaMessageCreator sferaMessageCreator) { | ||
| this.xmlHelper = xmlHelper; | ||
| this.streamBridge = streamBridge; | ||
| this.sferaMessageCreator = sferaMessageCreator; | ||
| } | ||
|
|
||
| public void publishRelatedTrainInformation(G2BEventPayload eventPayload, RequestContext requestContext) { | ||
| var header = sferaMessageCreator.createMessageHeader(UUID.randomUUID(), requestContext.tid()); | ||
| var event = sferaMessageCreator.createRelatedTrainInformation(eventPayload, header); | ||
| publishEvent(event, requestContext); | ||
| } | ||
|
|
||
| public void publishJourneyProfile(G2BEventPayload eventPayload, RequestContext requestContext) { | ||
| var header = sferaMessageCreator.createMessageHeader(UUID.randomUUID(), requestContext.tid()); | ||
| var event = sferaMessageCreator.createJourneyProfileEventMessage(eventPayload, header); | ||
| publishEvent(event, requestContext); | ||
| } | ||
|
|
||
| public void publishEventPayload(G2BEventPayload eventPayload, RequestContext requestContext) { | ||
| var header = sferaMessageCreator.createMessageHeader(UUID.randomUUID(), requestContext.tid()); | ||
| var event = sferaMessageCreator.createEventMessage(eventPayload, header); | ||
| publishEvent(event, requestContext); | ||
| } | ||
|
|
||
| private void publishEvent(SFERAG2BEventMessage eventMessage, RequestContext requestContext) { | ||
| String topic = SferaTopicHelper.getG2BEventTopic(publishDestination, requestContext); | ||
| log.info("Publishing Event Message: {} to topic {}", xmlHelper.toString(eventMessage), topic); | ||
| streamBridge.send(topic, SOLACE_BINDER, MessageBuilder | ||
| .withPayload(eventMessage) | ||
| .build(), | ||
| APPLICATION_XML); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.