Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix Kotlin tests according upstream changes

Browse Source 
* Upgrade dependencies
This commit is contained in:
Artem Bilan
2020-04-02 14:44:12 -04:00
committed by Artem Bilan
parent 213772dcaf
commit bfef5a49cf
1 changed files with 13 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt
View File

@@ -34,6 +34,7 @@ import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.integration.IntegrationMessageHeaderAccessor
import org.springframework.integration.MessageRejectedException
import org.springframework.integration.channel.PublishSubscribeChannel
import org.springframework.integration.channel.QueueChannel
import org.springframework.integration.config.EnableIntegration
import org.springframework.integration.dsl.Pollers
@@ -283,18 +284,18 @@ class KafkaDslKotlinTests {
fun sendToKafkaFlow() =
integrationFlow {
split<String> { p -> Stream.generate { p }.limit(101) }
publishSubscribeChannel {
subscribe(integrationFlow {
handle(kafkaMessageHandler(producerFactory(), TEST_TOPIC1)
.timestampExpression("T(Long).valueOf('1487694048633')")
) { id("kafkaProducer1") }
})
subscribe(integrationFlow {
handle(kafkaMessageHandler(producerFactory(), TEST_TOPIC2)
.timestamp<Any> { 1487694048644L }
) { id("kafkaProducer2") }
})
}
publishSubscribe(PublishSubscribeChannel(),
{
handle(kafkaMessageHandler(producerFactory(), TEST_TOPIC1)
.timestampExpression("T(Long).valueOf('1487694048633')")
) { id("kafkaProducer1") }
},
{
handle(kafkaMessageHandler(producerFactory(), TEST_TOPIC2)
.timestamp<Any> { 1487694048644L }
) { id("kafkaProducer2") }
}
)
}
@Bean