From e9edbd27d97e073d1a6992df2c9465f68c91c711 Mon Sep 17 00:00:00 2001
From: Nikita Koksharov <nkoksharov@redisson.pro>
Date: Tue, 20 Nov 2018 17:32:41 +0300
Subject: [PATCH] RStreamReactive and RStreamRx synced with RStream object

---
 .../org/redisson/api/RStreamReactive.java     | 161 ++++++++++++++++++
 .../main/java/org/redisson/api/RStreamRx.java | 161 ++++++++++++++++++
 2 files changed, 322 insertions(+)

diff --git a/redisson/src/main/java/org/redisson/api/RStreamReactive.java b/redisson/src/main/java/org/redisson/api/RStreamReactive.java
index 7dc84fa9c..a6172a701 100644
--- a/redisson/src/main/java/org/redisson/api/RStreamReactive.java
+++ b/redisson/src/main/java/org/redisson/api/RStreamReactive.java
@@ -188,6 +188,167 @@ public interface RStreamReactive<K, V> extends RExpirableReactive {
      */
     Publisher<Map<StreamMessageId, Map<K, V>>> readGroup(String groupName, String consumerName, int count, long timeout, TimeUnit unit, StreamMessageId ... ids);
 
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param id - starting message id for this stream
+     * @param nameToId - Stream Message ID mapped by stream name
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Publisher<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName,StreamMessageId id, Map<String, StreamMessageId> nameToId);
+    
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param count - stream data size limit
+     * @param id - starting message id for this stream
+     * @param nameToId - Stream Message ID mapped by stream name
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Publisher<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName, int count, StreamMessageId id, Map<String, StreamMessageId> nameToId);
+
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * Waits for the first stream data availability for specified <code>timeout</code> interval.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param count - stream data size limit
+     * @param timeout - time interval to wait for stream data availability
+     * @param unit - time interval unit
+     * @param id - starting message id for this stream
+     * @param key2 - name of second stream
+     * @param id2 - starting message id for second stream
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Publisher<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName, int count, long timeout, TimeUnit unit, StreamMessageId id, String key2, StreamMessageId id2);
+
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * Waits for the first stream data availability for specified <code>timeout</code> interval.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param count - stream data size limit
+     * @param timeout - time interval to wait for stream data availability
+     * @param unit - time interval unit
+     * @param id - starting message id for this stream
+     * @param key2 - name of second stream
+     * @param id2 - starting message id for second stream
+     * @param key3 - name of third stream
+     * @param id3 - starting message id for third stream
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Publisher<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName, int count, long timeout, TimeUnit unit, StreamMessageId id,
+            String key2, StreamMessageId id2, String key3, StreamMessageId id3);
+    
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * Waits for the first stream data availability for specified <code>timeout</code> interval.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param timeout - time interval to wait for stream data availability
+     * @param unit - time interval unit
+     * @param id - starting message id for this stream
+     * @param nameToId - Stream Message ID mapped by stream name
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Publisher<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName, long timeout, TimeUnit unit, StreamMessageId id, Map<String, StreamMessageId> nameToId);
+    
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param id - starting message id for this stream
+     * @param key2 - name of second stream
+     * @param id2 - starting message id for second stream
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Publisher<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName, StreamMessageId id, String key2, StreamMessageId id2);
+
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param id - starting message id for this stream
+     * @param key2 - name of second stream
+     * @param id2  - starting message id for second stream
+     * @param key3 - name of third stream
+     * @param id3  - starting message id for third stream
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Publisher<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName, StreamMessageId id, String key2, StreamMessageId id2, String key3,
+            StreamMessageId id3);
+
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param count - stream data size limit
+     * @param id - starting message id for this stream
+     * @param key2 - name of second stream
+     * @param id2  - starting message id for second stream
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Publisher<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName, int count, StreamMessageId id, String key2, StreamMessageId id2);
+
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param count - stream data size limit
+     * @param id - starting message id for this stream
+     * @param key2 - name of second stream
+     * @param id2 - starting message id for second stream
+     * @param key3 - name of third stream
+     * @param id3 - starting message id for third stream
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Publisher<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName, int count, StreamMessageId id, String key2, StreamMessageId id2,
+            String key3, StreamMessageId id3);
+
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * Waits for the first stream data availability for specified <code>timeout</code> interval.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param timeout - time interval to wait for stream data availability
+     * @param unit - time interval unit
+     * @param id - starting message id for this stream
+     * @param key2 - name of second stream
+     * @param id2 - starting message id for second stream
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Publisher<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName, long timeout, TimeUnit unit, StreamMessageId id,
+            String key2, StreamMessageId id2);
+
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * Waits for the first stream data availability for specified <code>timeout</code> interval.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param timeout - time interval to wait for stream data availability
+     * @param unit - time interval unit
+     * @param id - starting message id for this stream
+     * @param key2 - name of second stream
+     * @param id2 - starting message id for second stream
+     * @param key3 - name of third stream
+     * @param id3 - starting message id for third stream
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Publisher<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName, long timeout, TimeUnit unit, StreamMessageId id,
+            String key2, StreamMessageId id2, String key3, StreamMessageId id3);
     
     /**
      * Returns number of entries in stream
diff --git a/redisson/src/main/java/org/redisson/api/RStreamRx.java b/redisson/src/main/java/org/redisson/api/RStreamRx.java
index 681aaa016..fb3160dd2 100644
--- a/redisson/src/main/java/org/redisson/api/RStreamRx.java
+++ b/redisson/src/main/java/org/redisson/api/RStreamRx.java
@@ -188,6 +188,167 @@ public interface RStreamRx<K, V> extends RExpirableRx {
      */
     Flowable<Map<StreamMessageId, Map<K, V>>> readGroup(String groupName, String consumerName, int count, long timeout, TimeUnit unit, StreamMessageId ... ids);
 
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param id - starting message id for this stream
+     * @param nameToId - Stream Message ID mapped by stream name
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Flowable<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName,StreamMessageId id, Map<String, StreamMessageId> nameToId);
+    
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param count - stream data size limit
+     * @param id - starting message id for this stream
+     * @param nameToId - Stream Message ID mapped by stream name
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Flowable<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName, int count, StreamMessageId id, Map<String, StreamMessageId> nameToId);
+
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * Waits for the first stream data availability for specified <code>timeout</code> interval.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param count - stream data size limit
+     * @param timeout - time interval to wait for stream data availability
+     * @param unit - time interval unit
+     * @param id - starting message id for this stream
+     * @param key2 - name of second stream
+     * @param id2 - starting message id for second stream
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Flowable<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName, int count, long timeout, TimeUnit unit, StreamMessageId id, String key2, StreamMessageId id2);
+
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * Waits for the first stream data availability for specified <code>timeout</code> interval.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param count - stream data size limit
+     * @param timeout - time interval to wait for stream data availability
+     * @param unit - time interval unit
+     * @param id - starting message id for this stream
+     * @param key2 - name of second stream
+     * @param id2 - starting message id for second stream
+     * @param key3 - name of third stream
+     * @param id3 - starting message id for third stream
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Flowable<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName, int count, long timeout, TimeUnit unit, StreamMessageId id,
+            String key2, StreamMessageId id2, String key3, StreamMessageId id3);
+    
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * Waits for the first stream data availability for specified <code>timeout</code> interval.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param timeout - time interval to wait for stream data availability
+     * @param unit - time interval unit
+     * @param id - starting message id for this stream
+     * @param nameToId - Stream Message ID mapped by stream name
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Flowable<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName, long timeout, TimeUnit unit, StreamMessageId id, Map<String, StreamMessageId> nameToId);
+    
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param id - starting message id for this stream
+     * @param key2 - name of second stream
+     * @param id2 - starting message id for second stream
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Flowable<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName, StreamMessageId id, String key2, StreamMessageId id2);
+
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param id - starting message id for this stream
+     * @param key2 - name of second stream
+     * @param id2  - starting message id for second stream
+     * @param key3 - name of third stream
+     * @param id3  - starting message id for third stream
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Flowable<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName, StreamMessageId id, String key2, StreamMessageId id2, String key3,
+            StreamMessageId id3);
+
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param count - stream data size limit
+     * @param id - starting message id for this stream
+     * @param key2 - name of second stream
+     * @param id2  - starting message id for second stream
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Flowable<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName, int count, StreamMessageId id, String key2, StreamMessageId id2);
+
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param count - stream data size limit
+     * @param id - starting message id for this stream
+     * @param key2 - name of second stream
+     * @param id2 - starting message id for second stream
+     * @param key3 - name of third stream
+     * @param id3 - starting message id for third stream
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Flowable<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName, int count, StreamMessageId id, String key2, StreamMessageId id2,
+            String key3, StreamMessageId id3);
+
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * Waits for the first stream data availability for specified <code>timeout</code> interval.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param timeout - time interval to wait for stream data availability
+     * @param unit - time interval unit
+     * @param id - starting message id for this stream
+     * @param key2 - name of second stream
+     * @param id2 - starting message id for second stream
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Flowable<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName, long timeout, TimeUnit unit, StreamMessageId id,
+            String key2, StreamMessageId id2);
+
+    /**
+     * Read stream data from <code>groupName</code> by <code>consumerName</code>, starting by specified message ids for this and other streams.
+     * Waits for the first stream data availability for specified <code>timeout</code> interval.
+     * 
+     * @param groupName - name of group
+     * @param consumerName - name of consumer
+     * @param timeout - time interval to wait for stream data availability
+     * @param unit - time interval unit
+     * @param id - starting message id for this stream
+     * @param key2 - name of second stream
+     * @param id2 - starting message id for second stream
+     * @param key3 - name of third stream
+     * @param id3 - starting message id for third stream
+     * @return stream data mapped by key and Stream Message ID
+     */
+    Flowable<Map<String, Map<StreamMessageId, Map<K, V>>>> readGroup(String groupName, String consumerName, long timeout, TimeUnit unit, StreamMessageId id,
+            String key2, StreamMessageId id2, String key3, StreamMessageId id3);
     
     /**
      * Returns number of entries in stream