KafkaProducerRepository { Optional<Pair<KafkaProducer<String, String>, Pattern>> findMatchingKafkaProducer(String topic){ for (Map.Entry<Pattern, KafkaProducer<String, String>> entry : kafkaProducers.entrySet()) { Matcher matcher = entry.getKey().matcher(topic); if(matcher.matches()){ log.info("Found matching KafkaProducer with pattern '{}' for topic '{}' found", entry.getKey().pattern(), topic); return Optional.of(Pair.of(entry.getValue(), entry.getKey())); } } log.info("No matching KafkaProducer for topic '{}' found", topic); return Optional.empty(); } KafkaProducerRepository(Vertx vertx);   }
@Test public void findMatchingKafkaProducerEmptyMap(TestContext context) { context.assertFalse(repository.findMatchingKafkaProducer(null).isPresent()); context.assertFalse(repository.findMatchingKafkaProducer("my.topic.x").isPresent()); }
@Test public void findMatchingKafkaProducer(TestContext context) { Pattern pattern_1 = patternFrom("my.topic.*"); Pattern pattern_2 = patternFrom(".+"); repository.addKafkaProducer(new KafkaConfiguration(pattern_1, configs)); repository.addKafkaProducer(new KafkaConfiguration(pattern_2, configs_2)); final Optional<Pair<KafkaProducer<String, String>, Pattern>> anyTopic = repository.findMatchingKafkaProducer("anyTopic"); context.assertTrue(anyTopic.isPresent()); context.assertEquals(pattern_2.pattern(), anyTopic.get().getRight().pattern()); final Optional<Pair<KafkaProducer<String, String>, Pattern>> myTopicX = repository.findMatchingKafkaProducer("my.topic.x"); context.assertTrue(myTopicX.isPresent()); context.assertEquals(pattern_1.pattern(), myTopicX.get().getRight().pattern()); }