// Path: distribution/src/main/java/nl/crowndov/displaydirect/distribution/chb/StopStore.java
// public class StopStore {
// 
//     private static final Logger LOGGER = LoggerFactory.getLogger(StopStore.class);
// 
//     private static Map<String, String> store = new HashMap<>();
//     private static Set<String> missing = new HashSet<>();
// 
//     public static Optional<String> getQuayFromCode(String dataOwner, String stopCode) {
//         String key = dataOwner+"|"+stopCode;
//         if (store.containsKey(key)) {
//             return Optional.of(store.get(key));
//         }
//         missing.add(key);
//         return Optional.empty();
//     }
// 
//     public static boolean stopExists(String quayOrStopCode) {
//         return store.values().contains(quayOrStopCode);
//     }
// 
//     public static Map<String, String> getStore() {
//         return store;
//     }
// 
//     public static Set<String> getMissing() {
//         return missing;
//     }
// 
//     public static void setStore(Map<String, String> storeData) {
//         store = storeData;
//     }
// }
// 
// Path: distribution/src/main/java/nl/crowndov/displaydirect/distribution/domain/travelinfo/RealtimeMessage.java
// public class RealtimeMessage implements Serializable {
// 
//     private static final long serialVersionUID = -8692508260190453615L;
// 
//     private Type type;
//     private LocalDate date;
//     private String dataOwnerCode;
//     private String stopCode;
// 
//     private String quayCode;
// 
//     public RealtimeMessage() {
//     }
// 
//     public RealtimeMessage(Type type, LocalDate date, String dataOwnerCode, String stopCode) {
//         this.type = type;
//         this.date = date;
//         this.dataOwnerCode = dataOwnerCode;
//         this.stopCode = stopCode;
//     }
// 
//     public String getDataOwnerCode() {
//         return dataOwnerCode;
//     }
// 
//     public LocalDate getDate() {
//         return date;
//     }
// 
//     public Type getType() {
//         return type;
//     }
// 
//     public String getStopCode() {
//         return stopCode;
//     }
// 
//     public String getQuayCode() {
//         return quayCode;
//     }
// 
//     public void setQuayCode(String quayCode) {
//         this.quayCode = quayCode;
//     }
// 
//     public enum Type {
//         PASSTIME,
//         MESSAGE_UPDATE,
//         MESSAGE_DELETE,
//     }
// }
// 
// Path: distribution/src/main/java/nl/crowndov/displaydirect/distribution/stats/MetricStore.java
// public class MetricStore {
// 
//     private static List<Metric> metrics = new ArrayList<>();
//     private static ConcurrentMap<String, ConcurrentMap<LocalDateTime, Metric>> buckets = new ConcurrentHashMap<>();
//     private static MetricStore instance;
// 
//     public static synchronized MetricStore getInstance() {
//         if (instance == null) {
//             instance = new MetricStore();
//         }
//         return instance;
//     }
// 
//     public void storeMetric(String name, long value) {
//         Metric m = new Metric(name, value);
//         metrics.add(m);
//     }
// 
//     public void setBucketValue(String name, TemporalUnit bucket, long value) {
//         ensureMetricInBucket(name);
//         LocalDateTime bucketSlot = LocalDateTime.now().truncatedTo(bucket);
//         buckets.get(name).put(bucketSlot, new Metric(name, value, bucketSlot));
//     }
// 
//     public void increaseBucketValue(String name, TemporalUnit bucket) {
//         ensureMetricInBucket(name);
//         LocalDateTime key = LocalDateTime.now().truncatedTo(bucket);
//         if (!buckets.get(name).containsKey(key)) {
//             buckets.get(name).put(key, new Metric(name, 1));
//         } else {
//             Metric previous = buckets.get(name).get(key);
//             buckets.get(name).put(key, new Metric(name, previous.getValue() + 1, key));
//         }
//     }
// 
//     private void ensureMetricInBucket(String name) {
//         if (!buckets.containsKey(name)) {
//             buckets.put(name, new ConcurrentSkipListMap<>());
//         }
//     }
// 
//     public Map<String, List<Metric>> getMetrics() {
//         return Stream.concat(buckets.values().stream().flatMap(m -> m.values().stream()),  metrics.stream())
//                 .collect(Collectors.groupingBy(Metric::getName));
//     }
// 
//     public List<Metric> getMetric(String name) {
//         if (buckets.containsKey(name)) {
//             return new ArrayList<>(buckets.get(name).values());
//         }
//         return metrics.parallelStream()
//                 .filter(m -> m.getName().equalsIgnoreCase(name))
//                 .collect(Collectors.toList());
//     }
// }


import nl.crowndov.displaydirect.distribution.chb.StopStore;
import nl.crowndov.displaydirect.distribution.domain.travelinfo.RealtimeMessage;
import nl.crowndov.displaydirect.distribution.stats.MetricStore;
import java.time.temporal.ChronoUnit;
import java.util.Optional;
package nl.crowndov.displaydirect.distribution.messages.processing;



/**
 * Copyright 2017 CROW-NDOV
 *
 * This file is subject to the terms and conditions defined in file 'LICENSE.txt', which is part of this source code package.
 */
public class StopCodeProcessor implements Processor<RealtimeMessage> {

    private static MetricStore metrics = MetricStore.getInstance();

