// 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.stats.MetricStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zeromq.ZMQ;
import org.zeromq.ZMQException;
import java.time.temporal.ChronoUnit;
import java.util.concurrent.atomic.AtomicBoolean;
package nl.crowndov.displaydirect.distribution.input;



public class Kv78ProcessTask implements Runnable {

    private static final Logger LOGGER = LoggerFactory.getLogger(Kv78ProcessTask.class);
    private final int port;

    private ZMQ.Context processContext = ZMQ.context(1);
    private ZMQ.Socket pull = processContext.socket(ZMQ.PULL);

// 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/input/Kv78ProcessTask.java
import nl.crowndov.displaydirect.distribution.stats.MetricStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zeromq.ZMQ;
import org.zeromq.ZMQException;
import java.time.temporal.ChronoUnit;
import java.util.concurrent.atomic.AtomicBoolean;

package nl.crowndov.displaydirect.distribution.input;



public class Kv78ProcessTask implements Runnable {

    private static final Logger LOGGER = LoggerFactory.getLogger(Kv78ProcessTask.class);
    private final int port;

    private ZMQ.Context processContext = ZMQ.context(1);
    private ZMQ.Socket pull = processContext.socket(ZMQ.PULL);

    private MetricStore metrics = MetricStore.getInstance();