import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
public record PnCounter(
Map<ReplicaId, Long> positive,
Map<ReplicaId, Long> negative) {
public PnCounter {
positive = immutableNonNegative(positive);
negative = immutableNonNegative(negative);
}
public static PnCounter empty() {
return new PnCounter(Map.of(), Map.of());
}
public PnCounter apply(CounterCommand command) {
Objects.requireNonNull(command);
return switch (command) {
case CounterCommand.Increment inc ->
new PnCounter(bump(positive, inc.replica()), negative);
case CounterCommand.Decrement dec ->
new PnCounter(positive, bump(negative, dec.replica()));
};
}
public PnCounter merge(PnCounter other) {
Objects.requireNonNull(other);
return new PnCounter(
joinByMaximum(positive, other.positive),
joinByMaximum(negative, other.negative));
}
public long value() {
return Math.subtractExact(sum(positive), sum(negative));
}
private static Map<ReplicaId, Long> bump(
Map<ReplicaId, Long> current,
ReplicaId replica) {
var next = new HashMap<>(current);
next.merge(replica, 1L, Math::addExact);
return Map.copyOf(next);
}
private static Map<ReplicaId, Long> joinByMaximum(
Map<ReplicaId, Long> left,
Map<ReplicaId, Long> right) {
var merged = new HashMap<>(left);
right.forEach((replica, count) ->
merged.merge(replica, count, Math::max));
return Map.copyOf(merged);
}
private static long sum(Map<ReplicaId, Long> components) {
return components.values().stream()
.mapToLong(Long::longValue)
.reduce(0L, Math::addExact);
}
private static Map<ReplicaId, Long> immutableNonNegative(
Map<ReplicaId, Long> source) {
Objects.requireNonNull(source);
Map<ReplicaId, Long> copy = Map.copyOf(source);
copy.forEach((replica, count) -> {
Objects.requireNonNull(replica);
if (count < 0) {
throw new IllegalArgumentException("Components must be non-negative");
}
});
return copy;
}
}
