spring-kafka-quickstart (0.0.1)

brew install node npm i -g yo generator-jvm yo jvm -n spring-kafka-quickstart -t java-spring-boot idea spring-kafka-quickstart/pom.xml

spring cloud kafka

 <dependencies> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka-test</artifactId> <scope>test</scope> </dependency> </dependencies>

dependencies { implementation('org.springframework.kafka:spring-kafka') testImplementation('org.springframework.kafka:spring-kafka-test') }

@Log4j2 @Service public class MessageListener { @KafkaListener(topics = "messages") public void on(final ConsumerRecord<Object, Object> message) { log.info("received message: {}", message.value()); Mono.just(new Message().setAt(now()) .setBody(message.value().toString())) .subscribe(msg -> log.info("saving message object: '{}'", msg)); } }

@Configuration @RequiredArgsConstructor class WebfluxRoutesConfig { static final ParameterizedTypeReference<Map<String, String>> sendMessageRequestType = new ParameterizedTypeReference<Map<String, String>>() {}; final KafkaTemplate<Object, Object> kafka; @Bean HandlerFunction<ServerResponse> sendMessageHandler() { return request -> ok().body(request.bodyToMono(sendMessageRequestType) .map(it -> it.getOrDefault("message", "")) .filter(it -> !it.trim().isEmpty()) .doOnNext(message -> kafka.send("messages", message)) .map(s -> "message sent.") .flatMap(Mono::just), Object.class); } @Bean RouterFunction routes(final HandlerFunction<ServerResponse> fallbackHandler) { return route( POST("/"), sendMessageHandler() ).andOther( route( GET("/**"), fallbackHandler ) ) ; } }

./gradlew bash -jar build/libs/*.jar http :8080 message=ololo http :8080 message=trololo

./mvnw bash target/*.jar