    @Override
    public RealtimeMessage process(RealtimeMessage input) {
// Path: distribution/src/main/java/nl/crowndov/displaydirect/distribution/chb/StopStore.java
// public class StopStore {
// 
//     private static final Logger LOGGER = LoggerFactory.getLogger(StopStore.class);
// 
//     private static Map<String, String> store = new HashMap<>();
//     private static Set<String> missing = new HashSet<>();
// 
//     public static Optional<String> getQuayFromCode(String dataOwner, String stopCode) {
//         String key = dataOwner+"|"+stopCode;
//         if (store.containsKey(key)) {
//             return Optional.of(store.get(key));
//         }
//         missing.add(key);
//         return Optional.empty();
//     }
// 
//     public static boolean stopExists(String quayOrStopCode) {
//         return store.values().contains(quayOrStopCode);
//     }
// 
//     public static Map<String, String> getStore() {
//         return store;
//     }
// 
//     public static Set<String> getMissing() {
//         return missing;
//     }
// 
//     public static void setStore(Map<String, String> storeData) {
//         store = storeData;
//     }
// }
// 
// Path: distribution/src/main/java/nl/crowndov/displaydirect/distribution/domain/travelinfo/RealtimeMessage.java
// public class RealtimeMessage implements Serializable {
// 
//     private static final long serialVersionUID = -8692508260190453615L;
// 
//     private Type type;
//     private LocalDate date;
//     private String dataOwnerCode;
//     private String stopCode;
// 
//     private String quayCode;
// 
//     public RealtimeMessage() {
//     }
// 
//     public RealtimeMessage(Type type, LocalDate date, String dataOwnerCode, String stopCode) {
//         this.type = type;
//         this.date = date;
//         this.dataOwnerCode = dataOwnerCode;
//         this.stopCode = stopCode;
//     }
// 
//     public String getDataOwnerCode() {
//         return dataOwnerCode;
//     }
// 
//     public LocalDate getDate() {
//         return date;
//     }
// 
//     public Type getType() {
//         return type;
//     }
// 
//     public String getStopCode() {
//         return stopCode;
//     }
// 
//     public String getQuayCode() {
//         return quayCode;
//     }
// 
//     public void setQuayCode(String quayCode) {
//         this.quayCode = quayCode;
//     }
// 
//     public enum Type {
//         PASSTIME,
//         MESSAGE_UPDATE,
//         MESSAGE_DELETE,
//     }
// }
// 
// Path: distribution/src/main/java/nl/crowndov/displaydirect/distribution/stats/MetricStore.java
// public class MetricStore {
// 
//     private static List<Metric> metrics = new ArrayList<>();
//     private static ConcurrentMap<String, ConcurrentMap<LocalDateTime, Metric>> buckets = new ConcurrentHashMap<>();
//     private static MetricStore instance;
// 
//     public static synchronized MetricStore getInstance() {
//         if (instance == null) {
//             instance = new MetricStore();
//         }
//         return instance;
//     }
// 
//     public void storeMetric(String name, long value) {
//         Metric m = new Metric(name, value);
//         metrics.add(m);
//     }
// 
//     public void setBucketValue(String name, TemporalUnit bucket, long value) {
//         ensureMetricInBucket(name);
//         LocalDateTime bucketSlot = LocalDateTime.now().truncatedTo(bucket);
//         buckets.get(name).put(bucketSlot, new Metric(name, value, bucketSlot));
//     }
// 
//     public void increaseBucketValue(String name, TemporalUnit bucket) {
//         ensureMetricInBucket(name);
//         LocalDateTime key = LocalDateTime.now().truncatedTo(bucket);
//         if (!buckets.get(name).containsKey(key)) {
//             buckets.get(name).put(key, new Metric(name, 1));
//         } else {
//             Metric previous = buckets.get(name).get(key);
//             buckets.get(name).put(key, new Metric(name, previous.getValue() + 1, key));
//         }
//     }
// 
//     private void ensureMetricInBucket(String name) {
//         if (!buckets.containsKey(name)) {
//             buckets.put(name, new ConcurrentSkipListMap<>());
//         }
//     }
// 
//     public Map<String, List<Metric>> getMetrics() {
//         return Stream.concat(buckets.values().stream().flatMap(m -> m.values().stream()),  metrics.stream())
//                 .collect(Collectors.groupingBy(Metric::getName));
//     }
// 
//     public List<Metric> getMetric(String name) {
//         if (buckets.containsKey(name)) {
//             return new ArrayList<>(buckets.get(name).values());
//         }
//         return metrics.parallelStream()
//                 .filter(m -> m.getName().equalsIgnoreCase(name))
//                 .collect(Collectors.toList());
//     }
// }


// Path: distribution/src/main/java/nl/crowndov/displaydirect/distribution/messages/processing/StopCodeProcessor.java
import nl.crowndov.displaydirect.distribution.chb.StopStore;
import nl.crowndov.displaydirect.distribution.domain.travelinfo.RealtimeMessage;
import nl.crowndov.displaydirect.distribution.stats.MetricStore;
import java.time.temporal.ChronoUnit;
import java.util.Optional;

package nl.crowndov.displaydirect.distribution.messages.processing;



/**
 * Copyright 2017 CROW-NDOV
 *
 * This file is subject to the terms and conditions defined in file 'LICENSE.txt', which is part of this source code package.
 */
public class StopCodeProcessor implements Processor<RealtimeMessage> {

    private static MetricStore metrics = MetricStore.getInstance();

    @Override
    public RealtimeMessage process(RealtimeMessage input) {
        Optional<String> quay = StopStore.getQuayFromCode(input.getDataOwnerCode(), input.getStopCode());