GH-296: DSL: Support inline pub/sub subscriptions

Resolves #296

* * Add generic type to specsto avoid cast to `BroadcastCapableChannel`.

* * Remove unneeded fields in the abstract spec

* Restore log4j config

Author: garyrussell

src/main/java/org/springframework/integration/kafka/channel/AbstractKafkaChannel.java

@@ -75,7 +75,7 @@ protected boolean doSend(Message<?> message, long timeout) {
 			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();

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;
+	}
+
+}

src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java

@@ -21,7 +21,6 @@
 
 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 boolean isAutoStartup() {
 
 	@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 void onMessage(ConsumerRecord<Object, Object> record, Acknowledgment ackn
 		});
 	}
 
+	protected MessageDispatcher createDispatcher() {
+		UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher();
+		unicastingDispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy());
+		return unicastingDispatcher;
+	}
+
 	@Override
 	public void start() {
 		this.container.start();

src/main/java/org/springframework/integration/kafka/config/xml/KafkaChannelParser.java

@@ -23,6 +23,7 @@
 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 @@ protected BeanDefinitionBuilder buildBeanDefinition(Element element, ParserConte
 		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");

src/main/java/org/springframework/integration/kafka/dsl/AbstractKafkaChannelSpec.java

@@ -18,32 +18,23 @@
 
 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 <S> the spec type.
+ * @param <C> the channel type.
  *
  * @author Gary Russell
  * @since 3.3
  *
  */
-public abstract class AbstractKafkaChannelSpec<S extends AbstractKafkaChannelSpec<S>>
-		extends MessageChannelSpec<S, AbstractKafkaChannel> {
-
-	protected final KafkaTemplate<?, ?> template; // NOSONAR final
-
-	protected final String topic; // NOSONAR final
+public abstract class AbstractKafkaChannelSpec<S extends AbstractKafkaChannelSpec<S, C>, C extends AbstractKafkaChannel>
+		extends MessageChannelSpec<S, C> {
 
 	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);

src/main/java/org/springframework/integration/kafka/dsl/Kafka.java

@@ -545,10 +545,10 @@ public static <K, V, R> KafkaInboundGatewaySpec.KafkaInboundGatewayListenerConta
 	 * @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 static KafkaSubscribableChannelSpec channel(KafkaTemplate<?, ?> template,
 	 * @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);
 	}
 
 	/**

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<SubscribableKafkaChannel> {
+
+	protected KafkaPointToPointChannelSpec(KafkaTemplate<?, ?> template, KafkaListenerContainerFactory<?> factory,
+			String topic) {
+
+		this.channel = new SubscribableKafkaChannel(template, factory, topic);
+	}
+
+}

src/main/java/org/springframework/integration/kafka/dsl/KafkaPollableChannelSpec.java

@@ -27,10 +27,9 @@
  * @since 3.3
  *
  */
-public class KafkaPollableChannelSpec extends AbstractKafkaChannelSpec<KafkaPollableChannelSpec> {
+public class KafkaPollableChannelSpec extends AbstractKafkaChannelSpec<KafkaPollableChannelSpec, PollableKafkaChannel> {
 
 	protected KafkaPollableChannelSpec(KafkaTemplate<?, ?> template, KafkaMessageSource<?, ?> source) {
-		super(null, null);
 		this.channel = new PollableKafkaChannel(template, source);
 	}
 

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<PublishSubscribeKafkaChannel> {
+
+	protected KafkaPublishSubscribeChannelSpec(KafkaTemplate<?, ?> template, KafkaListenerContainerFactory<?> factory,
+			String topic) {
+
+		this.channel = new PublishSubscribeKafkaChannel(template, factory, topic);
+	}
+
+}

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 <C> the channel type.
+ *
  * @author Gary Russell
  * @since 3.3
  *
  */
-public class KafkaSubscribableChannelSpec extends AbstractKafkaChannelSpec<KafkaSubscribableChannelSpec> {
-
-	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<C extends SubscribableKafkaChannel>
+		extends AbstractKafkaChannelSpec<KafkaSubscribableChannelSpec<C>, C> {
 
 }