GH-3194: Add generics to Amqp and Jms channel impls

Fixes https://github.com/spring-projects/spring-integration/issues/3194

For now we keep the hierarchy, but try to generalize channel specs
a bit, so that ones who use `publishSubscribeChannel` can count on
`BroadcastCapableChannel` as interface to utilize.

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/Amqp.java

@@ -22,6 +22,7 @@ import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.integration.amqp.channel.PollableAmqpChannel;
 import org.springframework.integration.amqp.inbound.AmqpMessageSource.AmqpAckCallbackFactory;
 import org.springframework.lang.Nullable;
 
@@ -30,6 +31,7 @@ import org.springframework.lang.Nullable;
  *
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Artem Vozhdayenko
  *
  * @since 5.0
  */
@@ -269,7 +271,7 @@ public final class Amqp {
 	 * @param connectionFactory the connectionFactory.
 	 * @return the AmqpPollableMessageChannelSpec.
 	 */
-	public static AmqpPollableMessageChannelSpec<?> pollableChannel(ConnectionFactory connectionFactory) {
+	public static AmqpPollableMessageChannelSpec<?, PollableAmqpChannel> pollableChannel(ConnectionFactory connectionFactory) {
 		return pollableChannel(null, connectionFactory);
 	}
 
@@ -279,11 +281,11 @@ public final class Amqp {
 	 * @param connectionFactory the connectionFactory.
 	 * @return the AmqpPollableMessageChannelSpec.
 	 */
-	public static AmqpPollableMessageChannelSpec<?> pollableChannel(@Nullable String id,
+	public static AmqpPollableMessageChannelSpec<?, PollableAmqpChannel> pollableChannel(@Nullable String id,
 			ConnectionFactory connectionFactory) {
 
-		return new AmqpPollableMessageChannelSpec<>(connectionFactory)
-				.id(id);
+		AmqpPollableMessageChannelSpec<?, PollableAmqpChannel> spec = new AmqpPollableMessageChannelSpec<>(connectionFactory);
+		return spec.id(id);
 	}
 
 	/**
@@ -291,7 +293,7 @@ public final class Amqp {
 	 * @param connectionFactory the connectionFactory.
 	 * @return the AmqpMessageChannelSpec.
 	 */
-	public static AmqpMessageChannelSpec<?> channel(ConnectionFactory connectionFactory) {
+	public static AmqpMessageChannelSpec<?, ?> channel(ConnectionFactory connectionFactory) {
 		return channel(null, connectionFactory);
 	}
 
@@ -301,7 +303,7 @@ public final class Amqp {
 	 * @param connectionFactory the connectionFactory.
 	 * @return the AmqpMessageChannelSpec.
 	 */
-	public static AmqpMessageChannelSpec<?> channel(@Nullable String id, ConnectionFactory connectionFactory) {
+	public static AmqpMessageChannelSpec<?, ?> channel(@Nullable String id, ConnectionFactory connectionFactory) {
 		return new AmqpMessageChannelSpec<>(connectionFactory)
 				.id(id);
 	}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpMessageChannelSpec.java

@@ -39,10 +39,12 @@ import org.springframework.util.ErrorHandler;
  *
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Artem Vozhdayenko
  *
  * @since 5.0
  */
-public class AmqpMessageChannelSpec<S extends AmqpMessageChannelSpec<S>> extends AmqpPollableMessageChannelSpec<S> {
+public class AmqpMessageChannelSpec<S extends AmqpMessageChannelSpec<S, T>, T extends AbstractAmqpChannel>
+		extends AmqpPollableMessageChannelSpec<S, T> {
 
 	protected final List<Advice> adviceChain = new LinkedList<>(); // NOSONAR
 
@@ -215,7 +217,7 @@ public class AmqpMessageChannelSpec<S extends AmqpMessageChannelSpec<S>> extends
 	}
 
 	@Override
-	protected AbstractAmqpChannel doGet() {
+	protected T doGet() {
 		this.amqpChannelFactoryBean.setAdviceChain(this.adviceChain.toArray(new Advice[0]));
 		return super.doGet();
 	}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpPollableMessageChannelSpec.java

@@ -36,11 +36,12 @@ import org.springframework.util.Assert;
  *
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Artem Vozhdayenko
  *
  * @since 5.0
  */
-public class AmqpPollableMessageChannelSpec<S extends AmqpPollableMessageChannelSpec<S>>
-		extends MessageChannelSpec<S, AbstractAmqpChannel> {
+public class AmqpPollableMessageChannelSpec<S extends AmqpPollableMessageChannelSpec<S, T>, T extends AbstractAmqpChannel>
+		extends MessageChannelSpec<S, T> {
 
 	protected final AmqpChannelFactoryBean amqpChannelFactoryBean; // NOSONAR final
 
@@ -206,10 +207,11 @@ public class AmqpPollableMessageChannelSpec<S extends AmqpPollableMessageChannel
 	}
 
 	@Override
-	protected AbstractAmqpChannel doGet() {
+	@SuppressWarnings("unchecked")
+	protected T doGet() {
 		Assert.notNull(getId(), "The 'id' or 'queueName' must be specified");
 		try {
-			this.channel = this.amqpChannelFactoryBean.getObject();
+			this.channel = (T) this.amqpChannelFactoryBean.getObject();
 		}
 		catch (Exception e) {
 			throw new BeanCreationException("Cannot create the AMQP MessageChannel", e);

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpPublishSubscribeMessageChannelSpec.java

@@ -18,16 +18,19 @@ package org.springframework.integration.amqp.dsl;
 
 import org.springframework.amqp.core.FanoutExchange;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.integration.amqp.channel.PollableAmqpChannel;
 
 /**
  * A {@link AmqpMessageChannelSpec} for
  * {@link org.springframework.integration.amqp.channel.PublishSubscribeAmqpChannel}s.
  *
  * @author Artem Bilan
+ * @author Artem Vozhdayenko
+ *
  * @since 5.0
  */
 public class AmqpPublishSubscribeMessageChannelSpec
-		extends AmqpMessageChannelSpec<AmqpPublishSubscribeMessageChannelSpec> {
+		extends AmqpMessageChannelSpec<AmqpPublishSubscribeMessageChannelSpec, PollableAmqpChannel> {
 
 	protected AmqpPublishSubscribeMessageChannelSpec(ConnectionFactory connectionFactory) {
 		super(connectionFactory);