diff --git a/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt b/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt index da017e70f3..9d50d232d1 100644 --- a/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt +++ b/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt @@ -37,10 +37,8 @@ import org.springframework.integration.MessageRejectedException import org.springframework.integration.channel.QueueChannel import org.springframework.integration.config.EnableIntegration import org.springframework.integration.dsl.IntegrationFlow +import org.springframework.integration.dsl.IntegrationFlows import org.springframework.integration.dsl.Pollers -import org.springframework.integration.dsl.kotlin.filterReified -import org.springframework.integration.dsl.kotlin.integrationFlow -import org.springframework.integration.dsl.kotlin.split import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler @@ -89,7 +87,7 @@ import java.util.stream.Stream @SpringJUnitConfig @DirtiesContext @EmbeddedKafka(topics = [KafkaDslKotlinTests.TEST_TOPIC1, KafkaDslKotlinTests.TEST_TOPIC2, - KafkaDslKotlinTests.TEST_TOPIC3, KafkaDslKotlinTests.TEST_TOPIC4, KafkaDslKotlinTests.TEST_TOPIC5]) + KafkaDslKotlinTests.TEST_TOPIC3, KafkaDslKotlinTests.TEST_TOPIC4, KafkaDslKotlinTests.TEST_TOPIC5]) class KafkaDslKotlinTests { companion object { @@ -161,8 +159,8 @@ class KafkaDslKotlinTests { assertThat(receive!!.payload).isEqualTo("FOO") val headers = receive.headers assertThat(headers.containsKey(KafkaHeaders.ACKNOWLEDGMENT)).isTrue() - val acknowledgment = headers[KafkaHeaders.ACKNOWLEDGMENT] as Acknowledgment - acknowledgment.acknowledge() + val acknowledgment = headers.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment::class.java) + acknowledgment?.acknowledge() assertThat(headers[KafkaHeaders.RECEIVED_TOPIC]).isEqualTo(TEST_TOPIC1) assertThat(headers[KafkaHeaders.RECEIVED_MESSAGE_KEY]).isEqualTo(i + 1) assertThat(headers[KafkaHeaders.RECEIVED_PARTITION_ID]).isEqualTo(0) @@ -178,8 +176,8 @@ class KafkaDslKotlinTests { assertThat(receive!!.payload).isEqualTo("FOO") val headers = receive.headers assertThat(headers.containsKey(KafkaHeaders.ACKNOWLEDGMENT)).isTrue() - val acknowledgment = headers[KafkaHeaders.ACKNOWLEDGMENT] as Acknowledgment - acknowledgment.acknowledge() + val acknowledgment = headers.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment::class.java) + acknowledgment?.acknowledge() assertThat(headers[KafkaHeaders.RECEIVED_TOPIC]).isEqualTo(TEST_TOPIC2) assertThat(headers[KafkaHeaders.RECEIVED_MESSAGE_KEY]).isEqualTo(i + 1) assertThat(headers[KafkaHeaders.RECEIVED_PARTITION_ID]).isEqualTo(0) @@ -234,7 +232,7 @@ class KafkaDslKotlinTests { @Bean fun consumerFactory(): ConsumerFactory { val props = KafkaTestUtils.consumerProps("test1", "false", this.embeddedKafka) - props[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = "earliest" + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") return DefaultKafkaConsumerFactory(props) } @@ -243,7 +241,7 @@ class KafkaDslKotlinTests { @Bean fun topic1ListenerFromKafkaFlow() = - integrationFlow( + IntegrationFlows.from( Kafka.messageDrivenChannelAdapter(consumerFactory(), KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC1) .configureListenerContainer { @@ -253,52 +251,50 @@ class KafkaDslKotlinTests { .recoveryCallback(ErrorMessageSendingRecoverer(errorChannel(), RawRecordHeaderErrorMessageStrategy())) .retryTemplate(RetryTemplate()) - .filterInRetry(true)) { - it.filterReified>( - { m -> (m.headers[KafkaHeaders.RECEIVED_MESSAGE_KEY] as Int) < 101 }, - { f -> f.throwExceptionOnRejection(true) }) - .transform { it.toUpperCase() } - .channel { c -> c.queue("listeningFromKafkaResults1") } - } + .filterInRetry(true)) + .filter(Message::class.java, { m -> m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer::class.java)!! < 101 }, + { f -> f.throwExceptionOnRejection(true) }) + .transform { it.toUpperCase() } + .channel { c -> c.queue("listeningFromKafkaResults1") } + .get() @Bean fun topic2ListenerFromKafkaFlow() = - integrationFlow( + IntegrationFlows.from( Kafka.messageDrivenChannelAdapter(consumerFactory(), KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC2) .configureListenerContainer { it.ackMode(ContainerProperties.AckMode.MANUAL) } .recoveryCallback(ErrorMessageSendingRecoverer(errorChannel(), RawRecordHeaderErrorMessageStrategy())) .retryTemplate(RetryTemplate()) - .filterInRetry(true)) { - it.filterReified>( - { m -> (m.headers[KafkaHeaders.RECEIVED_MESSAGE_KEY] as Int) < 101 }, - { it.throwExceptionOnRejection(true) }) - .transform { it.toUpperCase() } - .channel { c -> c.queue("listeningFromKafkaResults2") } - } + .filterInRetry(true)) + .filter(Message::class.java, + { m -> m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer::class.java)!! < 101 }, + { it.throwExceptionOnRejection(true) }) + .transform { it.toUpperCase() } + .channel { c -> c.queue("listeningFromKafkaResults2") } + .get() @Bean fun producerFactory(): DefaultKafkaProducerFactory { val props = KafkaTestUtils.producerProps(this.embeddedKafka) - props[ProducerConfig.MAX_BLOCK_MS_CONFIG] = "10000" + props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "10000") return DefaultKafkaProducerFactory(props) } @Bean fun sendToKafkaFlow() = - IntegrationFlow { - it.split({ p -> Stream.generate { p }.limit(101) }) - .publishSubscribeChannel { - it - .subscribe { - it.handle( - kafkaMessageHandler(producerFactory(), TEST_TOPIC1) - .timestampExpression("T(Long).valueOf('1487694048633')") - ) { it.id("kafkaProducer1") } - } - .subscribe { - it.handle( + IntegrationFlow { f -> + f.split({ p -> Stream.generate { p }.limit(101) }, null) + .publishSubscribeChannel { c -> + c.subscribe { sf -> + sf.handle( + kafkaMessageHandler(producerFactory(), TEST_TOPIC1) + .timestampExpression("T(Long).valueOf('1487694048633')") + ) { it.id("kafkaProducer1") } + } + .subscribe { sf -> + sf.handle( kafkaMessageHandler(producerFactory(), TEST_TOPIC2) .timestamp { 1487694048644L } ) { it.id("kafkaProducer2") } @@ -314,20 +310,21 @@ class KafkaDslKotlinTests { .messageKey { m -> m.headers[IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER] } .headerMapper(mapper()) .sync(true) - .partitionId { 0 } + .partitionId { _ -> 0 } .topicExpression("headers[kafka_topic] ?: '$topic'") - .configureKafkaTemplate { it.id("kafkaTemplate:$topic") } + .configureKafkaTemplate { t -> t.id("kafkaTemplate:$topic") } @Bean fun sourceFlow() = - integrationFlow(Kafka.inboundChannelAdapter(consumerFactory(), ConsumerProperties(TEST_TOPIC3)), - { e -> e.poller(Pollers.fixedDelay(100)) }) { - it.handle { m -> - this.fromSource = m.payload - this.sourceFlowLatch.countDown() - } - } + IntegrationFlows + .from(Kafka.inboundChannelAdapter(consumerFactory(), ConsumerProperties(TEST_TOPIC3))) + { e -> e.poller(Pollers.fixedDelay(100)) } + .handle { p -> + this.fromSource = p.getPayload() + this.sourceFlowLatch.countDown() + } + .get() @Bean fun replyingKafkaTemplate() = @@ -338,10 +335,10 @@ class KafkaDslKotlinTests { @Bean fun outboundGateFlow() = - integrationFlow { - it.handle(Kafka.outboundGateway(replyingKafkaTemplate()) - .sync(true)) - } + IntegrationFlows.from(Gate::class.java) + .handle(Kafka.outboundGateway(replyingKafkaTemplate()) + .sync(true)) + .get() private fun replyContainer(): GenericMessageListenerContainer { val containerProperties = ContainerProperties(TEST_TOPIC5) @@ -362,9 +359,10 @@ class KafkaDslKotlinTests { @Bean fun serverGateway() = - integrationFlow(Kafka.inboundGateway(consumerFactory(), containerProperties(), producerFactory())) { - it.transform { it.toUpperCase() } - } + IntegrationFlows.from( + Kafka.inboundGateway(consumerFactory(), containerProperties(), producerFactory())) + .transform { it.toUpperCase() } + .get() private fun containerProperties() = ContainerProperties(TEST_TOPIC4)