package rxbroadcast;

import rxbroadcast.time.LamportClock;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;

public final class CausalOrder<T> implements BroadcastOrder<VectorTimestamped<T>, T> {
    private final class DelayQueueEntry {
        final Sender sender;

        final VectorTimestamped<T> message;

        private DelayQueueEntry(final Sender sender, final VectorTimestamped<T> message) {
            this.sender = sender;
            this.message = message;
        }
    }

    private final Sender me;

    private final Map<Sender, LamportClock> vt = new HashMap<>();

    private final List<DelayQueueEntry> delayQueue = new ArrayList<>();

    private long myNextIncomingTimestamp = 1L;

    public CausalOrder(final Sender me) {
        this.me = me;
        vt.put(me, new LamportClock());
    }

    @Override
    public VectorTimestamped<T> prepare(final T value) {
        return vt.get(me).tick((time) -> {
            final List<Sender> senders = new ArrayList<>(vt.keySet());
            final Sender[] ids = new Sender[senders.size()];
            final long[] timestamps = new long[senders.size()];

            for (int i = 0; i < senders.size(); i++) {
                final Sender s = senders.get(i);
                ids[i] = s;
                timestamps[i] = vt.get(s).time();
            }

            return new VectorTimestamped<>(value, new VectorTimestamp(ids, timestamps));
        });
    }

    @Override
    public void receive(final Sender sender, final Consumer<T> consumer, final VectorTimestamped<T> message) {
        final Map<Sender, Long> messageTimestamp = message.timestamp.asMap();
        messageTimestamp.keySet().forEach(k1 -> vt.putIfAbsent(k1, new LamportClock()));
        delayQueue.add(new DelayQueueEntry(sender, message));

        boolean wasMutated;
        do {
            wasMutated = false;
            final Iterator<DelayQueueEntry> iterator = delayQueue.iterator();
            while (iterator.hasNext()) {
                final DelayQueueEntry entry = iterator.next();
                final Map<Sender, Long> entryTimestamp = entry.message.timestamp.asMap();
                if (entry.sender.equals(me) && entryTimestamp.get(me) == myNextIncomingTimestamp) {
                    consumer.accept(entry.message.value);
                    myNextIncomingTimestamp++;
                    iterator.remove();
                    wasMutated = true;
                }
                if (shouldBeDelivered(entry.sender, entry.message)) {
                    consumer.accept(entry.message.value);
                    vt.forEach((k, v) -> v.set(Math.max(v.time(), entryTimestamp.getOrDefault(k, 0L))));
                    iterator.remove();
                    wasMutated = true;
                }
            }
        } while (wasMutated);
    }

    int delayQueueSize() {
        return delayQueue.size();
    }

    @SuppressWarnings("checkstyle:AvoidInlineConditionals")
    private boolean shouldBeDelivered(final Sender i, final VectorTimestamped<T> message) {
        final Map<Sender, Long> timestamp = message.timestamp.asMap();
        return timestamp.keySet().stream().allMatch((k) ->
            k.equals(i)
                ? timestamp.get(k) == (vt.get(k).time() + 1)
                : timestamp.get(k) <= (vt.get(k).time()));
    }
}

@@ -1064,94 +1064,43 @@
 nal 
-List%3C
 Sender
-%3E senders = new ArrayList%3C%3E(vt.keySet());%0A            final Sender%5B%5D ids = 
+%5B%5D ids = vt.keySet().toArray(
 new 
@@ -1106,31 +1106,26 @@
  Sender%5B
-senders
+vt
 .size()%5D
 ;%0A      
@@ -1116,16 +1116,17 @@
 .size()%5D
+)
 ;%0A      
@@ -1170,26 +1170,21 @@
 ong%5B
-senders.size()
+ids.length
 %5D;%0A
-%0A
     
@@ -1215,22 +1215,18 @@
 i %3C 
-senders.size()
+ids.length
 ; i+
@@ -1234,85 +1234,8 @@
 ) %7B%0A
-                final Sender s = senders.get(i);%0A                ids%5Bi%5D = s;%0A
     
@@ -1269,17 +1269,22 @@
  vt.get(
-s
+ids%5Bi%5D
 ).time()

Use collections toArray where possible