diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/AbstractKafkaChannel.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/AbstractKafkaChannel.java index 503fe5afe5..d397685e40 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/AbstractKafkaChannel.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/AbstractKafkaChannel.java @@ -75,7 +75,7 @@ public abstract class AbstractKafkaChannel extends AbstractMessageChannel { this.template.send(MessageBuilder.fromMessage(message) .setHeader(KafkaHeaders.TOPIC, this.topic) .build()) - .get(timeout, TimeUnit.MILLISECONDS); + .get(timeout < 0 ? Long.MAX_VALUE : timeout, TimeUnit.MILLISECONDS); } catch (@SuppressWarnings("unused") InterruptedException e) { Thread.currentThread().interrupt(); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PublishSubscribeKafkaChannel.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PublishSubscribeKafkaChannel.java new file mode 100644 index 0000000000..60c8dab9ba --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PublishSubscribeKafkaChannel.java @@ -0,0 +1,52 @@ +/* + * Copyright 2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.channel; + +import org.springframework.integration.channel.BroadcastCapableChannel; +import org.springframework.integration.dispatcher.BroadcastingDispatcher; +import org.springframework.integration.dispatcher.MessageDispatcher; +import org.springframework.kafka.config.KafkaListenerContainerFactory; +import org.springframework.kafka.core.KafkaOperations; + +/** + * Publish/subscribe channel backed by a Kafka topic. + * + * @author Gary Russell + * @since 3.3 + * + */ +public class PublishSubscribeKafkaChannel extends SubscribableKafkaChannel implements BroadcastCapableChannel { + + /** + * Construct an instance with the provided parameters. + * @param template template for sending. + * @param factory factory for creating a container for receiving. + * @param channelTopic the topic. + */ + public PublishSubscribeKafkaChannel(KafkaOperations template, KafkaListenerContainerFactory factory, + String channelTopic) { + super(template, factory, channelTopic); + } + + @Override + protected MessageDispatcher createDispatcher() { + BroadcastingDispatcher broadcastingDispatcher = new BroadcastingDispatcher(true); + broadcastingDispatcher.setBeanFactory(getBeanFactory()); + return broadcastingDispatcher; + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java index 2603caf26c..166586b936 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java @@ -21,7 +21,6 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; import org.springframework.context.Phased; import org.springframework.context.SmartLifecycle; -import org.springframework.integration.dispatcher.BroadcastingDispatcher; import org.springframework.integration.dispatcher.MessageDispatcher; import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; import org.springframework.integration.dispatcher.UnicastingDispatcher; @@ -47,8 +46,6 @@ public class SubscribableKafkaChannel extends AbstractKafkaChannel implements Su private final KafkaListenerContainerFactory factory; - private final boolean pubSub; - private MessageDispatcher dispatcher; private MessageListenerContainer container; @@ -68,23 +65,9 @@ public class SubscribableKafkaChannel extends AbstractKafkaChannel implements Su public SubscribableKafkaChannel(KafkaOperations template, KafkaListenerContainerFactory factory, String channelTopic) { - this(template, factory, channelTopic, false); - } - - /** - * Construct an instance with the provided parameters. - * @param template template for sending. - * @param factory factory for creating a container for receiving. - * @param channelTopic the topic. - * @param pubSub true for a publish/subscribe channel. - */ - public SubscribableKafkaChannel(KafkaOperations template, KafkaListenerContainerFactory factory, - String channelTopic, boolean pubSub) { - super(template, channelTopic); Assert.notNull(factory, "'factory' cannot be null"); this.factory = factory; - this.pubSub = pubSub; } @Override @@ -122,16 +105,7 @@ public class SubscribableKafkaChannel extends AbstractKafkaChannel implements Su @Override protected void onInit() { - if (this.pubSub) { - BroadcastingDispatcher broadcastingDispatcher = new BroadcastingDispatcher(true); - broadcastingDispatcher.setBeanFactory(this.getBeanFactory()); - this.dispatcher = broadcastingDispatcher; - } - else { - UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher(); - unicastingDispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy()); - this.dispatcher = unicastingDispatcher; - } + this.dispatcher = createDispatcher(); this.container = this.factory.createContainer(this.topic); String groupId = getGroupId(); this.container.getContainerProperties().setGroupId(groupId != null ? groupId : getBeanName()); @@ -149,6 +123,12 @@ public class SubscribableKafkaChannel extends AbstractKafkaChannel implements Su }); } + protected MessageDispatcher createDispatcher() { + UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher(); + unicastingDispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy()); + return unicastingDispatcher; + } + @Override public void start() { this.container.start(); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaChannelParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaChannelParser.java index cdf8eddeea..fa071f5745 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaChannelParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaChannelParser.java @@ -23,6 +23,7 @@ import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.xml.AbstractChannelParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.kafka.channel.PollableKafkaChannel; +import org.springframework.integration.kafka.channel.PublishSubscribeKafkaChannel; import org.springframework.integration.kafka.channel.SubscribableKafkaChannel; import org.springframework.util.StringUtils; @@ -46,11 +47,12 @@ public class KafkaChannelParser extends AbstractChannelParser { String topic = element.getAttribute("topic"); boolean pubSub = "publish-subscribe-channel".equals(element.getLocalName()); if (hasFactory) { - builder = BeanDefinitionBuilder.genericBeanDefinition(SubscribableKafkaChannel.class); + builder = BeanDefinitionBuilder.genericBeanDefinition(pubSub + ? PublishSubscribeKafkaChannel.class + : SubscribableKafkaChannel.class); builder.addConstructorArgReference(template); builder.addConstructorArgReference(factory); builder.addConstructorArgValue(topic); - builder.addConstructorArgValue(pubSub); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "role"); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/AbstractKafkaChannelSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/AbstractKafkaChannelSpec.java index 2a5fe2a240..2684dcdfa9 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/AbstractKafkaChannelSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/AbstractKafkaChannelSpec.java @@ -18,32 +18,23 @@ package org.springframework.integration.kafka.dsl; import org.springframework.integration.dsl.MessageChannelSpec; import org.springframework.integration.kafka.channel.AbstractKafkaChannel; -import org.springframework.kafka.core.KafkaTemplate; /** * * Spec for a message channel backed by a Kafka topic. * * @param the spec type. + * @param the channel type. * * @author Gary Russell * @since 3.3 * */ -public abstract class AbstractKafkaChannelSpec> - extends MessageChannelSpec { - - protected final KafkaTemplate template; // NOSONAR final - - protected final String topic; // NOSONAR final +public abstract class AbstractKafkaChannelSpec, C extends AbstractKafkaChannel> + extends MessageChannelSpec { protected String groupId; // NOSONAR - protected AbstractKafkaChannelSpec(KafkaTemplate template, String topic) { - this.template = template; - this.topic = topic; - } - @Override public S id(String idToSet) { // NOSONAR - increase visibility return super.id(idToSet); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java index 5d8680b6e6..b03aefa79b 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java @@ -545,10 +545,10 @@ public final class Kafka { * @return the spec. * @since 3.3 */ - public static KafkaSubscribableChannelSpec channel(KafkaTemplate template, + public static KafkaPointToPointChannelSpec channel(KafkaTemplate template, KafkaListenerContainerFactory containerFactory, String topic) { - return new KafkaSubscribableChannelSpec(template, containerFactory, topic, false); + return new KafkaPointToPointChannelSpec(template, containerFactory, topic); } /** @@ -559,10 +559,10 @@ public final class Kafka { * @return the spec. * @since 3.3 */ - public static KafkaSubscribableChannelSpec publishSubscribeChannel(KafkaTemplate template, + public static KafkaPublishSubscribeChannelSpec publishSubscribeChannel(KafkaTemplate template, KafkaListenerContainerFactory containerFactory, String topic) { - return new KafkaSubscribableChannelSpec(template, containerFactory, topic, true); + return new KafkaPublishSubscribeChannelSpec(template, containerFactory, topic); } /** diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPointToPointChannelSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPointToPointChannelSpec.java new file mode 100644 index 0000000000..65856b87f8 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPointToPointChannelSpec.java @@ -0,0 +1,38 @@ +/* + * Copyright 2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.dsl; + +import org.springframework.integration.kafka.channel.SubscribableKafkaChannel; +import org.springframework.kafka.config.KafkaListenerContainerFactory; +import org.springframework.kafka.core.KafkaTemplate; + +/** + * Spec for a point to point channel backed by a Kafka topic. + * + * @author Gary Russell + * @since 3.3 + * + */ +public class KafkaPointToPointChannelSpec extends KafkaSubscribableChannelSpec { + + protected KafkaPointToPointChannelSpec(KafkaTemplate template, KafkaListenerContainerFactory factory, + String topic) { + + this.channel = new SubscribableKafkaChannel(template, factory, topic); + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPollableChannelSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPollableChannelSpec.java index 09f8f6b500..18ddd3d0ec 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPollableChannelSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPollableChannelSpec.java @@ -27,10 +27,9 @@ import org.springframework.kafka.core.KafkaTemplate; * @since 3.3 * */ -public class KafkaPollableChannelSpec extends AbstractKafkaChannelSpec { +public class KafkaPollableChannelSpec extends AbstractKafkaChannelSpec { protected KafkaPollableChannelSpec(KafkaTemplate template, KafkaMessageSource source) { - super(null, null); this.channel = new PollableKafkaChannel(template, source); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPublishSubscribeChannelSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPublishSubscribeChannelSpec.java new file mode 100644 index 0000000000..3af04029d0 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPublishSubscribeChannelSpec.java @@ -0,0 +1,39 @@ +/* + * Copyright 2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.dsl; + +import org.springframework.integration.kafka.channel.PublishSubscribeKafkaChannel; +import org.springframework.kafka.config.KafkaListenerContainerFactory; +import org.springframework.kafka.core.KafkaTemplate; + +/** + * Spec for a publish/subscribe channel backed by a Kafka topic. + * + * @author Gary Russell + * @since 3.3 + * + */ +public class KafkaPublishSubscribeChannelSpec + extends KafkaSubscribableChannelSpec { + + protected KafkaPublishSubscribeChannelSpec(KafkaTemplate template, KafkaListenerContainerFactory factory, + String topic) { + + this.channel = new PublishSubscribeKafkaChannel(template, factory, topic); + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaSubscribableChannelSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaSubscribableChannelSpec.java index a33bc2ac92..7d4cf11528 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaSubscribableChannelSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaSubscribableChannelSpec.java @@ -17,23 +17,17 @@ package org.springframework.integration.kafka.dsl; import org.springframework.integration.kafka.channel.SubscribableKafkaChannel; -import org.springframework.kafka.config.KafkaListenerContainerFactory; -import org.springframework.kafka.core.KafkaTemplate; /** * Spec for a subscribable channel. * + * @param the channel type. + * * @author Gary Russell * @since 3.3 * */ -public class KafkaSubscribableChannelSpec extends AbstractKafkaChannelSpec { - - protected KafkaSubscribableChannelSpec(KafkaTemplate template, KafkaListenerContainerFactory factory, - String topic, boolean pubSub) { - - super(template, topic); - this.channel = new SubscribableKafkaChannel(template, factory, topic, pubSub); - } +public abstract class KafkaSubscribableChannelSpec + extends AbstractKafkaChannelSpec, C> { } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java index 2d3cabe777..c20c1c22ba 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java @@ -30,6 +30,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.integration.kafka.channel.PollableKafkaChannel; +import org.springframework.integration.kafka.channel.PublishSubscribeKafkaChannel; import org.springframework.integration.kafka.channel.SubscribableKafkaChannel; import org.springframework.integration.kafka.inbound.KafkaMessageSource; import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; @@ -146,7 +147,7 @@ public class ChannelTests { public SubscribableKafkaChannel pubSub(KafkaTemplate template, KafkaListenerContainerFactory factory) { - SubscribableKafkaChannel channel = new SubscribableKafkaChannel(template, factory, "channel.2", true); + SubscribableKafkaChannel channel = new PublishSubscribeKafkaChannel(template, factory, "channel.2"); channel.setGroupId("channel.2"); return channel; } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java index b5f1a548b8..e3ab95f6a1 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java @@ -38,6 +38,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.BroadcastCapableChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.dsl.IntegrationFlow; @@ -94,7 +95,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; @DirtiesContext @EmbeddedKafka(topics = { KafkaDslTests.TEST_TOPIC1, KafkaDslTests.TEST_TOPIC2, KafkaDslTests.TEST_TOPIC3, KafkaDslTests.TEST_TOPIC4, KafkaDslTests.TEST_TOPIC5, KafkaDslTests.TEST_TOPIC6, KafkaDslTests.TEST_TOPIC7, - KafkaDslTests.TEST_TOPIC8 }) + KafkaDslTests.TEST_TOPIC8, KafkaDslTests.TEST_TOPIC9 }) public class KafkaDslTests { static final String TEST_TOPIC1 = "test-topic1"; @@ -113,6 +114,8 @@ public class KafkaDslTests { static final String TEST_TOPIC8 = "test-topic8"; + static final String TEST_TOPIC9 = "test-topic9"; + @Autowired @Qualifier("sendToKafkaFlow.input") private MessageChannel sendToKafkaFlowInput; @@ -156,7 +159,7 @@ public class KafkaDslTests { private Gate gate; @Test - public void testKafkaAdapters() throws Exception { + void testKafkaAdapters() throws Exception { this.sendToKafkaFlowInput.send(new GenericMessage<>("foo", Collections.singletonMap("foo", "bar"))); assertThat(TestUtils.getPropertyValue(this.kafkaProducer1, "headerMapper")).isSameAs(this.mapper); @@ -217,20 +220,25 @@ public class KafkaDslTests { } @Test - public void testGateways() throws Exception { + void testGateways() throws Exception { assertThat(this.config.replyContainerLatch.await(30, TimeUnit.SECONDS)).isTrue(); assertThat(this.gate.exchange(TEST_TOPIC4, "foo")).isEqualTo("FOO"); } @Test - void channels(@Autowired MessageChannel topic6Channel, @Autowired PollableKafkaChannel topic8Channel) { + void channels(@Autowired MessageChannel topic6Channel, @Autowired PollableKafkaChannel topic8Channel, + @Autowired PollableKafkaChannel topic9Channel) { topic6Channel.send(new GenericMessage<>("foo")); Message received = topic8Channel.receive(); assertThat(received) .isNotNull() .extracting("payload") .isEqualTo("foo"); - } + received = topic9Channel.receive(); + assertThat(received) + .isNotNull() + .extracting("payload") + .isEqualTo("foo"); } @Configuration @EnableIntegration @@ -379,8 +387,9 @@ public class KafkaDslTests { } @Bean - public KafkaSubscribableChannelSpec topic6Channel(KafkaTemplate template, + public KafkaPointToPointChannelSpec topic6Channel(KafkaTemplate template, ConcurrentKafkaListenerContainerFactory containerFactory) { + return Kafka.channel(template, containerFactory, TEST_TOPIC6); } @@ -400,8 +409,19 @@ public class KafkaDslTests { KafkaMessageSource channelSource) { return IntegrationFlows.from(topic6Channel(template, containerFactory)) - .channel(Kafka.publishSubscribeChannel(template, containerFactory, TEST_TOPIC7)) - .channel(Kafka.pollableChannel(template, channelSource).id("topic8Channel")) + .publishSubscribeChannel(pubSub(template, containerFactory), channel -> channel + .subscribe(f -> f.channel( + Kafka.pollableChannel(template, channelSource).id("topic8Channel"))) + .subscribe(f -> f.channel( + Kafka.pollableChannel(template, channelSource).id("topic9Channel")))) + .get(); + } + + @Bean + public BroadcastCapableChannel pubSub(KafkaTemplate template, + ConcurrentKafkaListenerContainerFactory containerFactory) { + + return Kafka.publishSubscribeChannel(template, containerFactory, TEST_TOPIC7) .get(); }