diff --git a/src/main/asciidoc/new-features.adoc b/src/main/asciidoc/new-features.adoc index e758b109e..0b1947d65 100644 --- a/src/main/asciidoc/new-features.adoc +++ b/src/main/asciidoc/new-features.adoc @@ -10,6 +10,7 @@ This section briefly covers items that are new and noteworthy in the latest rele * ACL authentication support for Redis Standalone, Redis Cluster and Master/Replica. * Password support for Redis Sentinel using Jedis. * Support for `ZREVRANGEBYLEX` and `ZLEXCOUNT` commands. +* Support for Stream Commands using Jedis. [[new-in-2.3.0]] == New in Spring Data Redis 2.3 diff --git a/src/main/asciidoc/reference/redis-streams.adoc b/src/main/asciidoc/reference/redis-streams.adoc index f6b6b32be..9529f2c26 100644 --- a/src/main/asciidoc/reference/redis-streams.adoc +++ b/src/main/asciidoc/reference/redis-streams.adoc @@ -16,8 +16,6 @@ While Pub/Sub relies on the broadcasting of transient messages (i.e. if you don' The `org.springframework.data.redis.connection` and `org.springframework.data.redis.stream` packages provide the core functionality for Redis Streams. -NOTE: Redis Stream support is currently only available through the <> as it is not yet supported by <>. - [[redis.streams.send]] == Appending diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java index e31ecc2eb..76926df04 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java @@ -303,6 +303,15 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection { return new JedisClusterSetCommands(this); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisConnection#streamCommands() + */ + @Override + public RedisStreamCommands streamCommands() { + return new JedisClusterStreamCommands(this); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisConnection#zSetCommands() diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterStreamCommands.java new file mode 100644 index 000000000..52900832d --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterStreamCommands.java @@ -0,0 +1,416 @@ +/* + * Copyright 2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.redis.connection.jedis; + +import static org.springframework.data.redis.connection.jedis.StreamConverters.*; + +import redis.clients.jedis.BuilderFactory; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import org.springframework.dao.DataAccessException; +import org.springframework.data.domain.Range; +import org.springframework.data.redis.connection.RedisStreamCommands; +import org.springframework.data.redis.connection.RedisZSetCommands; +import org.springframework.data.redis.connection.stream.ByteRecord; +import org.springframework.data.redis.connection.stream.Consumer; +import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.connection.stream.PendingMessages; +import org.springframework.data.redis.connection.stream.PendingMessagesSummary; +import org.springframework.data.redis.connection.stream.ReadOffset; +import org.springframework.data.redis.connection.stream.RecordId; +import org.springframework.data.redis.connection.stream.StreamInfo; +import org.springframework.data.redis.connection.stream.StreamOffset; +import org.springframework.data.redis.connection.stream.StreamReadOptions; +import org.springframework.util.Assert; + +/** + * @author Dengliming + * @since 2.3 + */ +class JedisClusterStreamCommands implements RedisStreamCommands { + + private final JedisClusterConnection connection; + + JedisClusterStreamCommands(JedisClusterConnection connection) { + this.connection = connection; + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xAck(byte[], String, org.springframework.data.redis.connection.stream.RecordId[]) + */ + @Override + public Long xAck(byte[] key, String group, RecordId... recordIds) { + + Assert.notNull(key, "Key must not be null!"); + Assert.hasText(group, "Group name must not be null or empty!"); + Assert.notNull(recordIds, "recordIds must not be null!"); + + try { + return connection.getCluster().xack(key, JedisConverters.toBytes(group), + entryIdsToBytes(Arrays.asList(recordIds))); + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xAdd(MapRecord, XAddOptions) + */ + @Override + public RecordId xAdd(MapRecord record, XAddOptions options) { + + Assert.notNull(record, "Record must not be null!"); + Assert.notNull(record.getStream(), "Stream must not be null!"); + + byte[] id = JedisConverters.toBytes(record.getId().getValue()); + long maxLength = Long.MAX_VALUE; + if (options.hasMaxlen()) { + maxLength = options.getMaxlen(); + } + + try { + return RecordId.of(JedisConverters + .toString(connection.getCluster().xadd(record.getStream(), id, record.getValue(), maxLength, false))); + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xClaimJustId(byte[], java.lang.String, java.lang.String, org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions) + */ + @Override + public List xClaimJustId(byte[] key, String group, String newOwner, XClaimOptions options) { + throw new UnsupportedOperationException("JedisCluster does not support xClaimJustId."); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xClaim(byte[], java.lang.String, java.lang.String, org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions) + */ + @Override + public List xClaim(byte[] key, String group, String newOwner, XClaimOptions options) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(group, "Group must not be null!"); + Assert.notNull(newOwner, "NewOwner must not be null!"); + + long minIdleTime = options.getMinIdleTime() == null ? -1L : options.getMinIdleTime().toMillis(); + int retryCount = options.getRetryCount() == null ? -1 : options.getRetryCount().intValue(); + long unixTime = options.getUnixTime() == null ? -1L : options.getUnixTime().toEpochMilli(); + + try { + return convertToByteRecord(key, + connection.getCluster().xclaim(key, JedisConverters.toBytes(group), JedisConverters.toBytes(newOwner), + minIdleTime, unixTime, retryCount, options.isForce(), entryIdsToBytes(options.getIds()))); + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xDel(byte[], java.lang.String[]) + */ + @Override + public Long xDel(byte[] key, RecordId... recordIds) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(recordIds, "recordIds must not be null!"); + + try { + return connection.getCluster().xdel(key, entryIdsToBytes(Arrays.asList(recordIds))); + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xGroupCreate(byte[], org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset, java.lang.String) + */ + @Override + public String xGroupCreate(byte[] key, String groupName, ReadOffset readOffset) { + return xGroupCreate(key, groupName, readOffset, false); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xGroupCreate(byte[], org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset, java.lang.String, boolean) + */ + @Override + public String xGroupCreate(byte[] key, String groupName, ReadOffset readOffset, boolean mkStream) { + + Assert.notNull(key, "Key must not be null!"); + Assert.hasText(groupName, "Group name must not be null or empty!"); + Assert.notNull(readOffset, "ReadOffset must not be null!"); + + try { + return connection.getCluster().xgroupCreate(key, JedisConverters.toBytes(groupName), + JedisConverters.toBytes(readOffset.getOffset()), mkStream); + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xGroupDelConsumer(byte[], org.springframework.data.redis.connection.RedisStreamCommands.Consumer) + */ + @Override + public Boolean xGroupDelConsumer(byte[] key, Consumer consumer) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(consumer, "Consumer must not be null!"); + + try { + return connection.getCluster().xgroupDelConsumer(key, JedisConverters.toBytes(consumer.getGroup()), + JedisConverters.toBytes(consumer.getName())) != 0L; + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xGroupDestroy(byte[], java.lang.String) + */ + @Override + public Boolean xGroupDestroy(byte[] key, String groupName) { + + Assert.notNull(key, "Key must not be null!"); + Assert.hasText(groupName, "Group name must not be null or empty!"); + + try { + return connection.getCluster().xgroupDestroy(key, JedisConverters.toBytes(groupName)) != 0L; + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xInfo(byte[]) + */ + @Override + public StreamInfo.XInfoStream xInfo(byte[] key) { + throw new UnsupportedOperationException("JedisCluster does not support XINFO."); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xInfoGroups(byte[]) + */ + @Override + public StreamInfo.XInfoGroups xInfoGroups(byte[] key) { + throw new UnsupportedOperationException("JedisCluster does not support XINFO."); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xInfoConsumers(byte[], java.lang.String) + */ + @Override + public StreamInfo.XInfoConsumers xInfoConsumers(byte[] key, String groupName) { + throw new UnsupportedOperationException("JedisCluster does not support XINFO."); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xLen(byte[]) + */ + @Override + public Long xLen(byte[] key) { + + Assert.notNull(key, "Key must not be null!"); + + try { + return connection.getCluster().xlen(key); + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xPending(byte[], java.lang.String) + */ + @Override + public PendingMessagesSummary xPending(byte[] key, String groupName) { + throw new UnsupportedOperationException("Jedis does not support returning PendingMessagesSummary."); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xPending(byte[], java.lang.String, org.springframework.data.redis.connection.RedisStreamCommands.XPendingOptions) + */ + @Override + public PendingMessages xPending(byte[] key, String groupName, XPendingOptions options) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(groupName, "GroupName must not be null!"); + + Range range = (Range) options.getRange(); + byte[] group = JedisConverters.toBytes(groupName); + + try { + + List response = connection.getCluster().xpending(key, group, + JedisConverters.toBytes(getLowerValue(range)), JedisConverters.toBytes(getUpperValue(range)), + options.getCount().intValue(), JedisConverters.toBytes(options.getConsumerName())); + + return StreamConverters.toPendingMessages(groupName, range, + BuilderFactory.STREAM_PENDING_ENTRY_LIST.build(response)); + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xRange(byte[], org.springframework.data.domain.Range, org.springframework.data.redis.connection.RedisZSetCommands.Limit) + */ + @Override + public List xRange(byte[] key, Range range, RedisZSetCommands.Limit limit) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(range, "Range must not be null!"); + Assert.notNull(limit, "Limit must not be null!"); + + int count = limit.isUnlimited() ? Integer.MAX_VALUE : limit.getCount(); + + try { + return convertToByteRecord(key, connection.getCluster().xrange(key, JedisConverters.toBytes(getLowerValue(range)), + JedisConverters.toBytes(getUpperValue(range)), count)); + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xRead(org.springframework.data.redis.connection.RedisStreamCommands.StreamReadOptions, org.springframework.data.redis.connection.RedisStreamCommands.StreamOffset[]) + */ + @Override + public List xRead(StreamReadOptions readOptions, StreamOffset... streams) { + + Assert.notNull(readOptions, "StreamReadOptions must not be null!"); + Assert.notNull(streams, "StreamOffsets must not be null!"); + + long block = readOptions.getBlock() == null ? -1L : readOptions.getBlock(); + int count = readOptions.getCount() != null ? readOptions.getCount().intValue() : Integer.MAX_VALUE; + + try { + + List xread = connection.getCluster().xread(count, block, toStreamOffsets(streams)); + + if (xread == null) { + return Collections.emptyList(); + } + + return StreamConverters.convertToByteRecords(xread); + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xReadGroup(org.springframework.data.redis.connection.RedisStreamCommands.Consumer, org.springframework.data.redis.connection.RedisStreamCommands.StreamReadOptions, org.springframework.data.redis.connection.RedisStreamCommands.StreamOffset[]) + */ + @Override + public List xReadGroup(Consumer consumer, StreamReadOptions readOptions, + StreamOffset... streams) { + + Assert.notNull(consumer, "Consumer must not be null!"); + Assert.notNull(readOptions, "StreamReadOptions must not be null!"); + Assert.notNull(streams, "StreamOffsets must not be null!"); + + long block = readOptions.getBlock() == null ? -1L : readOptions.getBlock(); + int count = readOptions.getCount() == null ? -1 : readOptions.getCount().intValue(); + + try { + + List xread = connection.getCluster().xreadGroup(JedisConverters.toBytes(consumer.getGroup()), + JedisConverters.toBytes(consumer.getName()), count, block, readOptions.isNoack(), toStreamOffsets(streams)); + + if (xread == null) { + return Collections.emptyList(); + } + + return StreamConverters.convertToByteRecords(xread); + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xRevRange(byte[], org.springframework.data.domain.Range, org.springframework.data.redis.connection.RedisZSetCommands.Limit) + */ + @Override + public List xRevRange(byte[] key, Range range, RedisZSetCommands.Limit limit) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(range, "Range must not be null!"); + Assert.notNull(limit, "Limit must not be null!"); + + int count = limit.isUnlimited() ? Integer.MAX_VALUE : limit.getCount(); + + try { + return convertToByteRecord(key, connection.getCluster().xrevrange(key, + JedisConverters.toBytes(getUpperValue(range)), JedisConverters.toBytes(getLowerValue(range)), count)); + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xTrim(byte[], long) + */ + @Override + public Long xTrim(byte[] key, long count) { + return xTrim(key, count, false); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xTrim(byte[], long, boolean) + */ + @Override + public Long xTrim(byte[] key, long count, boolean approximateTrimming) { + + Assert.notNull(key, "Key must not be null!"); + + try { + return connection.getCluster().xtrim(key, count, approximateTrimming); + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + + private DataAccessException convertJedisAccessException(Exception ex) { + return connection.convertJedisAccessException(ex); + } + +} diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java index 320f7f833..22ab9a0b3 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java @@ -65,6 +65,7 @@ import org.springframework.util.StringUtils; * @author Mark Paluch * @author Ninad Divadkar * @author Guy Korland + * @author Dengliming */ public class JedisConnection extends AbstractRedisConnection { diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisInvoker.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisInvoker.java index 2925ecda7..581d5bf2f 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisInvoker.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisInvoker.java @@ -363,6 +363,21 @@ class JedisInvoker { it -> pipelineFunction.apply(it, t1, t2, t3, t4, t5, t6)); } + /** + * Compose a invocation pipeline from the {@link ConnectionFunction0} that returns a {@link Collection}-like result + * and return a {@link ManyInvocationSpec} for further composition. + * + * @param function must not be {@literal null}. + */ + , E> ManyInvocationSpec fromMany(ConnectionFunction0 function) { + + Assert.notNull(function, "ConnectionFunction must not be null!"); + + return fromMany(function, connection -> { + throw new UnsupportedOperationException("Operation not supported in pipelining/transaction mode"); + }); + } + /** * Compose a invocation pipeline from the {@link ConnectionFunction0} that returns a {@link Collection}-like result * and return a {@link ManyInvocationSpec} for further composition. diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisStreamCommands.java index 3122f834b..1e25c6b89 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisStreamCommands.java @@ -15,13 +15,16 @@ */ package org.springframework.data.redis.connection.jedis; -import static org.springframework.data.redis.connection.jedis.StreamConverters.convertToByteRecord; +import redis.clients.jedis.BinaryJedis; +import redis.clients.jedis.BuilderFactory; +import redis.clients.jedis.MultiKeyPipelineBase; +import redis.clients.jedis.StreamConsumersInfo; +import redis.clients.jedis.StreamGroupInfo; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.List; -import java.util.Map; -import java.util.stream.Collectors; import org.springframework.data.domain.Range; import org.springframework.data.redis.connection.RedisStreamCommands; @@ -38,12 +41,6 @@ import org.springframework.data.redis.connection.stream.StreamOffset; import org.springframework.data.redis.connection.stream.StreamReadOptions; import org.springframework.util.Assert; -import redis.clients.jedis.BinaryJedis; -import redis.clients.jedis.BuilderFactory; -import redis.clients.jedis.MultiKeyPipelineBase; -import redis.clients.jedis.StreamConsumersInfo; -import redis.clients.jedis.StreamGroupInfo; - /** * @author Dengliming * @since 2.3 @@ -62,12 +59,13 @@ class JedisStreamCommands implements RedisStreamCommands { */ @Override public Long xAck(byte[] key, String group, RecordId... recordIds) { + Assert.notNull(key, "Key must not be null!"); Assert.hasText(group, "Group name must not be null or empty!"); Assert.notNull(recordIds, "recordIds must not be null!"); return connection.invoke().just(BinaryJedis::xack, MultiKeyPipelineBase::xack, key, JedisConverters.toBytes(group), - entryIdsToBytes(recordIds)); + StreamConverters.entryIdsToBytes(Arrays.asList(recordIds))); } /* @@ -76,11 +74,12 @@ class JedisStreamCommands implements RedisStreamCommands { */ @Override public RecordId xAdd(MapRecord record, XAddOptions options) { + Assert.notNull(record, "Record must not be null!"); Assert.notNull(record.getStream(), "Stream must not be null!"); byte[] id = JedisConverters.toBytes(record.getId().getValue()); - Long maxLength = Long.MAX_VALUE; + long maxLength = Long.MAX_VALUE; if (options.hasMaxlen()) { maxLength = options.getMaxlen(); } @@ -104,21 +103,22 @@ class JedisStreamCommands implements RedisStreamCommands { */ @Override public List xClaim(byte[] key, String group, String newOwner, XClaimOptions options) { + Assert.notNull(key, "Key must not be null!"); Assert.notNull(group, "Group must not be null!"); Assert.notNull(newOwner, "NewOwner must not be null!"); - final long minIdleTime = options.getMinIdleTime() == null ? -1L : options.getMinIdleTime().toMillis(); - final int retryCount = options.getRetryCount() == null ? -1 : options.getRetryCount().intValue(); - final long unixTime = options.getUnixTime() == null ? -1L : options.getUnixTime().toEpochMilli(); + long minIdleTime = options.getMinIdleTime() == null ? -1L : options.getMinIdleTime().toMillis(); + int retryCount = options.getRetryCount() == null ? -1 : options.getRetryCount().intValue(); + long unixTime = options.getUnixTime() == null ? -1L : options.getUnixTime().toEpochMilli(); return connection.invoke() .from( it -> it.xclaim(key, JedisConverters.toBytes(group), JedisConverters.toBytes(newOwner), minIdleTime, - unixTime, retryCount, options.isForce(), entryIdsToBytes(options.getIds())), + unixTime, retryCount, options.isForce(), StreamConverters.entryIdsToBytes(options.getIds())), it -> it.xclaim(key, JedisConverters.toBytes(group), JedisConverters.toBytes(newOwner), minIdleTime, - unixTime, retryCount, options.isForce(), entryIdsToBytes(options.getIds()))) - .get(r -> convertToByteRecord(key, r)); + unixTime, retryCount, options.isForce(), StreamConverters.entryIdsToBytes(options.getIds()))) + .get(r -> StreamConverters.convertToByteRecord(key, r)); } /* @@ -127,10 +127,12 @@ class JedisStreamCommands implements RedisStreamCommands { */ @Override public Long xDel(byte[] key, RecordId... recordIds) { + Assert.notNull(key, "Key must not be null!"); Assert.notNull(recordIds, "recordIds must not be null!"); - return connection.invoke().just(BinaryJedis::xdel, MultiKeyPipelineBase::xdel, key, entryIdsToBytes(recordIds)); + return connection.invoke().just(BinaryJedis::xdel, MultiKeyPipelineBase::xdel, key, + StreamConverters.entryIdsToBytes(Arrays.asList(recordIds))); } /* @@ -148,6 +150,7 @@ class JedisStreamCommands implements RedisStreamCommands { */ @Override public String xGroupCreate(byte[] key, String groupName, ReadOffset readOffset, boolean mkStream) { + Assert.notNull(key, "Key must not be null!"); Assert.hasText(groupName, "Group name must not be null or empty!"); Assert.notNull(readOffset, "ReadOffset must not be null!"); @@ -162,6 +165,7 @@ class JedisStreamCommands implements RedisStreamCommands { */ @Override public Boolean xGroupDelConsumer(byte[] key, Consumer consumer) { + Assert.notNull(key, "Key must not be null!"); Assert.notNull(consumer, "Consumer must not be null!"); @@ -175,6 +179,7 @@ class JedisStreamCommands implements RedisStreamCommands { */ @Override public Boolean xGroupDestroy(byte[] key, String groupName) { + Assert.notNull(key, "Key must not be null!"); Assert.hasText(groupName, "Group name must not be null or empty!"); @@ -189,6 +194,7 @@ class JedisStreamCommands implements RedisStreamCommands { */ @Override public StreamInfo.XInfoStream xInfo(byte[] key) { + Assert.notNull(key, "Key must not be null!"); if (isQueueing() || isPipelined()) { @@ -197,7 +203,7 @@ class JedisStreamCommands implements RedisStreamCommands { return connection.invoke().just(it -> { redis.clients.jedis.StreamInfo streamInfo = it.xinfoStream(key); - return StreamInfo.XInfoStream.fromList(mapToList(streamInfo.getStreamInfo())); + return StreamInfo.XInfoStream.fromList(StreamConverters.mapToList(streamInfo.getStreamInfo())); }); } @@ -207,6 +213,7 @@ class JedisStreamCommands implements RedisStreamCommands { */ @Override public StreamInfo.XInfoGroups xInfoGroups(byte[] key) { + Assert.notNull(key, "Key must not be null!"); if (isQueueing() || isPipelined()) { @@ -216,7 +223,8 @@ class JedisStreamCommands implements RedisStreamCommands { return connection.invoke().just(it -> { List streamGroupInfos = it.xinfoGroup(key); List sources = new ArrayList<>(); - streamGroupInfos.forEach(streamGroupInfo -> sources.add(mapToList(streamGroupInfo.getGroupInfo()))); + streamGroupInfos + .forEach(streamGroupInfo -> sources.add(StreamConverters.mapToList(streamGroupInfo.getGroupInfo()))); return StreamInfo.XInfoGroups.fromList(sources); }); } @@ -227,6 +235,7 @@ class JedisStreamCommands implements RedisStreamCommands { */ @Override public StreamInfo.XInfoConsumers xInfoConsumers(byte[] key, String groupName) { + Assert.notNull(key, "Key must not be null!"); Assert.hasText(groupName, "Group name must not be null or empty!"); @@ -238,7 +247,8 @@ class JedisStreamCommands implements RedisStreamCommands { List streamConsumersInfos = it.xinfoConsumers(key, JedisConverters.toBytes(groupName)); List sources = new ArrayList<>(); streamConsumersInfos - .forEach(streamConsumersInfo -> sources.add(mapToList(streamConsumersInfo.getConsumerInfo()))); + .forEach( + streamConsumersInfo -> sources.add(StreamConverters.mapToList(streamConsumersInfo.getConsumerInfo()))); return StreamInfo.XInfoConsumers.fromList(groupName, sources); }); } @@ -249,6 +259,7 @@ class JedisStreamCommands implements RedisStreamCommands { */ @Override public Long xLen(byte[] key) { + Assert.notNull(key, "Key must not be null!"); return connection.invoke().just(BinaryJedis::xlen, MultiKeyPipelineBase::xlen, key); @@ -269,6 +280,7 @@ class JedisStreamCommands implements RedisStreamCommands { */ @Override public PendingMessages xPending(byte[] key, String groupName, XPendingOptions options) { + Assert.notNull(key, "Key must not be null!"); Assert.notNull(groupName, "GroupName must not be null!"); @@ -277,9 +289,10 @@ class JedisStreamCommands implements RedisStreamCommands { return connection.invoke().from((it, t1, t2, t3, t4, t5, t6) -> { Object r = it.xpending(t1, t2, t3, t4, t5, t6); + return BuilderFactory.STREAM_PENDING_ENTRY_LIST.build(r); - }, MultiKeyPipelineBase::xpending, key, group, JedisConverters.toBytes(getLowerValue(range)), - JedisConverters.toBytes(getUpperValue(range)), options.getCount().intValue(), + }, MultiKeyPipelineBase::xpending, key, group, JedisConverters.toBytes(StreamConverters.getLowerValue(range)), + JedisConverters.toBytes(StreamConverters.getUpperValue(range)), options.getCount().intValue(), JedisConverters.toBytes(options.getConsumerName())) .get(r -> StreamConverters.toPendingMessages(groupName, range, r)); } @@ -290,6 +303,7 @@ class JedisStreamCommands implements RedisStreamCommands { */ @Override public List xRange(byte[] key, Range range, RedisZSetCommands.Limit limit) { + Assert.notNull(key, "Key must not be null!"); Assert.notNull(range, "Range must not be null!"); Assert.notNull(limit, "Limit must not be null!"); @@ -298,11 +312,11 @@ class JedisStreamCommands implements RedisStreamCommands { return connection.invoke() .from( - it -> it.xrange(key, JedisConverters.toBytes(getLowerValue(range)), - JedisConverters.toBytes(getUpperValue(range)), count), - it -> it.xrange(key, JedisConverters.toBytes(getLowerValue(range)), - JedisConverters.toBytes(getUpperValue(range)), count)) - .get(r -> convertToByteRecord(key, r)); + it -> it.xrange(key, JedisConverters.toBytes(StreamConverters.getLowerValue(range)), + JedisConverters.toBytes(StreamConverters.getUpperValue(range)), count), + it -> it.xrange(key, JedisConverters.toBytes(StreamConverters.getLowerValue(range)), + JedisConverters.toBytes(StreamConverters.getUpperValue(range)), count)) + .get(r -> StreamConverters.convertToByteRecord(key, r)); } /* @@ -311,6 +325,7 @@ class JedisStreamCommands implements RedisStreamCommands { */ @Override public List xRead(StreamReadOptions readOptions, StreamOffset... streams) { + Assert.notNull(readOptions, "StreamReadOptions must not be null!"); Assert.notNull(streams, "StreamOffsets must not be null!"); @@ -318,13 +333,11 @@ class JedisStreamCommands implements RedisStreamCommands { throw new UnsupportedOperationException("'XREAD' cannot be called in pipeline / transaction mode."); } - final long block = readOptions.getBlock() == null ? -1L : readOptions.getBlock(); - final int count = readOptions.getCount() != null ? readOptions.getCount().intValue() : Integer.MAX_VALUE; + long block = readOptions.getBlock() == null ? -1L : readOptions.getBlock(); + int count = readOptions.getCount() != null ? readOptions.getCount().intValue() : Integer.MAX_VALUE; - return connection.invoke().just(it -> { - List streamsEntries = it.xread(count, block, toStreamOffsets(streams)); - return convertToByteRecord(streamsEntries); - }); + return connection.invoke().from(it -> it.xread(count, block, StreamConverters.toStreamOffsets(streams))) + .getOrElse(StreamConverters::convertToByteRecords, Collections::emptyList); } /* @@ -334,6 +347,7 @@ class JedisStreamCommands implements RedisStreamCommands { @Override public List xReadGroup(Consumer consumer, StreamReadOptions readOptions, StreamOffset... streams) { + Assert.notNull(consumer, "Consumer must not be null!"); Assert.notNull(readOptions, "StreamReadOptions must not be null!"); Assert.notNull(streams, "StreamOffsets must not be null!"); @@ -342,14 +356,14 @@ class JedisStreamCommands implements RedisStreamCommands { throw new UnsupportedOperationException("'XREADGROUP' cannot be called in pipeline / transaction mode."); } - final long block = readOptions.getBlock() == null ? -1L : readOptions.getBlock(); - final int count = readOptions.getCount() == null ? -1 : readOptions.getCount().intValue(); + long block = readOptions.getBlock() == null ? -1L : readOptions.getBlock(); + int count = readOptions.getCount() == null ? -1 : readOptions.getCount().intValue(); - return connection.invoke().just(it -> { - List streamsEntries = it.xreadGroup(JedisConverters.toBytes(consumer.getGroup()), - JedisConverters.toBytes(consumer.getName()), count, block, readOptions.isNoack(), toStreamOffsets(streams)); - return convertToByteRecord(streamsEntries); - }); + return connection.invoke().from(it -> { + + return it.xreadGroup(JedisConverters.toBytes(consumer.getGroup()), JedisConverters.toBytes(consumer.getName()), + count, block, readOptions.isNoack(), StreamConverters.toStreamOffsets(streams)); + }).getOrElse(StreamConverters::convertToByteRecords, Collections::emptyList); } /* @@ -358,6 +372,7 @@ class JedisStreamCommands implements RedisStreamCommands { */ @Override public List xRevRange(byte[] key, Range range, RedisZSetCommands.Limit limit) { + Assert.notNull(key, "Key must not be null!"); Assert.notNull(range, "Range must not be null!"); Assert.notNull(limit, "Limit must not be null!"); @@ -365,8 +380,9 @@ class JedisStreamCommands implements RedisStreamCommands { int count = limit.isUnlimited() ? Integer.MAX_VALUE : limit.getCount(); return connection.invoke() .from(BinaryJedis::xrevrange, MultiKeyPipelineBase::xrevrange, key, - JedisConverters.toBytes(getUpperValue(range)), JedisConverters.toBytes(getLowerValue(range)), count) - .get(it -> convertToByteRecord(key, it)); + JedisConverters.toBytes(StreamConverters.getUpperValue(range)), + JedisConverters.toBytes(StreamConverters.getLowerValue(range)), count) + .get(it -> StreamConverters.convertToByteRecord(key, it)); } /* @@ -384,6 +400,7 @@ class JedisStreamCommands implements RedisStreamCommands { */ @Override public Long xTrim(byte[] key, long count, boolean approximateTrimming) { + Assert.notNull(key, "Key must not be null!"); return connection.invoke().just(BinaryJedis::xtrim, MultiKeyPipelineBase::xtrim, key, count, approximateTrimming); @@ -397,57 +414,4 @@ class JedisStreamCommands implements RedisStreamCommands { return connection.isQueueing(); } - private byte[][] entryIdsToBytes(RecordId[] recordIds) { - - final byte[][] bids = new byte[recordIds.length][]; - for (int i = 0; i < recordIds.length; ++i) { - RecordId id = recordIds[i]; - bids[i] = JedisConverters.toBytes(id.getValue()); - } - - return bids; - } - - private byte[][] entryIdsToBytes(List recordIds) { - - final byte[][] bids = new byte[recordIds.size()][]; - for (int i = 0; i < recordIds.size(); ++i) { - RecordId id = recordIds.get(i); - bids[i] = JedisConverters.toBytes(id.getValue()); - } - - return bids; - } - - private String getLowerValue(Range range) { - - if (range.getLowerBound().equals(Range.Bound.unbounded())) { - return "-"; - } - - return range.getLowerBound().getValue().orElse("-"); - } - - private String getUpperValue(Range range) { - - if (range.getUpperBound().equals(Range.Bound.unbounded())) { - return "+"; - } - - return range.getUpperBound().getValue().orElse("+"); - } - - private List mapToList(Map map) { - List sources = new ArrayList<>(map.size() * 2); - map.forEach((k, v) -> { - sources.add(k); - sources.add(v); - }); - return sources; - } - - private Map toStreamOffsets(StreamOffset... streams) { - return Arrays.stream(streams) - .collect(Collectors.toMap(k -> k.getKey(), v -> JedisConverters.toBytes(v.getOffset().getOffset()))); - } } diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/StreamConverters.java b/src/main/java/org/springframework/data/redis/connection/jedis/StreamConverters.java index 1c259c1b2..565d0fd0d 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/StreamConverters.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/StreamConverters.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2021 the original author or authors. + * Copyright 2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -15,25 +15,29 @@ */ package org.springframework.data.redis.connection.jedis; +import redis.clients.jedis.StreamEntry; +import redis.clients.jedis.StreamEntryID; +import redis.clients.jedis.StreamPendingEntry; +import redis.clients.jedis.util.SafeEncoder; + import java.time.Duration; import java.util.ArrayList; -import java.util.Collections; +import java.util.Arrays; import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.stream.Collectors; +import org.springframework.data.domain.Range; import org.springframework.data.redis.connection.stream.ByteRecord; import org.springframework.data.redis.connection.stream.Consumer; import org.springframework.data.redis.connection.stream.PendingMessage; import org.springframework.data.redis.connection.stream.PendingMessages; import org.springframework.data.redis.connection.stream.RecordId; +import org.springframework.data.redis.connection.stream.StreamOffset; import org.springframework.data.redis.connection.stream.StreamRecords; -import redis.clients.jedis.StreamPendingEntry; -import redis.clients.jedis.util.SafeEncoder; - /** * Converters for Redis Stream-specific types. *

@@ -41,26 +45,83 @@ import redis.clients.jedis.util.SafeEncoder; * serialization/deserialization happens here). * * @author dengliming + * @author Mark Paluch * @since 2.3 */ -@SuppressWarnings({ "rawtypes" }) class StreamConverters { - static final List convertToByteRecord(byte[] key, Object source) { - if (null == source) { - return Collections.emptyList(); + static byte[][] entryIdsToBytes(List recordIds) { + + byte[][] target = new byte[recordIds.size()][]; + + for (int i = 0; i < recordIds.size(); ++i) { + RecordId id = recordIds.get(i); + target[i] = JedisConverters.toBytes(id.getValue()); } + + return target; + } + + static String getLowerValue(Range range) { + + if (range.getLowerBound().equals(Range.Bound.unbounded())) { + return "-"; + } + + return range.getLowerBound().getValue().orElse("-"); + } + + static String getUpperValue(Range range) { + + if (range.getUpperBound().equals(Range.Bound.unbounded())) { + return "+"; + } + + return range.getUpperBound().getValue().orElse("+"); + } + + static List mapToList(Map map) { + + List sources = new ArrayList<>(map.size() * 2); + map.forEach((k, v) -> { + sources.add(k); + + if (v instanceof StreamEntryID) { + sources.add(v.toString()); + } else if (v instanceof StreamEntry) { + List entries = new ArrayList<>(2); + StreamEntry streamEntry = (StreamEntry) v; + entries.add(streamEntry.getID().toString()); + entries.add(streamEntry.getFields()); + sources.add(entries); + } else { + sources.add(v); + } + }); + return sources; + } + + static Map toStreamOffsets(StreamOffset[] streams) { + return Arrays.stream(streams) + .collect(Collectors.toMap(StreamOffset::getKey, v -> JedisConverters.toBytes(v.getOffset().getOffset()))); + } + + static List convertToByteRecord(byte[] key, Object source) { + List> objectList = (List>) source; List result = new ArrayList<>(objectList.size() / 2); + if (objectList.isEmpty()) { return result; } for (List res : objectList) { + if (res == null) { result.add(null); continue; } + String entryIdString = SafeEncoder.encode((byte[]) res.get(0)); List hash = (List) res.get(1); @@ -75,15 +136,15 @@ class StreamConverters { return result; } - static final List convertToByteRecord(List sources) { - if (sources == null) { - return Collections.emptyList(); - } - List result = new ArrayList<>(); - for (Object streamObj : sources) { - List stream = (List) streamObj; + static List convertToByteRecords(List sources) { + + List result = new ArrayList<>(sources.size() / 2); + + for (Object source : sources) { + List stream = (List) source; result.addAll(convertToByteRecord((byte[]) stream.get(0), stream.get(1))); } + return result; } @@ -98,10 +159,6 @@ class StreamConverters { static org.springframework.data.redis.connection.stream.PendingMessages toPendingMessages(String groupName, org.springframework.data.domain.Range range, List response) { - if (null == response) { - return null; - } - List messages = response.stream() .map(streamPendingEntry -> new PendingMessage(RecordId.of(streamPendingEntry.getID().toString()), Consumer.from(groupName, streamPendingEntry.getConsumerName()), diff --git a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java index 536241099..61d553aad 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -3073,7 +3073,6 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-864 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) void xAddShouldCreateStream() { actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); @@ -3087,8 +3086,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-864 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) - void xReadShouldReadMessage() { + public void xReadShouldReadMessage() { actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); actual.add(connection.xReadAsString(StreamOffset.create(KEY_1, ReadOffset.from("0")))); @@ -3103,8 +3101,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-864 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) - void xReadGroupShouldReadMessage() { + public void xReadGroupShouldReadMessage() { actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group")); @@ -3125,8 +3122,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-864 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) - void xGroupCreateShouldWorkWithAndWithoutExistingStream() { + public void xGroupCreateShouldWorkWithAndWithoutExistingStream() { actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group", true)); actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); @@ -3148,7 +3144,6 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-864 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) void xRangeShouldReportMessages() { actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); @@ -3169,8 +3164,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-864 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) - void xRevRangeShouldReportMessages() { + public void xRevRangeShouldReportMessages() { actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_3, VALUE_3))); @@ -3191,8 +3185,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-1207 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) - void xRevRangeShouldWorkWithBoundedRange() { + public void xRevRangeShouldWorkWithBoundedRange() { actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_3, VALUE_3))); @@ -3213,7 +3206,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-1084 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) + @EnabledOnRedisDriver(RedisDriver.LETTUCE) void xPendingShouldLoadOverviewCorrectly() { actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); @@ -3235,7 +3228,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-1084 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) + @EnabledOnRedisDriver(RedisDriver.LETTUCE) void xPendingShouldLoadEmptyOverviewCorrectly() { actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); @@ -3254,8 +3247,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-1084 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) - void xPendingShouldLoadPendingMessages() { + public void xPendingShouldLoadPendingMessages() { actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group")); @@ -3277,8 +3269,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-1207 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) - void xPendingShouldWorkWithBoundedRange() { + public void xPendingShouldWorkWithBoundedRange() { actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group")); @@ -3300,8 +3291,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-1084 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) - void xPendingShouldLoadPendingMessagesForConsumer() { + public void xPendingShouldLoadPendingMessagesForConsumer() { actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group")); @@ -3324,8 +3314,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-1084 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) - void xPendingShouldLoadPendingMessagesForNonExistingConsumer() { + public void xPendingShouldLoadPendingMessagesForNonExistingConsumer() { actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group")); @@ -3344,7 +3333,6 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-1084 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) void xPendingShouldLoadEmptyPendingMessages() { actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); @@ -3361,7 +3349,6 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-1084 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) public void xClaim() throws InterruptedException { actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); @@ -3382,8 +3369,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-1119 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) - void xinfo() { + public void xinfo() { actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group-without-stream", true)); actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); @@ -3411,8 +3397,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-1119 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) - void xinfoNoGroup() { + public void xinfoNoGroup() { actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_3, VALUE_3))); @@ -3436,8 +3421,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-1119 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) - void xinfoGroups() { + public void xinfoGroups() { actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_3, VALUE_3))); @@ -3461,8 +3445,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-1119 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) - void xinfoGroupsNoGroup() { + public void xinfoGroupsNoGroup() { actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_3, VALUE_3))); @@ -3478,8 +3461,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-1119 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) - void xinfoGroupsNoConsumer() { + public void xinfoGroupsNoConsumer() { actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_3, VALUE_3))); @@ -3501,8 +3483,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-1119 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) - void xinfoConsumers() { + public void xinfoConsumers() { actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_3, VALUE_3))); @@ -3525,8 +3506,7 @@ public abstract class AbstractConnectionIntegrationTests { @Test // DATAREDIS-1119 @EnabledOnCommand("XADD") - @EnabledOnRedisDriver({ RedisDriver.LETTUCE }) - void xinfoConsumersNoConsumer() { + public void xinfoConsumersNoConsumer() { actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_3, VALUE_3))); diff --git a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionPipelineIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionPipelineIntegrationTests.java index 287f7de58..8710ed5e7 100644 --- a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionPipelineIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionPipelineIntegrationTests.java @@ -31,6 +31,7 @@ import org.springframework.data.redis.SettingsUtils; import org.springframework.data.redis.connection.AbstractConnectionPipelineIntegrationTests; import org.springframework.data.redis.connection.DefaultStringRedisConnection; import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.test.condition.EnabledOnCommand; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit.jupiter.SpringExtension; @@ -221,6 +222,108 @@ public class JedisConnectionPipelineIntegrationTests extends AbstractConnectionP assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::clientSetNameWorksCorrectly); } + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xReadShouldReadMessage() { + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::xReadShouldReadMessage); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xReadGroupShouldReadMessage() { + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::xReadGroupShouldReadMessage); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xGroupCreateShouldWorkWithAndWithoutExistingStream() { + assertThatExceptionOfType(UnsupportedOperationException.class) + .isThrownBy(super::xGroupCreateShouldWorkWithAndWithoutExistingStream); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xPendingShouldLoadPendingMessages() { + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::xPendingShouldLoadPendingMessages); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xPendingShouldWorkWithBoundedRange() { + assertThatExceptionOfType(UnsupportedOperationException.class) + .isThrownBy(super::xPendingShouldWorkWithBoundedRange); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xPendingShouldLoadPendingMessagesForConsumer() { + assertThatExceptionOfType(UnsupportedOperationException.class) + .isThrownBy(super::xPendingShouldLoadPendingMessagesForConsumer); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xPendingShouldLoadPendingMessagesForNonExistingConsumer() { + assertThatExceptionOfType(UnsupportedOperationException.class) + .isThrownBy(super::xPendingShouldLoadPendingMessagesForNonExistingConsumer); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xinfo() { + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::xinfo); + } + + @Test + @EnabledOnCommand("XADD") + @Override + public void xinfoNoGroup() { + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::xinfoNoGroup); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xinfoGroups() { + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::xinfoGroups); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xinfoGroupsNoGroup() { + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::xinfoGroupsNoGroup); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xinfoGroupsNoConsumer() { + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::xinfoGroupsNoConsumer); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xinfoConsumers() { + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::xinfoConsumers); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xinfoConsumersNoConsumer() { + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::xinfoConsumersNoConsumer); + } + @Test @Override // DATAREDIS-268 diff --git a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionPipelineTxIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionPipelineTxIntegrationTests.java index 10ca94ac0..887d98554 100644 --- a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionPipelineTxIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionPipelineTxIntegrationTests.java @@ -55,4 +55,5 @@ public class JedisConnectionPipelineTxIntegrationTests extends JedisConnectionTr @Test @Disabled public void testListClientsContainsAtLeastOneElement() {} + } diff --git a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionTransactionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionTransactionIntegrationTests.java index 53b54eb56..da4abbb2b 100644 --- a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionTransactionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionTransactionIntegrationTests.java @@ -24,6 +24,7 @@ import org.junit.jupiter.api.extension.ExtendWith; import org.springframework.data.redis.connection.AbstractConnectionTransactionIntegrationTests; import org.springframework.data.redis.connection.ReturnType; +import org.springframework.data.redis.test.condition.EnabledOnCommand; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit.jupiter.SpringExtension; @@ -187,4 +188,107 @@ public class JedisConnectionTransactionIntegrationTests extends AbstractConnecti assertThatExceptionOfType(UnsupportedOperationException.class) .isThrownBy(super::testListClientsContainsAtLeastOneElement); } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xReadShouldReadMessage() { + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::xReadShouldReadMessage); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xReadGroupShouldReadMessage() { + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::xReadGroupShouldReadMessage); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xGroupCreateShouldWorkWithAndWithoutExistingStream() { + assertThatExceptionOfType(UnsupportedOperationException.class) + .isThrownBy(super::xGroupCreateShouldWorkWithAndWithoutExistingStream); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xPendingShouldLoadPendingMessages() { + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::xPendingShouldLoadPendingMessages); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xPendingShouldWorkWithBoundedRange() { + assertThatExceptionOfType(UnsupportedOperationException.class) + .isThrownBy(super::xPendingShouldWorkWithBoundedRange); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xPendingShouldLoadPendingMessagesForConsumer() { + assertThatExceptionOfType(UnsupportedOperationException.class) + .isThrownBy(super::xPendingShouldLoadPendingMessagesForConsumer); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xPendingShouldLoadPendingMessagesForNonExistingConsumer() { + assertThatExceptionOfType(UnsupportedOperationException.class) + .isThrownBy(super::xPendingShouldLoadPendingMessagesForNonExistingConsumer); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xinfo() { + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::xinfo); + } + + @Test + @EnabledOnCommand("XADD") + @Override + public void xinfoNoGroup() { + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::xinfoNoGroup); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xinfoGroups() { + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::xinfoGroups); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xinfoGroupsNoGroup() { + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::xinfoGroupsNoGroup); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xinfoGroupsNoConsumer() { + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::xinfoGroupsNoConsumer); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xinfoConsumers() { + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::xinfoConsumers); + } + + @Test // GH-1711 + @EnabledOnCommand("XADD") + @Override + public void xinfoConsumersNoConsumer() { + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(super::xinfoConsumersNoConsumer); + } + } diff --git a/src/test/java/org/springframework/data/redis/core/AbstractOperationsTestParams.java b/src/test/java/org/springframework/data/redis/core/AbstractOperationsTestParams.java index f76732748..b9ff4cc68 100644 --- a/src/test/java/org/springframework/data/redis/core/AbstractOperationsTestParams.java +++ b/src/test/java/org/springframework/data/redis/core/AbstractOperationsTestParams.java @@ -25,8 +25,8 @@ import org.springframework.data.redis.Person; import org.springframework.data.redis.PersonObjectFactory; import org.springframework.data.redis.RawObjectFactory; import org.springframework.data.redis.StringObjectFactory; -import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; -import org.springframework.data.redis.connection.lettuce.extension.LettuceConnectionFactoryExtension; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.connection.jedis.extension.JedisConnectionFactoryExtension; import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer; import org.springframework.data.redis.serializer.GenericToStringSerializer; import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer; @@ -47,6 +47,11 @@ abstract public class AbstractOperationsTestParams { // DATAREDIS-241 public static Collection testParams() { + return testParams(JedisConnectionFactoryExtension.getConnectionFactory(RedisStanalone.class)); + } + + // DATAREDIS-241 + public static Collection testParams(RedisConnectionFactory connectionFactory) { ObjectFactory stringFactory = new StringObjectFactory(); ObjectFactory longFactory = new LongObjectFactory(); @@ -54,54 +59,52 @@ abstract public class AbstractOperationsTestParams { ObjectFactory rawFactory = new RawObjectFactory(); ObjectFactory personFactory = new PersonObjectFactory(); - LettuceConnectionFactory lettuceConnectionFactory = LettuceConnectionFactoryExtension - .getConnectionFactory(RedisStanalone.class); RedisTemplate stringTemplate = new StringRedisTemplate(); - stringTemplate.setConnectionFactory(lettuceConnectionFactory); + stringTemplate.setConnectionFactory(connectionFactory); stringTemplate.afterPropertiesSet(); RedisTemplate longTemplate = new RedisTemplate<>(); longTemplate.setKeySerializer(StringRedisSerializer.UTF_8); longTemplate.setValueSerializer(new GenericToStringSerializer<>(Long.class)); - longTemplate.setConnectionFactory(lettuceConnectionFactory); + longTemplate.setConnectionFactory(connectionFactory); longTemplate.afterPropertiesSet(); RedisTemplate doubleTemplate = new RedisTemplate<>(); doubleTemplate.setKeySerializer(StringRedisSerializer.UTF_8); doubleTemplate.setValueSerializer(new GenericToStringSerializer<>(Double.class)); - doubleTemplate.setConnectionFactory(lettuceConnectionFactory); + doubleTemplate.setConnectionFactory(connectionFactory); doubleTemplate.afterPropertiesSet(); RedisTemplate rawTemplate = new RedisTemplate<>(); rawTemplate.setEnableDefaultSerializer(false); - rawTemplate.setConnectionFactory(lettuceConnectionFactory); + rawTemplate.setConnectionFactory(connectionFactory); rawTemplate.afterPropertiesSet(); RedisTemplate personTemplate = new RedisTemplate<>(); - personTemplate.setConnectionFactory(lettuceConnectionFactory); + personTemplate.setConnectionFactory(connectionFactory); personTemplate.afterPropertiesSet(); OxmSerializer serializer = XstreamOxmSerializerSingleton.getInstance(); RedisTemplate xstreamStringTemplate = new RedisTemplate<>(); - xstreamStringTemplate.setConnectionFactory(lettuceConnectionFactory); + xstreamStringTemplate.setConnectionFactory(connectionFactory); xstreamStringTemplate.setDefaultSerializer(serializer); xstreamStringTemplate.afterPropertiesSet(); RedisTemplate xstreamPersonTemplate = new RedisTemplate<>(); - xstreamPersonTemplate.setConnectionFactory(lettuceConnectionFactory); + xstreamPersonTemplate.setConnectionFactory(connectionFactory); xstreamPersonTemplate.setValueSerializer(serializer); xstreamPersonTemplate.afterPropertiesSet(); Jackson2JsonRedisSerializer jackson2JsonSerializer = new Jackson2JsonRedisSerializer<>(Person.class); RedisTemplate jackson2JsonPersonTemplate = new RedisTemplate<>(); - jackson2JsonPersonTemplate.setConnectionFactory(lettuceConnectionFactory); + jackson2JsonPersonTemplate.setConnectionFactory(connectionFactory); jackson2JsonPersonTemplate.setValueSerializer(jackson2JsonSerializer); jackson2JsonPersonTemplate.afterPropertiesSet(); GenericJackson2JsonRedisSerializer genericJackson2JsonSerializer = new GenericJackson2JsonRedisSerializer(); RedisTemplate genericJackson2JsonPersonTemplate = new RedisTemplate<>(); - genericJackson2JsonPersonTemplate.setConnectionFactory(lettuceConnectionFactory); + genericJackson2JsonPersonTemplate.setConnectionFactory(connectionFactory); genericJackson2JsonPersonTemplate.setValueSerializer(genericJackson2JsonSerializer); genericJackson2JsonPersonTemplate.afterPropertiesSet(); diff --git a/src/test/java/org/springframework/data/redis/core/DefaultStreamOperationsIntegrationTests.java b/src/test/java/org/springframework/data/redis/core/DefaultStreamOperationsIntegrationTests.java index 70fd6ddc0..6ea5f9456 100644 --- a/src/test/java/org/springframework/data/redis/core/DefaultStreamOperationsIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/core/DefaultStreamOperationsIntegrationTests.java @@ -18,6 +18,7 @@ package org.springframework.data.redis.core; import static org.assertj.core.api.Assertions.*; import static org.assertj.core.api.Assumptions.*; +import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.List; @@ -30,7 +31,9 @@ import org.springframework.data.redis.ObjectFactory; import org.springframework.data.redis.Person; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.connection.RedisZSetCommands.Limit; +import org.springframework.data.redis.connection.jedis.extension.JedisConnectionFactoryExtension; import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; +import org.springframework.data.redis.connection.lettuce.extension.LettuceConnectionFactoryExtension; import org.springframework.data.redis.connection.stream.Consumer; import org.springframework.data.redis.connection.stream.MapRecord; import org.springframework.data.redis.connection.stream.ObjectRecord; @@ -43,6 +46,8 @@ import org.springframework.data.redis.connection.stream.StreamReadOptions; import org.springframework.data.redis.connection.stream.StreamRecords; import org.springframework.data.redis.test.condition.EnabledOnCommand; import org.springframework.data.redis.test.condition.EnabledOnRedisDriver; +import org.springframework.data.redis.test.extension.RedisCluster; +import org.springframework.data.redis.test.extension.RedisStanalone; import org.springframework.data.redis.test.extension.parametrized.MethodSource; import org.springframework.data.redis.test.extension.parametrized.ParameterizedRedisTest; @@ -67,10 +72,6 @@ public class DefaultStreamOperationsIntegrationTests { public DefaultStreamOperationsIntegrationTests(RedisTemplate redisTemplate, ObjectFactory keyFactory, ObjectFactory objectFactory) { - // Currently, only Lettuce supports Redis Streams. - // See https://github.com/xetorthio/jedis/issues/1820 - assumeThat(redisTemplate.getRequiredConnectionFactory()).isInstanceOf(LettuceConnectionFactory.class); - this.redisTemplate = redisTemplate; this.connectionFactory = redisTemplate.getRequiredConnectionFactory(); this.keyFactory = keyFactory; @@ -80,7 +81,21 @@ public class DefaultStreamOperationsIntegrationTests { } public static Collection testParams() { - return AbstractOperationsTestParams.testParams(); + + List params = new ArrayList<>(); + params.addAll(AbstractOperationsTestParams + .testParams(JedisConnectionFactoryExtension.getConnectionFactory(RedisStanalone.class))); + + params.addAll(AbstractOperationsTestParams + .testParams(JedisConnectionFactoryExtension.getConnectionFactory(RedisCluster.class))); + + params.addAll(AbstractOperationsTestParams + .testParams(LettuceConnectionFactoryExtension.getConnectionFactory(RedisStanalone.class))); + + params.addAll(AbstractOperationsTestParams + .testParams(LettuceConnectionFactoryExtension.getConnectionFactory(RedisCluster.class))); + + return params; } @BeforeEach @@ -315,6 +330,8 @@ public class DefaultStreamOperationsIntegrationTests { @ParameterizedRedisTest // DATAREDIS-1084 void pendingShouldReadMessageSummary() { + // XPENDING summary not supported by Jedis + assumeThat(redisTemplate.getRequiredConnectionFactory()).isInstanceOf(LettuceConnectionFactory.class); K key = keyFactory.instance(); HK hashKey = hashKeyFactory.instance(); diff --git a/src/test/java/org/springframework/data/redis/core/JedisStreamOperationsIntegrationTests.java b/src/test/java/org/springframework/data/redis/core/JedisStreamOperationsIntegrationTests.java deleted file mode 100644 index 31e761987..000000000 --- a/src/test/java/org/springframework/data/redis/core/JedisStreamOperationsIntegrationTests.java +++ /dev/null @@ -1,216 +0,0 @@ -/* - * Copyright 2018-2021 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.springframework.data.redis.core; - -import org.junit.jupiter.api.BeforeEach; -import org.springframework.data.domain.Range; -import org.springframework.data.redis.ObjectFactory; -import org.springframework.data.redis.RawObjectFactory; -import org.springframework.data.redis.StringObjectFactory; -import org.springframework.data.redis.connection.RedisConnectionFactory; -import org.springframework.data.redis.connection.RedisZSetCommands; -import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; -import org.springframework.data.redis.connection.jedis.extension.JedisConnectionFactoryExtension; -import org.springframework.data.redis.connection.stream.Consumer; -import org.springframework.data.redis.connection.stream.MapRecord; -import org.springframework.data.redis.connection.stream.PendingMessages; -import org.springframework.data.redis.connection.stream.ReadOffset; -import org.springframework.data.redis.connection.stream.RecordId; -import org.springframework.data.redis.connection.stream.StreamOffset; -import org.springframework.data.redis.test.condition.EnabledOnCommand; -import org.springframework.data.redis.test.extension.RedisStanalone; -import org.springframework.data.redis.test.extension.parametrized.MethodSource; -import org.springframework.data.redis.test.extension.parametrized.ParameterizedRedisTest; - -import java.util.Arrays; -import java.util.Collection; -import java.util.Collections; -import java.util.List; - -import static org.assertj.core.api.Assertions.assertThat; - -/** - * Integration tests for JedisStreamOperations. - * - * @author dengliming - */ -@MethodSource("testParams") -@EnabledOnCommand("XREAD") -public class JedisStreamOperationsIntegrationTests { - - private final RedisConnectionFactory connectionFactory; - private final RedisTemplate redisTemplate; - private final ObjectFactory keyFactory; - private final ObjectFactory hashKeyFactory; - private final ObjectFactory hashValueFactory; - private final StreamOperations streamOps; - - public JedisStreamOperationsIntegrationTests(RedisTemplate redisTemplate, ObjectFactory keyFactory, - ObjectFactory objectFactory) { - this.redisTemplate = redisTemplate; - this.connectionFactory = redisTemplate.getRequiredConnectionFactory(); - this.keyFactory = keyFactory; - this.hashKeyFactory = (ObjectFactory) keyFactory; - this.hashValueFactory = (ObjectFactory) objectFactory; - this.streamOps = redisTemplate.opsForStream(); - } - - public static Collection testParams() { - ObjectFactory stringFactory = new StringObjectFactory(); - ObjectFactory rawFactory = new RawObjectFactory(); - - JedisConnectionFactory jedisConnectionFactory = JedisConnectionFactoryExtension - .getConnectionFactory(RedisStanalone.class); - - RedisTemplate stringTemplate = new StringRedisTemplate(); - stringTemplate.setConnectionFactory(jedisConnectionFactory); - stringTemplate.afterPropertiesSet(); - - RedisTemplate rawTemplate = new RedisTemplate<>(); - rawTemplate.setConnectionFactory(jedisConnectionFactory); - rawTemplate.setEnableDefaultSerializer(false); - rawTemplate.afterPropertiesSet(); - - return Arrays.asList( - new Object[][] { { stringTemplate, stringFactory, stringFactory }, { rawTemplate, rawFactory, rawFactory } }); - } - - @BeforeEach - void before() { - redisTemplate.execute((RedisCallback) connection -> { - connection.flushDb(); - return null; - }); - } - - @ParameterizedRedisTest // DATAREDIS-1140 - void add() { - - K key = keyFactory.instance(); - HK hashKey = hashKeyFactory.instance(); - HV value = hashValueFactory.instance(); - - RecordId messageId = streamOps.add(key, Collections.singletonMap(hashKey, value)); - - List> messages = streamOps.range(key, Range.unbounded()); - - assertThat(messages).hasSize(1); - - MapRecord message = messages.get(0); - - assertThat(message.getId()).isEqualTo(messageId); - assertThat(message.getStream()).isEqualTo(key); - - if (!(key instanceof byte[] || value instanceof byte[])) { - assertThat(message.getValue()).containsEntry(hashKey, value); - } - } - - @ParameterizedRedisTest // DATAREDIS-1140 - void range() { - - K key = keyFactory.instance(); - HK hashKey = hashKeyFactory.instance(); - HV value = hashValueFactory.instance(); - - RecordId messageId1 = streamOps.add(key, Collections.singletonMap(hashKey, value)); - RecordId messageId2 = streamOps.add(key, Collections.singletonMap(hashKey, value)); - - List> messages = streamOps.range(key, - Range.from(Range.Bound.inclusive(messageId1.getValue())).to(Range.Bound.inclusive(messageId2.getValue())), - RedisZSetCommands.Limit.limit().count(1)); - - assertThat(messages).hasSize(1); - - MapRecord message = messages.get(0); - - assertThat(message.getId()).isEqualTo(messageId1); - } - - @ParameterizedRedisTest // DATAREDIS-1140 - void reverseRange() { - - K key = keyFactory.instance(); - HK hashKey = hashKeyFactory.instance(); - HV value = hashValueFactory.instance(); - - RecordId messageId1 = streamOps.add(key, Collections.singletonMap(hashKey, value)); - RecordId messageId2 = streamOps.add(key, Collections.singletonMap(hashKey, value)); - - List> messages = streamOps.reverseRange(key, Range.unbounded()); - - assertThat(messages).hasSize(2).extracting("id").containsSequence(messageId2, messageId1); - } - - @ParameterizedRedisTest // DATAREDIS-1140 - void read() { - - K key = keyFactory.instance(); - HK hashKey = hashKeyFactory.instance(); - HV value = hashValueFactory.instance(); - - RecordId messageId = streamOps.add(key, Collections.singletonMap(hashKey, value)); - streamOps.createGroup(key, ReadOffset.from("0-0"), "my-group"); - - List> messages = streamOps.read(Consumer.from("my-group", "my-consumer"), - StreamOffset.create(key, ReadOffset.lastConsumed())); - - assertThat(messages).hasSize(1); - - MapRecord message = messages.get(0); - - assertThat(message.getId()).isEqualTo(messageId); - assertThat(message.getStream()).isEqualTo(key); - - if (!(key instanceof byte[] || value instanceof byte[])) { - assertThat(message.getValue()).containsEntry(hashKey, value); - } - } - - @ParameterizedRedisTest // DATAREDIS-1140 - void size() { - - K key = keyFactory.instance(); - HK hashKey = hashKeyFactory.instance(); - HV value = hashValueFactory.instance(); - - streamOps.add(key, Collections.singletonMap(hashKey, value)); - assertThat(streamOps.size(key)).isEqualTo(1); - - streamOps.add(key, Collections.singletonMap(hashKey, value)); - assertThat(streamOps.size(key)).isEqualTo(2); - } - - @ParameterizedRedisTest // DATAREDIS-1140 - void pending() { - - K key = keyFactory.instance(); - HK hashKey = hashKeyFactory.instance(); - HV value = hashValueFactory.instance(); - - RecordId messageId = streamOps.add(key, Collections.singletonMap(hashKey, value)); - streamOps.createGroup(key, ReadOffset.from("0-0"), "my-group"); - - streamOps.read(Consumer.from("my-group", "my-consumer"), StreamOffset.create(key, ReadOffset.lastConsumed())); - - PendingMessages pending = streamOps.pending(key, "my-group", Range.unbounded(), 10L); - - assertThat(pending).hasSize(1); - assertThat(pending.get(0).getGroupName()).isEqualTo("my-group"); - assertThat(pending.get(0).getConsumerName()).isEqualTo("my-consumer"); - assertThat(pending.get(0).getTotalDeliveryCount()).isOne(); - } -} diff --git a/src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java b/src/test/java/org/springframework/data/redis/stream/AbstractStreamMessageListenerContainerIntegrationTests.java similarity index 94% rename from src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java rename to src/test/java/org/springframework/data/redis/stream/AbstractStreamMessageListenerContainerIntegrationTests.java index 10108161c..016d163c7 100644 --- a/src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/stream/AbstractStreamMessageListenerContainerIntegrationTests.java @@ -33,15 +33,11 @@ import org.awaitility.Awaitility; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInstance; -import org.junit.jupiter.api.extension.ExtendWith; import org.springframework.core.convert.ConversionFailedException; -import org.springframework.data.redis.SettingsUtils; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisConnectionFactory; -import org.springframework.data.redis.connection.RedisStandaloneConfiguration; import org.springframework.data.redis.connection.lettuce.LettuceConnection; -import org.springframework.data.redis.connection.lettuce.extension.LettuceConnectionFactoryExtension; import org.springframework.data.redis.connection.stream.ByteRecord; import org.springframework.data.redis.connection.stream.Consumer; import org.springframework.data.redis.connection.stream.MapRecord; @@ -62,13 +58,10 @@ import org.springframework.util.NumberUtils; * @author Mark Paluch * @author Christoph Strobl */ -@ExtendWith(LettuceConnectionFactoryExtension.class) @EnabledOnCommand("XREAD") @TestInstance(TestInstance.Lifecycle.PER_CLASS) -public class StreamMessageListenerContainerIntegrationTests { +abstract class AbstractStreamMessageListenerContainerIntegrationTests { - private static final RedisStandaloneConfiguration standaloneConfiguration = new RedisStandaloneConfiguration( - SettingsUtils.getHost(), SettingsUtils.getPort()); private static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(2); private final RedisConnectionFactory connectionFactory; @@ -76,7 +69,7 @@ public class StreamMessageListenerContainerIntegrationTests { private final StreamMessageListenerContainerOptions> containerOptions = StreamMessageListenerContainerOptions .builder().pollTimeout(Duration.ofMillis(100)).build(); - public StreamMessageListenerContainerIntegrationTests(RedisConnectionFactory connectionFactory) { + AbstractStreamMessageListenerContainerIntegrationTests(RedisConnectionFactory connectionFactory) { this.connectionFactory = connectionFactory; this.redisTemplate = new StringRedisTemplate(connectionFactory); this.redisTemplate.afterPropertiesSet(); @@ -400,9 +393,18 @@ public class StreamMessageListenerContainerIntegrationTests { private int getNumberOfPending(String stream, String group) { - String value = ((List) ((LettuceConnection) connectionFactory.getConnection()).execute("XPENDING", - new NestedMultiOutput<>(StringCodec.UTF8), new byte[][] { stream.getBytes(), group.getBytes() })).get(0) - .toString(); + RedisConnection connection = connectionFactory.getConnection(); + + if (connection instanceof LettuceConnection) { + + String value = ((List) ((LettuceConnection) connectionFactory.getConnection()).execute("XPENDING", + new NestedMultiOutput<>(StringCodec.UTF8), new byte[][] { stream.getBytes(), group.getBytes() })).get(0) + .toString(); + return NumberUtils.parseNumber(value, Integer.class); + } + + String value = ((List) connectionFactory.getConnection().execute("XPENDING", stream.getBytes(), group.getBytes())) + .get(0).toString(); return NumberUtils.parseNumber(value, Integer.class); } diff --git a/src/test/java/org/springframework/data/redis/stream/JedisStreamMessageListenerContainerIntegrationTests.java b/src/test/java/org/springframework/data/redis/stream/JedisStreamMessageListenerContainerIntegrationTests.java new file mode 100644 index 000000000..951d6f9eb --- /dev/null +++ b/src/test/java/org/springframework/data/redis/stream/JedisStreamMessageListenerContainerIntegrationTests.java @@ -0,0 +1,40 @@ +/* + * Copyright 2018-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.redis.stream; + +import org.junit.jupiter.api.TestInstance; +import org.junit.jupiter.api.extension.ExtendWith; + +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.connection.jedis.extension.JedisConnectionFactoryExtension; +import org.springframework.data.redis.test.condition.EnabledOnCommand; + +/** + * Integration tests for {@link StreamMessageListenerContainer} using Jedis. + * + * @author Mark Paluch + */ +@ExtendWith(JedisConnectionFactoryExtension.class) +@EnabledOnCommand("XREAD") +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public class JedisStreamMessageListenerContainerIntegrationTests + extends AbstractStreamMessageListenerContainerIntegrationTests { + + public JedisStreamMessageListenerContainerIntegrationTests(RedisConnectionFactory connectionFactory) { + super(connectionFactory); + } + +} diff --git a/src/test/java/org/springframework/data/redis/stream/LettuceStreamMessageListenerContainerIntegrationTests.java b/src/test/java/org/springframework/data/redis/stream/LettuceStreamMessageListenerContainerIntegrationTests.java new file mode 100644 index 000000000..6cdadeb4d --- /dev/null +++ b/src/test/java/org/springframework/data/redis/stream/LettuceStreamMessageListenerContainerIntegrationTests.java @@ -0,0 +1,41 @@ +/* + * Copyright 2018-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.redis.stream; + +import org.junit.jupiter.api.TestInstance; +import org.junit.jupiter.api.extension.ExtendWith; + +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.connection.lettuce.extension.LettuceConnectionFactoryExtension; +import org.springframework.data.redis.test.condition.EnabledOnCommand; + +/** + * Integration tests for {@link StreamMessageListenerContainer} using Lettuce. + * + * @author Mark Paluch + * @author Christoph Strobl + */ +@ExtendWith(LettuceConnectionFactoryExtension.class) +@EnabledOnCommand("XREAD") +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public class LettuceStreamMessageListenerContainerIntegrationTests + extends AbstractStreamMessageListenerContainerIntegrationTests { + + public LettuceStreamMessageListenerContainerIntegrationTests(RedisConnectionFactory connectionFactory) { + super(connectionFactory); + } + +}