- Bytes value = new Bytes(stringSerializer.serialize(TOPIC, Integer.toString(message)));
- send(key, value, logicErrors, messageSender);
+ String seeräuber = dieWilden13[i];
+ Bytes key = new Bytes(stringSerializer.serialize(TOPIC, seeräuber));
+
+ if (message[i] > number[i])
+ {
+ send(
+ key,
+ calculateMessage,
+ Message.Type.CALC,
+ poisonPill(poisonPills, pass, counterMessages),
+ logicError(logicErrors, pass, counterMessages),
+ messageSender);
+ state.get(seeräuber).add(new AdderResult(number[i], (number[i] + 1) * number[i] / 2));
+ // Pick next number to calculate
+ number[i] = numbers[next++%numbers.length];
+ message[i] = 1;
+ log.debug("Seeräuber {} will die Summe für {} berechnen", seeräuber, number[i]);
+ }
+
+ send(
+ key,
+ new Bytes(stringSerializer.serialize(TOPIC, "{\"next\":" + message[i]++ + "}")),
+ Message.Type.ADD,
+ poisonPill(poisonPills, pass, counterMessages),
+ logicError(logicErrors, pass, counterMessages),
+ messageSender);