Polishing.

Add new method to BoundStreamOperations and remove add accepting a Publisher. Original Pull Request: spring-projects#2936

Author: christophstrobl

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

@@ -403,7 +403,7 @@ default Mono<RecordId> xAdd(ByteBufferRecord record) {
  * @param xAddOptions parameters for the {@literal XADD} call. Must not be {@literal null}.
  * @return {@link Mono} the {@link RecordId id}.
  * @see <a href="https://redis.io/commands/xadd">Redis Documentation: XADD</a>
- * @since 3.3
+ * @since 3.4
  */
 default Mono<RecordId> xAdd(ByteBufferRecord record, XAddOptions xAddOptions) {
 

src/main/java/org/springframework/data/redis/core/BoundStreamOperations.java

@@ -20,6 +20,7 @@
 
 import org.springframework.data.domain.Range;
 import org.springframework.data.redis.connection.Limit;
+import org.springframework.data.redis.connection.RedisStreamCommands.XAddOptions;
 import org.springframework.data.redis.connection.stream.Consumer;
 import org.springframework.data.redis.connection.stream.MapRecord;
 import org.springframework.data.redis.connection.stream.ReadOffset;
@@ -58,6 +59,18 @@ public interface BoundStreamOperations<K, HK, HV> {
 @Nullable
 RecordId add(Map<HK, HV> body);
 
+/**
+ * Append a record to the stream {@code key} with the specified options.
+ *
+ * @param content record content as Map.
+ * @param xAddOptions additional parameters for the {@literal XADD} call.
+ * @return the record Id. {@literal null} when used in pipeline / transaction.
+ * @see <a href="https://redis.io/commands/xadd">Redis Documentation: XADD</a>
+ * @since 3.4
+ */
+@Nullable
+RecordId add(Map<HK, HV> content, XAddOptions xAddOptions);
+
 /**
  * Removes the specified entries from the stream. Returns the number of items deleted, that may be different from the
  * number of IDs passed in case certain IDs do not exist.

src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java

@@ -96,21 +96,6 @@ default Mono<Long> acknowledge(String group, Record<K, ?> record) {
 return acknowledge(record.getRequiredStream(), group, record.getId());
 }
 
-/**
- * Append one or more records to the stream {@code key} with the specified options.
- *
- * @param key the stream key.
- * @param bodyPublisher record body {@link Publisher}.
- * @param xAddOptions parameters for the {@literal XADD} call.
- * @return the record Ids.
- * @see <a href="https://redis.io/commands/xadd">Redis Documentation: XADD</a>
- * @since 3.3
- */
-default Flux<RecordId> add (K key, Publisher<? extends Map<? extends HK, ? extends HV>> bodyPublisher,
-XAddOptions xAddOptions) {
-return Flux.from(bodyPublisher).flatMap(it -> add(key, it, xAddOptions));
-}
-
 /**
  * Append a record to the stream {@code key} with the specified options.
  *
@@ -119,7 +104,7 @@ default Flux<RecordId> add (K key, Publisher<? extends Map<? extends HK, ? exten
  * @param xAddOptions parameters for the {@literal XADD} call.
  * @return the {@link Mono} emitting the {@link RecordId}.
  * @see <a href="https://redis.io/commands/xadd">Redis Documentation: XADD</a>
- * @since 3.3
+ * @since 3.4
  */
 default Mono<RecordId> add(K key, Map<? extends HK, ? extends HV> content, XAddOptions xAddOptions) {
 return add(StreamRecords.newRecord().in(key).ofMap(content), xAddOptions);
@@ -132,7 +117,7 @@ default Mono<RecordId> add(K key, Map<? extends HK, ? extends HV> content, XAddO
  * @param xAddOptions parameters for the {@literal XADD} call.
  * @return the {@link Mono} emitting the {@link RecordId}.
  * @see <a href="https://redis.io/commands/xadd">Redis Documentation: XADD</a>
- * @since 3.3
+ * @since 3.4
  */
 @SuppressWarnings("unchecked")
 default Mono<RecordId> add(MapRecord<K, ? extends HK, ? extends HV> record, XAddOptions xAddOptions) {
@@ -149,7 +134,7 @@ default Mono<RecordId> add(MapRecord<K, ? extends HK, ? extends HV> record, XAdd
  * @see MapRecord
  * @see ObjectRecord
  * @see <a href="https://redis.io/commands/xadd">Redis Documentation: XADD</a>
- * @since 3.3
+ * @since 3.4
  */
 Mono<RecordId> add(Record<K, ?> record, XAddOptions xAddOptions);
 

src/main/java/org/springframework/data/redis/core/StreamOperations.java

@@ -105,7 +105,7 @@ default Long acknowledge(String group, Record<K, ?> record) {
  * @param xAddOptions additional parameters for the {@literal XADD} call.
  * @return the record Id. {@literal null} when used in pipeline / transaction.
  * @see <a href="https://redis.io/commands/xadd">Redis Documentation: XADD</a>
- * @since 3.3
+ * @since 3.4
  */
 @SuppressWarnings("unchecked")
 @Nullable
@@ -120,7 +120,7 @@ default RecordId add(K key, Map<? extends HK, ? extends HV> content, XAddOptions
  * @param xAddOptions additional parameters for the {@literal XADD} call.
  * @return the record Id. {@literal null} when used in pipeline / transaction.
  * @see <a href="https://redis.io/commands/xadd">Redis Documentation: XADD</a>
- * @since 3.3
+ * @since 3.4
  */
 @SuppressWarnings("unchecked")
 @Nullable
@@ -138,7 +138,7 @@ default RecordId add(MapRecord<K, ? extends HK, ? extends HV> record, XAddOption
  * @see MapRecord
  * @see ObjectRecord
  * @see <a href="https://redis.io/commands/xadd">Redis Documentation: XADD</a>
- * @since 3.3
+ * @since 3.4
  */
 @SuppressWarnings("unchecked")
 @Nullable