Add stream-key auto-conversion for ByteRecord.

Add automatic type conversion logic in StreamRecords.ofBytes() method
to handle different stream key types safely.

Signed-off-by: Seo Bo Gyeong <[email protected]>
Closes #3204

Author: dodu232

src/main/java/org/springframework/data/redis/connection/stream/StreamRecords.java

@@ -30,6 +30,7 @@
  * {@link StreamRecords} provides utilities to create specific {@link Record} instances.
  *
  * @author Christoph Strobl
+ * @author Seo Bo Gyeong
  * @since 2.2
  */
 public class StreamRecords {
@@ -197,8 +198,24 @@ public <V> ObjectRecord<S, V> ofObject(V value) {
 		 */
 		public ByteRecord ofBytes(Map<byte[], byte[]> value) {
 
-			// todo auto conversion of known values
-			return new ByteMapBackedRecord((byte[]) stream, id, value);
+			byte[] streamKey = convertStreamToByteArray(stream);
+			return new ByteMapBackedRecord(streamKey, id, value);
+		}
+
+		private byte[] convertStreamToByteArray(@Nullable Object stream) {
+			if (stream instanceof byte[]) {
+				return (byte[]) stream;
+			} else if (stream instanceof String) {
+				return ((String) stream).getBytes();
+			} else if (stream instanceof ByteBuffer buffer) {
+				byte[] result = new byte[buffer.remaining()];
+				buffer.get(result);
+				return result;
+			} else if (stream == null) {
+				return null;
+			} else {
+				throw new IllegalArgumentException("Stream key %s cannot be converted to byte array".formatted(stream));
+			}
 		}
 
 		/**

src/test/java/org/springframework/data/redis/connection/StreamRecordsUnitTests.java

@@ -18,6 +18,7 @@
 import static org.assertj.core.api.Assertions.*;
 
 import java.io.Serializable;
+import java.nio.ByteBuffer;
 import java.util.Collections;
 import java.util.Map;
 
@@ -164,4 +165,72 @@ static HashMapper<Object, String, String> simpleString(String value) {
 		}
 	}
 
+	@Test // Stream key auto conversion for ofBytes method
+	void ofBytesWithStringStreamKey() {
+
+		ByteRecord record = StreamRecords.newRecord()
+				.in(STRING_STREAM_KEY)
+				.withId(RECORD_ID)
+				.ofBytes(Collections.singletonMap(SERIALIZED_STRING_MAP_KEY, SERIALIZED_STRING_VAL));
+
+		assertThat(record.getId()).isEqualTo(RECORD_ID);
+		assertThat(record.getStream()).isEqualTo(SERIALIZED_STRING_STREAM_KEY);
+		assertThat(record.getValue().keySet().iterator().next()).isEqualTo(SERIALIZED_STRING_MAP_KEY);
+		assertThat(record.getValue().values().iterator().next()).isEqualTo(SERIALIZED_STRING_VAL);
+	}
+
+	@Test // Stream key auto conversion for ofBytes method with byte array
+	void ofBytesWithByteArrayStreamKey() {
+
+		ByteRecord record = StreamRecords.newRecord()
+				.in(SERIALIZED_STRING_STREAM_KEY)
+				.withId(RECORD_ID)
+				.ofBytes(Collections.singletonMap(SERIALIZED_STRING_MAP_KEY, SERIALIZED_STRING_VAL));
+
+		assertThat(record.getId()).isEqualTo(RECORD_ID);
+		assertThat(record.getStream()).isEqualTo(SERIALIZED_STRING_STREAM_KEY);
+		assertThat(record.getValue().keySet().iterator().next()).isEqualTo(SERIALIZED_STRING_MAP_KEY);
+		assertThat(record.getValue().values().iterator().next()).isEqualTo(SERIALIZED_STRING_VAL);
+	}
+
+	@Test // Stream key auto conversion for ofBytes method with null stream key
+	void ofBytesWithNullStreamKey() {
+
+		ByteRecord record = StreamRecords.newRecord()
+				.withId(RECORD_ID)
+				.ofBytes(Collections.singletonMap(SERIALIZED_STRING_MAP_KEY, SERIALIZED_STRING_VAL));
+
+		assertThat(record.getId()).isEqualTo(RECORD_ID);
+		assertThat(record.getStream()).isNull();
+		assertThat(record.getValue().keySet().iterator().next()).isEqualTo(SERIALIZED_STRING_MAP_KEY);
+		assertThat(record.getValue().values().iterator().next()).isEqualTo(SERIALIZED_STRING_VAL);
+	}
+
+	@Test // Stream key auto conversion for ofBytes method with ByteBuffer stream key
+	void ofBytesWithByteBufferStreamKey() {
+
+		ByteBuffer streamKeyBuffer = ByteBuffer.wrap(SERIALIZED_STRING_STREAM_KEY);
+
+		ByteRecord record = StreamRecords.newRecord()
+				.in(streamKeyBuffer)
+				.withId(RECORD_ID)
+				.ofBytes(Collections.singletonMap(SERIALIZED_STRING_MAP_KEY, SERIALIZED_STRING_VAL));
+
+		assertThat(record.getId()).isEqualTo(RECORD_ID);
+		assertThat(record.getStream()).isEqualTo(SERIALIZED_STRING_STREAM_KEY);
+		assertThat(record.getValue().keySet().iterator().next()).isEqualTo(SERIALIZED_STRING_MAP_KEY);
+		assertThat(record.getValue().values().iterator().next()).isEqualTo(SERIALIZED_STRING_VAL);
+	}
+
+	@Test // Stream key auto conversion for ofBytes method with unsupported type
+	void ofBytesWithUnsupportedStreamKeyType() {
+
+		assertThatThrownBy(() -> StreamRecords.newRecord()
+				.in(123L) // Unsupported type
+				.withId(RECORD_ID)
+				.ofBytes(Collections.singletonMap(SERIALIZED_STRING_MAP_KEY, SERIALIZED_STRING_VAL)))
+				.isInstanceOf(IllegalArgumentException.class)
+				.hasMessageContaining("Stream key 123 cannot be converted to byte array");
+	}
+
 }