X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationTests.java;fp=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationTests.java;h=1a336d684992afaf659fae096445456f67599704;hb=55d4ad6e7ccbd5d284d7bcda7cfc857555fcd0a5;hp=9dda0798d6ab70da75ae16f5d36633ad6f423a87;hpb=a4be0eac8d9f234cd53ae1917a084f65d91d0460;p=demos%2Fkafka%2Ftraining diff --git a/src/test/java/de/juplo/kafka/ApplicationTests.java b/src/test/java/de/juplo/kafka/ApplicationTests.java index 9dda079..1a336d6 100644 --- a/src/test/java/de/juplo/kafka/ApplicationTests.java +++ b/src/test/java/de/juplo/kafka/ApplicationTests.java @@ -12,71 +12,72 @@ public class ApplicationTests extends GenericApplicationTests { public ApplicationTests() { - super( - new RecordGenerator() + super(new ApplicationTestRecrodGenerator()); + } + + + static class ApplicationTestRecrodGenerator implements RecordGenerator + { + final int[] numbers = {1, 7, 3, 2, 33, 6, 11}; + final String[] dieWilden13 = + IntStream + .range(1, 14) + .mapToObj(i -> "seeräuber-" + i) + .toArray(i -> new String[i]); + final StringSerializer stringSerializer = new StringSerializer(); + final Bytes calculateMessage = new Bytes(stringSerializer.serialize(TOPIC, "CALCULATE")); + + int counter = 0; + + @Override + public int generate( + boolean poisonPills, + boolean logicErrors, + Consumer> messageSender) + { + counter = 0; + + for (int i = 0; i < 33; i++) + { + String seeräuber = dieWilden13[i % 13]; + int number = numbers[i % 7]; + + Bytes key = new Bytes(stringSerializer.serialize(TOPIC, seeräuber)); + + for (int message = 1; message <= number; message++) { - final int[] numbers = { 1, 7, 3, 2, 33, 6, 11 }; - final String[] dieWilden13 = - IntStream - .range(1,14) - .mapToObj(i -> "seeräuber-" + i) - .toArray(i -> new String[i]); - final StringSerializer stringSerializer = new StringSerializer(); - final Bytes calculateMessage = new Bytes(stringSerializer.serialize(TOPIC, "CALCULATE")); - - int counter = 0; - - - @Override - public int generate( - boolean poisonPills, - boolean logicErrors, - Consumer> messageSender) - { - counter = 0; - - for (int i = 0; i < 33; i++) - { - String seeräuber = dieWilden13[i%13]; - int number = numbers[i%7]; - - Bytes key = new Bytes(stringSerializer.serialize(TOPIC, seeräuber)); - - for (int message = 1; message <= number; message++) - { - Bytes value = new Bytes(stringSerializer.serialize(TOPIC, Integer.toString(message))); - send(key, value, logicErrors, messageSender); - } - send(key, calculateMessage, logicErrors, messageSender); - } - - return counter; - } - - void send( - Bytes key, - Bytes value, - boolean logicErrors, - Consumer> messageSender) - { - counter++; - - if (counter == 77) - { - if (logicErrors) - { - value = new Bytes(stringSerializer.serialize(TOPIC, Integer.toString(-1))); - } - } - - messageSender.accept(new ProducerRecord<>(TOPIC, key, value)); - } - - @Override - public boolean canGeneratePoisonPill() - { - return false; - } - }); + Bytes value = new Bytes(stringSerializer.serialize(TOPIC, Integer.toString(message))); + send(key, value, logicErrors, messageSender); + } + send(key, calculateMessage, logicErrors, messageSender); + } + + return counter; + } + + void send( + Bytes key, + Bytes value, + boolean logicErrors, + Consumer> messageSender) + { + counter++; + + if (counter == 77) + { + if (logicErrors) + { + value = new Bytes(stringSerializer.serialize(TOPIC, Integer.toString(-1))); + } + } + + messageSender.accept(new ProducerRecord<>(TOPIC, key, value)); + } + + @Override + public boolean canGeneratePoisonPill() + { + return false; + } } }