DATAREDIS-1000 - Fix raw/byte array serialization via RedisSerializationContext.

We now use the correct generic parameters for raw serialization via RedisSerializationContext returning ByteBuffer.

Introduce RedisSerializer.raw() and cleanups.

Original Pull Request: #459

src/main/java/org/springframework/data/redis/serializer/RawRedisSerializer.java

@@ -13,13 +13,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.springframework.data.redis.stream;
+package org.springframework.data.redis.serializer;
 
-import org.springframework.data.redis.serializer.RedisSerializer;
-import org.springframework.data.redis.serializer.SerializationException;
 import org.springframework.lang.Nullable;
 
 /**
+ * Raw {@link RedisSerializer} using {@code byte[]}.
+ *
  * @author Mark Paluch
  */
 enum RawRedisSerializer implements RedisSerializer<byte[]> {

src/main/java/org/springframework/data/redis/serializer/RedisSerializationContext.java

@@ -79,12 +79,23 @@ public interface RedisSerializationContext<K, V> {
 	}
 
 	/**
-	 * Creates a new {@link RedisSerializationContext} using a {@link SerializationPair#raw()} serialization pair.
+	 * Creates a new {@link RedisSerializationContext} using a {@link SerializationPair#raw() ByteBuffer} serialization
+	 * pair.
 	 *
 	 * @return
 	 */
-	static RedisSerializationContext<byte[], byte[]> raw() {
-		return just(SerializationPair.raw());
+	static RedisSerializationContext<ByteBuffer, ByteBuffer> raw() {
+		return just(RedisSerializerToSerializationPairAdapter.raw());
+	}
+
+	/**
+	 * Creates a new {@link RedisSerializationContext} using a {@link RedisSerializer#raw() byte[]} serialization.
+	 *
+	 * @return
+	 * @since 2.2
+	 */
+	static RedisSerializationContext<byte[], byte[]> byteArray() {
+		return just(RedisSerializerToSerializationPairAdapter.byteArray());
 	}
 
 	/**
@@ -207,7 +218,7 @@ public interface RedisSerializationContext<K, V> {
 		 *
 		 * @return a pass through {@link SerializationPair}.
 		 */
-		static <T> SerializationPair<T> raw() {
+		static SerializationPair<ByteBuffer> raw() {
 			return RedisSerializerToSerializationPairAdapter.raw();
 		}
 

src/main/java/org/springframework/data/redis/serializer/RedisSerializer.java

@@ -92,6 +92,16 @@ public interface RedisSerializer<T> {
 		return StringRedisSerializer.UTF_8;
 	}
 
+	/**
+	 * Obtain a raw {@link RedisSerializer} that passes thru {@code byte[]}.
+	 *
+	 * @return never {@literal null}.
+	 * @since 2.2
+	 */
+	static RedisSerializer<byte[]> raw() {
+		return RawRedisSerializer.INSTANCE;
+	}
+
 	default boolean canSerialize(Class<?> type) {
 		return ClassUtils.isAssignable(getTargetType(), type);
 	}

src/main/java/org/springframework/data/redis/serializer/RedisSerializerToSerializationPairAdapter.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.redis.serializer;
 
+import java.nio.ByteBuffer;
+
 import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -31,18 +33,25 @@ class RedisSerializerToSerializationPairAdapter<T> implements SerializationPair<
 	private final static RedisSerializerToSerializationPairAdapter<?> RAW = new RedisSerializerToSerializationPairAdapter<>(
 			null);
 
+	private final static RedisSerializerToSerializationPairAdapter<byte[]> BYTE_ARRAY = new RedisSerializerToSerializationPairAdapter<>(
+			RedisSerializer.raw());
+
 	private final DefaultSerializationPair pair;
 
-	protected RedisSerializerToSerializationPairAdapter(@Nullable RedisSerializer<T> serializer) {
-		pair = new DefaultSerializationPair(new DefaultRedisElementReader<>(serializer),
+	RedisSerializerToSerializationPairAdapter(@Nullable RedisSerializer<T> serializer) {
+		pair = new DefaultSerializationPair<>(new DefaultRedisElementReader<>(serializer),
 				new DefaultRedisElementWriter<>(serializer));
 	}
 
 	@SuppressWarnings("unchecked")
-	public static <T> SerializationPair<T> raw() {
+	static SerializationPair<ByteBuffer> raw() {
 		return (SerializationPair) RAW;
 	}
 
+	static SerializationPair<byte[]> byteArray() {
+		return BYTE_ARRAY;
+	}
+
 	/**
 	 * Create a {@link SerializationPair} from given {@link RedisSerializer}.
 	 *

src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java

@@ -722,8 +722,8 @@ public interface StreamMessageListenerContainer<K, V extends Record<K, ?>> exten
 
 			if (this.hashMapper == null) {
 
-				hashKeySerializer(RawRedisSerializer.INSTANCE);
-				hashValueSerializer(RawRedisSerializer.INSTANCE);
+				hashKeySerializer(RedisSerializer.raw());
+				hashValueSerializer(RedisSerializer.raw());
 				return (StreamMessageListenerContainerOptionsBuilder) objectMapper(new ObjectHashMapper());
 			}