diff --git a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java index 1bafa35f7..7c0219857 100644 --- a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java @@ -51,6 +51,8 @@ import org.springframework.util.Assert; */ public class DefaultStringRedisConnection implements StringRedisConnection { + private static final byte[][] EMPTY_2D_BYTE_ARRAY = new byte[0][]; + private final Log log = LogFactory.getLog(DefaultStringRedisConnection.class); private final RedisConnection delegate; private final RedisSerializer serializer; @@ -1247,6 +1249,11 @@ public class DefaultStringRedisConnection implements StringRedisConnection { } private byte[][] serializeMulti(String... keys) { + + if (keys == null) { + return EMPTY_2D_BYTE_ARRAY; + } + byte[][] ret = new byte[keys.length][]; for (int i = 0; i < ret.length; i++) { @@ -2413,12 +2420,12 @@ public class DefaultStringRedisConnection implements StringRedisConnection { @Override public Set zRangeByScore(byte[] key, String min, String max) { - + Set results = delegate.zRangeByScore(key, min, max); if (isFutureConversion()) { addResultConverter(identityConverter); } - + return results; } @@ -2429,8 +2436,62 @@ public class DefaultStringRedisConnection implements StringRedisConnection { if (isFutureConversion()) { addResultConverter(identityConverter); } - + return results; } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfAdd(byte[], byte[][]) + */ + @Override + public Long pfAdd(byte[] key, byte[]... values) { + return delegate.pfAdd(key, values); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.StringRedisConnection#pfAdd(java.lang.String, java.lang.String[]) + */ + @Override + public Long pfAdd(String key, String... values) { + return this.pfAdd(serialize(key), serializeMulti(values)); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfCount(byte[][]) + */ + @Override + public Long pfCount(byte[]... keys) { + return delegate.pfCount(keys); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.StringRedisConnection#pfCount(java.lang.String[]) + */ + @Override + public Long pfCount(String... keys) { + return this.pfCount(serializeMulti(keys)); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfMerge(byte[], byte[][]) + */ + @Override + public void pfMerge(byte[] destinationKey, byte[]... sourceKeys) { + delegate.pfMerge(destinationKey, sourceKeys); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.StringRedisConnection#pfMerge(java.lang.String, java.lang.String[][]) + */ + @Override + public void pfMerge(String destinationKey, String... sourceKeys) { + this.pfMerge(serialize(destinationKey), serializeMulti(sourceKeys)); + } + } diff --git a/src/main/java/org/springframework/data/redis/connection/HyperLogLogCommands.java b/src/main/java/org/springframework/data/redis/connection/HyperLogLogCommands.java new file mode 100644 index 000000000..63cb2de6e --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/HyperLogLogCommands.java @@ -0,0 +1,51 @@ +/* + * Copyright 2014 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 + * + * http://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; + +/** + * {@literal HyperLogLog} specific commands supported by Redis. + * + * @author Christoph Strobl + * @since 1.5 + */ +public interface HyperLogLogCommands { + + /** + * Adds given {@literal values} to the HyperLogLog stored at given {@literal key}. + * + * @param key + * @param values + * @return + */ + Long pfAdd(byte[] key, byte[]... values); + + /** + * Return the approximated cardinality of the structures observed by the HyperLogLog at {@literal key(s)}. + * + * @param keys + * @return + */ + Long pfCount(byte[]... keys); + + /** + * Merge N different HyperLogLogs at {@literal sourceKeys} into a single {@literal destinationKey}. + * + * @param destinationKey + * @param sourceKeys + */ + void pfMerge(byte[] destinationKey, byte[]... sourceKeys); + +} diff --git a/src/main/java/org/springframework/data/redis/connection/RedisCommands.java b/src/main/java/org/springframework/data/redis/connection/RedisCommands.java index 24283111a..1da8cda6f 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisCommands.java @@ -1,5 +1,5 @@ /* - * Copyright 2011-2013 the original author or authors. + * Copyright 2011-2014 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. @@ -20,10 +20,11 @@ package org.springframework.data.redis.connection; * Interface for the commands supported by Redis. * * @author Costin Leau + * @author Christoph Strobl */ public interface RedisCommands extends RedisKeyCommands, RedisStringCommands, RedisListCommands, RedisSetCommands, RedisZSetCommands, RedisHashCommands, RedisTxCommands, RedisPubSubCommands, RedisConnectionCommands, - RedisServerCommands, RedisScriptingCommands { + RedisServerCommands, RedisScriptingCommands, HyperLogLogCommands { /** * 'Native' or 'raw' execution of the given command along-side the given arguments. The command is executed as is, diff --git a/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java index 9b0c70d5e..a69410be8 100644 --- a/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java @@ -35,7 +35,6 @@ import org.springframework.data.redis.serializer.RedisSerializer; * @author Christoph Strobl * @author Thomas Darimont * @author David Liu - * * @see RedisCallback * @see RedisSerializer * @see StringRedisTemplate @@ -368,4 +367,28 @@ public interface StringRedisConnection extends RedisConnection { * @return */ Set zRangeByScore(String key, String min, String max, long offset, long count); + + /** + * Adds given {@literal values} to the HyperLogLog stored at given {@literal key}. + * + * @param key + * @param values + * @return + * @since 1.5 + */ + Long pfAdd(String key, String... values); + + /** + * @param keys + * @return + * @since 1.5 + */ + Long pfCount(String... keys); + + /** + * @param destinationKey + * @param sourceKeys + * @since 1.5 + */ + void pfMerge(String destinationKey, String... sourceKeys); } 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 4de105420..f3fa64d6d 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 @@ -3166,7 +3166,7 @@ public class JedisConnection extends AbstractRedisConnection { @Override public Set zRangeByScore(byte[] key, String min, String max) { - + try { String keyStr = new String(key, "UTF-8"); if (isPipelined()) { @@ -3185,7 +3185,7 @@ public class JedisConnection extends AbstractRedisConnection { @Override public Set zRangeByScore(byte[] key, String min, String max, long offset, long count) { - + try { String keyStr = new String(key, "UTF-8"); if (isPipelined()) { @@ -3196,7 +3196,80 @@ public class JedisConnection extends AbstractRedisConnection { transaction(new JedisResult(transaction.zrangeByScore(keyStr, min, max, (int) offset, (int) count))); return null; } - return JedisConverters.stringSetToByteSet().convert(jedis.zrangeByScore(keyStr, min, max, (int) offset, (int) count)); + return JedisConverters.stringSetToByteSet().convert( + jedis.zrangeByScore(keyStr, min, max, (int) offset, (int) count)); + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfAdd(byte[], byte[][]) + */ + @Override + public Long pfAdd(byte[] key, byte[]... values) { + + Assert.notEmpty(values, "PFADD requires at least one non 'null' value."); + Assert.noNullElements(values, "Values for PFADD must not contain 'null'."); + + try { + if (isPipelined()) { + pipeline(new JedisResult(pipeline.pfadd(key, values))); + return null; + } + if (isQueueing()) { + transaction(new JedisResult(transaction.pfadd(key, values))); + return null; + } + return jedis.pfadd(key, values); + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfCount(byte[][]) + */ + @Override + public Long pfCount(byte[]... keys) { + + Assert.notEmpty(keys, "PFCOUNT requires at least one non 'null' key."); + Assert.noNullElements(keys, "Keys for PFOUNT must not contain 'null'."); + + try { + if (isPipelined()) { + pipeline(new JedisResult(pipeline.pfcount(keys))); + return null; + } + if (isQueueing()) { + transaction(new JedisResult(transaction.pfcount(keys))); + return null; + } + return jedis.pfcount(keys); + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfMerge(byte[], byte[][]) + */ + @Override + public void pfMerge(byte[] destinationKey, byte[]... sourceKeys) { + + try { + if (isPipelined()) { + pipeline(new JedisResult(pipeline.pfmerge(destinationKey, sourceKeys))); + return; + } + if (isQueueing()) { + transaction(new JedisResult(transaction.pfmerge(destinationKey, sourceKeys))); + return; + } + jedis.pfmerge(destinationKey, sourceKeys); } catch (Exception ex) { throw convertJedisAccessException(ex); } diff --git a/src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java b/src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java index eb36ea8ef..cca0a97c5 100644 --- a/src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java @@ -1282,7 +1282,7 @@ public class JredisConnection extends AbstractRedisConnection { public Cursor> hScan(byte[] key, ScanOptions options) { throw new UnsupportedOperationException("'HSCAN' command is not uspported for jredis"); } - + @Override public Set zRangeByScore(byte[] key, String min, String max) { throw new UnsupportedOperationException("'zRangeByScore' command is not uspported for jredis"); @@ -1292,4 +1292,31 @@ public class JredisConnection extends AbstractRedisConnection { public Set zRangeByScore(byte[] key, String min, String max, long offset, long count) { throw new UnsupportedOperationException("'zRangeByScore' command is not uspported for jredis"); } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfAdd(byte[], byte[][]) + */ + @Override + public Long pfAdd(byte[] key, byte[]... values) { + throw new UnsupportedOperationException(); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfCount(byte[][]) + */ + @Override + public Long pfCount(byte[]... keys) { + throw new UnsupportedOperationException(); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfMerge(byte[], byte[][]) + */ + @Override + public void pfMerge(byte[] destinationKey, byte[]... sourceKeys) { + throw new UnsupportedOperationException(); + } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java index 30f5b58b1..08fefdad8 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java @@ -3618,7 +3618,7 @@ public class LettuceConnection extends AbstractRedisConnection { @Override public Set zRangeByScore(byte[] key, String min, String max) { - + try { if (isPipelined()) { pipeline(new LettuceResult(getAsyncConnection().zrangebyscore(key, min, max), @@ -3638,7 +3638,7 @@ public class LettuceConnection extends AbstractRedisConnection { @Override public Set zRangeByScore(byte[] key, String min, String max, long offset, long count) { - + try { if (isPipelined()) { pipeline(new LettuceResult(getAsyncConnection().zrangebyscore(key, min, max, offset, count), @@ -3655,5 +3655,32 @@ public class LettuceConnection extends AbstractRedisConnection { throw convertLettuceAccessException(ex); } } - + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfAdd(byte[], byte[][]) + */ + @Override + public Long pfAdd(byte[] key, byte[]... values) { + throw new UnsupportedOperationException(); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfCount(byte[][]) + */ + @Override + public Long pfCount(byte[]... keys) { + throw new UnsupportedOperationException(); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfMerge(byte[], byte[][]) + */ + @Override + public void pfMerge(byte[] destinationKey, byte[]... sourceKeys) { + throw new UnsupportedOperationException(); + } + } diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java index 2addbea8e..bdd4470e7 100644 --- a/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java @@ -2474,7 +2474,7 @@ public class SrpConnection extends AbstractRedisConnection { @Override public Set zRangeByScore(byte[] key, String min, String max) { - + try { String keyStr = new String(key, "UTF-8"); if (isPipelined()) { @@ -2503,4 +2503,31 @@ public class SrpConnection extends AbstractRedisConnection { throw convertSrpAccessException(ex); } } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfAdd(byte[], byte[][]) + */ + @Override + public Long pfAdd(byte[] key, byte[]... values) { + throw new UnsupportedOperationException(); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfCount(byte[][]) + */ + @Override + public Long pfCount(byte[]... keys) { + throw new UnsupportedOperationException(); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfMerge(byte[], byte[][]) + */ + @Override + public void pfMerge(byte[] destinationKey, byte[]... sourceKeys) { + throw new UnsupportedOperationException(); + } } diff --git a/src/main/java/org/springframework/data/redis/core/DefaultHyperLogLogOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultHyperLogLogOperations.java new file mode 100644 index 000000000..37b0fe8a0 --- /dev/null +++ b/src/main/java/org/springframework/data/redis/core/DefaultHyperLogLogOperations.java @@ -0,0 +1,102 @@ +/* + * Copyright 2014 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 + * + * http://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 java.util.Arrays; + +import org.springframework.dao.DataAccessException; +import org.springframework.data.redis.connection.RedisConnection; + +/** + * @author Christoph Strobl + * @since 1.5 + * @param + * @param + */ +public class DefaultHyperLogLogOperations extends AbstractOperations implements HyperLogLogOperations { + + public DefaultHyperLogLogOperations(RedisTemplate template) { + super(template); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.HyperLogLogOperations#add(java.lang.Object, java.lang.Object[]) + */ + @Override + public Long add(K key, V... values) { + + final byte[] rawKey = rawKey(key); + final byte[][] rawValues = rawValues(values); + + return execute(new RedisCallback() { + + @Override + public Long doInRedis(RedisConnection connection) throws DataAccessException { + return connection.pfAdd(rawKey, rawValues); + } + }, true); + + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.HyperLogLogOperations#size(java.lang.Object[]) + */ + @Override + public Long size(K... keys) { + + final byte[][] rawKeys = rawKeys(Arrays.asList(keys)); + + return execute(new RedisCallback() { + + @Override + public Long doInRedis(RedisConnection connection) throws DataAccessException { + return connection.pfCount(rawKeys); + } + }, true); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.HyperLogLogOperations#union(java.lang.Object, java.lang.Object[]) + */ + @Override + public Long union(K destination, K... sourceKeys) { + + final byte[] rawDestinationKey = rawKey(destination); + final byte[][] rawSourceKeys = rawKeys(Arrays.asList(sourceKeys)); + + return execute(new RedisCallback() { + + @Override + public Long doInRedis(RedisConnection connection) throws DataAccessException { + + connection.pfMerge(rawDestinationKey, rawSourceKeys); + return connection.pfCount(rawDestinationKey); + } + }, true); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.HyperLogLogOperations#delete(java.lang.Object) + */ + @Override + public void delete(K key) { + template.delete(key); + } +} diff --git a/src/main/java/org/springframework/data/redis/core/HyperLogLogOperations.java b/src/main/java/org/springframework/data/redis/core/HyperLogLogOperations.java new file mode 100644 index 000000000..54ef650e7 --- /dev/null +++ b/src/main/java/org/springframework/data/redis/core/HyperLogLogOperations.java @@ -0,0 +1,56 @@ +/* + * Copyright 2014 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 + * + * http://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; + +/** + * @author Christoph Strobl + * @since 1.5 + */ +public interface HyperLogLogOperations { + + /** + * Adds the given {@literal values} to the {@literal key}. + * + * @param key must not be {@literal null}. + * @param values must not be {@literal null}. + * @return 1 of at least one of the values was added to the key; 0 otherwise. + */ + Long add(K key, V... values); + + /** + * Gets the current number of elements within the {@literal key}. + * + * @param keys must not be {@literal null) or {@literal empty}. + * @return + */ + Long size(K... keys); + + /** + * Merges all values of given {@literal sourceKeys} into {@literal destination} key. + * + * @param destination key of HyperLogLog to move source keys into. + * @param sourceKeys must not be {@literal null) or {@literal empty}. + */ + Long union(K destination, K... sourceKeys); + + /** + * Removes the given {@literal key}. + * + * @param key must not be {@literal null}. + */ + void delete(K key); + +} diff --git a/src/main/java/org/springframework/data/redis/core/RedisOperations.java b/src/main/java/org/springframework/data/redis/core/RedisOperations.java index ab49a6564..8cd506d87 100644 --- a/src/main/java/org/springframework/data/redis/core/RedisOperations.java +++ b/src/main/java/org/springframework/data/redis/core/RedisOperations.java @@ -250,6 +250,12 @@ public interface RedisOperations { */ ZSetOperations opsForZSet(); + /** + * @return + * @since 1.5 + */ + HyperLogLogOperations opsForHyperLogLog(); + /** * Returns the operations performed on zset values (also known as sorted sets) bound to the given key. * diff --git a/src/main/java/org/springframework/data/redis/core/RedisTemplate.java b/src/main/java/org/springframework/data/redis/core/RedisTemplate.java index fe7673a51..4086ff5d7 100644 --- a/src/main/java/org/springframework/data/redis/core/RedisTemplate.java +++ b/src/main/java/org/springframework/data/redis/core/RedisTemplate.java @@ -96,6 +96,7 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation private ListOperations listOps; private SetOperations setOps; private ZSetOperations zSetOps; + private HyperLogLogOperations hllOps; /** * Constructs a new RedisTemplate instance. @@ -996,6 +997,19 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation return zSetOps; } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.RedisOperations#opsForHyperLogLog() + */ + @Override + public HyperLogLogOperations opsForHyperLogLog() { + + if (hllOps == null) { + hllOps = new DefaultHyperLogLogOperations(this); + } + return hllOps; + } + public BoundHashOperations boundHashOps(K key) { return new DefaultBoundHashOperations(key, this); } @@ -1075,4 +1089,5 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation public void setEnableTransactionSupport(boolean enableTransactionSupport) { this.enableTransactionSupport = enableTransactionSupport; } + } 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 5fc2000fd..7938eed8b 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -2067,6 +2067,93 @@ public abstract class AbstractConnectionIntegrationTests { assertThat(i, is(3)); } + /** + * @see DATAREDIS-308 + */ + @Test + public void pfAddShouldAddToNonExistingKeyCorrectly() { + + if (!ConnectionUtils.isJedis(connectionFactory)) { + throw new AssumptionViolatedException("PFADD is only available for jedis"); + } + + actual.add(connection.pfAdd("hll", "a", "b", "c")); + + List results = getResults(); + assertThat((Long) results.get(0), is(1L)); + } + + /** + * @see DATAREDIS-308 + */ + @Test + public void pfAddShouldReturnZeroWhenValueAlreadyExists() { + + if (!ConnectionUtils.isJedis(connectionFactory)) { + throw new AssumptionViolatedException("PFADD is only available for jedis"); + } + + actual.add(connection.pfAdd("hll", "a", "b", "c")); + actual.add(connection.pfAdd("hll2", "c", "d", "e")); + actual.add(connection.pfAdd("hll2", "e")); + + List results = getResults(); + assertThat((Long) results.get(0), is(1L)); + assertThat((Long) results.get(1), is(1L)); + assertThat((Long) results.get(2), is(0L)); + } + + /** + * @see DATAREDIS-308 + */ + @Test + public void pfCountShouldReturnCorrectly() { + + if (!ConnectionUtils.isJedis(connectionFactory)) { + throw new AssumptionViolatedException("PFADD is only available for jedis"); + } + + actual.add(connection.pfAdd("hll", "a", "b", "c")); + actual.add(connection.pfCount("hll")); + + List results = getResults(); + assertThat((Long) results.get(0), is(1L)); + assertThat((Long) results.get(1), is(3L)); + } + + /** + * @see DATAREDIS-308 + */ + @Test + public void pfCountWithMultipleKeysShouldReturnCorrectly() { + + if (!ConnectionUtils.isJedis(connectionFactory)) { + throw new AssumptionViolatedException("PFADD is only available for jedis"); + } + + actual.add(connection.pfAdd("hll", "a", "b", "c")); + actual.add(connection.pfAdd("hll2", "d", "e", "f")); + actual.add(connection.pfCount("hll", "hll2")); + + List results = getResults(); + assertThat((Long) results.get(0), is(1L)); + assertThat((Long) results.get(1), is(1L)); + assertThat((Long) results.get(2), is(6L)); + } + + /** + * @see DATAREDIS-308 + */ + @Test(expected = IllegalArgumentException.class) + public void pfCountWithNullKeysShouldThrowIllegalArgumentException() { + + if (!ConnectionUtils.isJedis(connectionFactory)) { + throw new AssumptionViolatedException("PFADD is only available for jedis"); + } + + actual.add(connection.pfCount((String[]) null)); + } + protected void verifyResults(List expected) { assertEquals(expected, getResults()); } diff --git a/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java b/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java index 162b5f79a..81230bf10 100644 --- a/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java +++ b/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java @@ -1720,6 +1720,38 @@ public class DefaultStringRedisConnectionTests { verify(nativeConnection, times(1)).setClientName(eq("foo".getBytes())); } + /** + * @see DATAREDIS-308 + */ + @Test + public void pfAddShouldDelegateToNativeConnectionCorrectly() { + + connection.pfAdd("hll", "spring", "data", "redis"); + verify(nativeConnection, times(1)).pfAdd("hll".getBytes(), "spring".getBytes(), "data".getBytes(), + "redis".getBytes()); + } + + /** + * @see DATAREDIS-308 + */ + @Test + public void pfCountShouldDelegateToNativeConnectionCorrectly() { + + connection.pfCount("hll", "hyperLogLog"); + verify(nativeConnection, times(1)).pfCount("hll".getBytes(), "hyperLogLog".getBytes()); + } + + /** + * @see DATAREDIS-308 + */ + @Test + public void pfMergeShouldDelegateToNativeConnectionCorrectly() { + + connection.pfMerge("merged", "spring", "data", "redis"); + verify(nativeConnection, times(1)).pfMerge("merged".getBytes(), "spring".getBytes(), "data".getBytes(), + "redis".getBytes()); + } + /** * @see DATAREDIS-270 */ diff --git a/src/test/java/org/springframework/data/redis/connection/RedisConnectionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/RedisConnectionUnitTests.java index 935023ae4..bb89db8ea 100644 --- a/src/test/java/org/springframework/data/redis/connection/RedisConnectionUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/RedisConnectionUnitTests.java @@ -791,7 +791,7 @@ public class RedisConnectionUnitTests { public T evalSha(byte[] scriptSha, ReturnType returnType, int numKeys, byte[]... keysAndArgs) { return delegate.evalSha(scriptSha, returnType, numKeys, keysAndArgs); } - + @Override public Set zRangeByScore(byte[] key, String min, String max) { return delegate.zRangeByScore(key, min, max); @@ -801,5 +801,20 @@ public class RedisConnectionUnitTests { public Set zRangeByScore(byte[] key, String min, String max, long offset, long count) { return delegate.zRangeByScore(key, min, max, offset, count); } + + @Override + public Long pfAdd(byte[] key, byte[]... values) { + return delegate.pfAdd(key, values); + } + + @Override + public Long pfCount(byte[]... keys) { + return delegate.pfCount(keys); + } + + @Override + public void pfMerge(byte[] destinationKey, byte[]... sourceKeys) { + delegate.pfMerge(destinationKey, sourceKeys); + } } } diff --git a/src/test/java/org/springframework/data/redis/core/DefaultHyperLogLogOperationsTests.java b/src/test/java/org/springframework/data/redis/core/DefaultHyperLogLogOperationsTests.java new file mode 100644 index 000000000..eebca3412 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/core/DefaultHyperLogLogOperationsTests.java @@ -0,0 +1,177 @@ +/* + * Copyright 2014 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 + * + * http://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 static org.hamcrest.CoreMatchers.*; +import static org.junit.Assert.*; + +import java.util.Collection; + +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.junit.runners.Parameterized.Parameters; +import org.springframework.data.redis.ObjectFactory; +import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.test.util.MinimumRedisVersionRule; +import org.springframework.test.annotation.IfProfileValue; + +/** + * @author Christoph Strobl + */ +@RunWith(Parameterized.class) +public class DefaultHyperLogLogOperationsTests { + + private RedisTemplate redisTemplate; + + private ObjectFactory keyFactory; + + private ObjectFactory valueFactory; + + private HyperLogLogOperations hyperLogLogOps; + + public @Rule MinimumRedisVersionRule versionRule = new MinimumRedisVersionRule(); + + public DefaultHyperLogLogOperationsTests(RedisTemplate redisTemplate, ObjectFactory keyFactory, + ObjectFactory valueFactory) { + this.redisTemplate = redisTemplate; + this.keyFactory = keyFactory; + this.valueFactory = valueFactory; + } + + @Parameters + public static Collection testParams() { + return AbstractOperationsTestParams.testParams(); + } + + @Before + public void setUp() { + hyperLogLogOps = redisTemplate.opsForHyperLogLog(); + } + + @After + public void tearDown() { + redisTemplate.execute(new RedisCallback() { + public Object doInRedis(RedisConnection connection) { + connection.flushDb(); + return null; + } + }); + } + + /** + * @see DATAREDIS + */ + @Test + @SuppressWarnings("unchecked") + @IfProfileValue(name = "redisVersion", value = "2.8+") + public void addShouldAddDistinctValuesCorrectly() { + + K key = keyFactory.instance(); + V v1 = valueFactory.instance(); + V v2 = valueFactory.instance(); + V v3 = valueFactory.instance(); + + assertThat(hyperLogLogOps.add(key, v1, v2, v3), equalTo(1L)); + } + + /** + * @see DATAREDIS-308 + */ + @Test + @SuppressWarnings("unchecked") + @IfProfileValue(name = "redisVersion", value = "2.8+") + public void addShouldNotAddExistingValuesCorrectly() { + + K key = keyFactory.instance(); + V v1 = valueFactory.instance(); + V v2 = valueFactory.instance(); + V v3 = valueFactory.instance(); + + hyperLogLogOps.add(key, v1, v2, v3); + assertThat(hyperLogLogOps.add(key, v2), equalTo(0L)); + } + + /** + * @see DATAREDIS-308 + */ + @Test + @SuppressWarnings("unchecked") + @IfProfileValue(name = "redisVersion", value = "2.8+") + public void sizeShouldCountValuesCorrectly() { + + K key = keyFactory.instance(); + V v1 = valueFactory.instance(); + V v2 = valueFactory.instance(); + V v3 = valueFactory.instance(); + + hyperLogLogOps.add(key, v1, v2, v3); + assertThat(hyperLogLogOps.size(key), equalTo(3L)); + } + + /** + * @see DATAREDIS-308 + */ + @Test + @SuppressWarnings("unchecked") + @IfProfileValue(name = "redisVersion", value = "2.8+") + public void sizeShouldCountValuesOfMultipleKeysCorrectly() { + + K key = keyFactory.instance(); + V v1 = valueFactory.instance(); + V v2 = valueFactory.instance(); + V v3 = valueFactory.instance(); + + K key2 = keyFactory.instance(); + V v4 = valueFactory.instance(); + + hyperLogLogOps.add(key, v1, v2, v3); + hyperLogLogOps.add(key2, v4); + assertThat(hyperLogLogOps.size(key, key2), equalTo(4L)); + } + + /** + * @throws InterruptedException + * @see DATAREDIS-308 + */ + @Test + @SuppressWarnings("unchecked") + @IfProfileValue(name = "redisVersion", value = "2.8+") + public void unionShouldMergeValuesOfMultipleKeysCorrectly() throws InterruptedException { + + K sourceKey_1 = keyFactory.instance(); + V v1 = valueFactory.instance(); + V v2 = valueFactory.instance(); + V v3 = valueFactory.instance(); + + K sourceKey_2 = keyFactory.instance(); + V v4 = valueFactory.instance(); + + K desinationKey = keyFactory.instance(); + + hyperLogLogOps.add(sourceKey_1, v1, v2, v3); + hyperLogLogOps.add(sourceKey_2, v4); + + Thread.sleep(10); // give redis a little time to catch up + hyperLogLogOps.union(desinationKey, sourceKey_1, sourceKey_2); + Thread.sleep(10); // give redis a little time to catch up + + assertThat(hyperLogLogOps.size(desinationKey), equalTo(4L)); + } +}