Polishing.

Add explicit cast update documentation and rename SubcriptionListener instance to NO_OP_...
Fix Java 16 compile error.

Original Pull Request: #2052

Author: christophstrobl

src/main/java/org/springframework/data/redis/connection/ReactivePubSubCommands.java

@@ -34,11 +34,13 @@ public interface ReactivePubSubCommands {
 
 	/**
 	 * Creates a subscription for this connection. Connections can have multiple {@link ReactiveSubscription}s.
+	 * <p/>
+	 * Use {@link #createSubscription(SubscriptionListener)} to get notified when the subscription completes.
 	 *
 	 * @return the subscription.
 	 */
 	default Mono<ReactiveSubscription> createSubscription() {
-		return createSubscription(SubscriptionListener.EMPTY);
+		return createSubscription(SubscriptionListener.NO_OP_SUBSCRIPTION_LISTENER);
 	}
 
 	/**

src/main/java/org/springframework/data/redis/connection/SubscriptionListener.java

@@ -29,7 +29,7 @@ public interface SubscriptionListener {
 	/**
 	 * Empty {@link SubscriptionListener}.
 	 */
-	SubscriptionListener EMPTY = new SubscriptionListener() {};
+	SubscriptionListener NO_OP_SUBSCRIPTION_LISTENER = new SubscriptionListener() {};
 
 	/**
 	 * Notification when Redis has confirmed a channel subscription.

src/main/java/org/springframework/data/redis/connection/jedis/JedisMessageListener.java

@@ -34,10 +34,12 @@ class JedisMessageListener extends BinaryJedisPubSub {
 	private final SubscriptionListener subscriptionListener;
 
 	JedisMessageListener(MessageListener listener) {
+
 		Assert.notNull(listener, "MessageListener is required");
+
 		this.listener = listener;
 		this.subscriptionListener = listener instanceof SubscriptionListener ? (SubscriptionListener) listener
-				: SubscriptionListener.EMPTY;
+				: SubscriptionListener.NO_OP_SUBSCRIPTION_LISTENER;
 	}
 
 	public void onMessage(byte[] channel, byte[] message) {

src/main/java/org/springframework/data/redis/connection/jedis/JedisSubscription.java

@@ -30,7 +30,7 @@ class JedisSubscription extends AbstractSubscription {
 
 	private final BinaryJedisPubSub jedisPubSub;
 
-	JedisSubscription(MessageListener listener, JedisMessageListener jedisPubSub, @Nullable byte[][] channels,
+	JedisSubscription(MessageListener listener, BinaryJedisPubSub jedisPubSub, @Nullable byte[][] channels,
 			@Nullable byte[][] patterns) {
 		super(listener, channels, patterns);
 		this.jedisPubSub = jedisPubSub;

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceByteBufferPubSubListenerWrapper.java

@@ -19,6 +19,7 @@
 
 import java.nio.ByteBuffer;
 
+import org.springframework.data.redis.util.ByteUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -99,14 +100,6 @@ private static byte[] getBytes(@Nullable ByteBuffer byteBuffer) {
 			return new byte[0];
 		}
 
-		if (byteBuffer.hasArray()) {
-			return byteBuffer.array();
-		}
-
-		byteBuffer.mark();
-		byte[] bytes = new byte[byteBuffer.remaining()];
-		byteBuffer.get(bytes);
-		byteBuffer.reset();
-		return bytes;
+		return ByteUtils.getBytes(byteBuffer);
 	}
 }

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java

@@ -60,7 +60,7 @@ protected LettuceSubscription(MessageListener listener,
 
 		this.connection = pubsubConnection;
 		this.listener = new LettuceMessageListener(listener,
-				listener instanceof SubscriptionListener ? (SubscriptionListener) listener : SubscriptionListener.EMPTY);
+				listener instanceof SubscriptionListener ? (SubscriptionListener) listener : SubscriptionListener.NO_OP_SUBSCRIPTION_LISTENER);
 		this.connectionProvider = connectionProvider;
 		this.pubsub = connection.sync();
 		this.pubSubAsync = connection.async();

src/main/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainer.java

@@ -261,7 +261,7 @@ public Flux<Message<String, String>> receive(Iterable<? extends Topic> topics,
 	 */
 	public <C, B> Flux<Message<C, B>> receive(Iterable<? extends Topic> topics, SerializationPair<C> channelSerializer,
 			SerializationPair<B> messageSerializer) {
-		return receive(topics, channelSerializer, messageSerializer, SubscriptionListener.EMPTY);
+		return receive(topics, channelSerializer, messageSerializer, SubscriptionListener.NO_OP_SUBSCRIPTION_LISTENER);
 	}
 
 	/**

src/main/java/org/springframework/data/redis/util/ByteUtils.java

@@ -139,6 +139,10 @@ public static byte[] getBytes(ByteBuffer byteBuffer) {
 
 		Assert.notNull(byteBuffer, "ByteBuffer must not be null!");
 
+		if (byteBuffer.hasArray()) {
+			return byteBuffer.array();
+		}
+
 		ByteBuffer duplicate = byteBuffer.duplicate();
 		byte[] bytes = new byte[duplicate.remaining()];
 		duplicate.get(bytes);

src/test/java/org/springframework/data/redis/connection/jedis/JedisSubscriptionUnitTests.java

@@ -39,7 +39,7 @@
 @ExtendWith(MockitoExtension.class)
 class JedisSubscriptionUnitTests {
 
-	@Mock JedisMessageListener jedisPubSub;
+	@Mock BinaryJedisPubSub jedisPubSub;
 
 	@Mock MessageListener listener;
 

src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateIntegrationTests.java

@@ -465,6 +465,7 @@ void listenToLaterChannelShouldReceiveChannelMessagesCorrectly() {
 
 		redisTemplate.listenToChannelLater(channel) //
 				.doOnNext(it -> redisTemplate.convertAndSend(channel, message).subscribe()).flatMapMany(Function.identity()) //
+				.cast(Message.class)  // why? java16 why?
 				.as(StepVerifier::create) //
 				.assertNext(received -> {
 
@@ -515,6 +516,7 @@ void listenToPatternLaterShouldReceiveChannelMessagesCorrectly() {
 
 		stream.doOnNext(it -> redisTemplate.convertAndSend(channel, message).subscribe()) //
 				.flatMapMany(Function.identity()) //
+				.cast(Message.class) // why? java16 why?
 				.as(StepVerifier::create) //
 				.assertNext(received -> {