diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/BoundListOperations.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/BoundListOperations.java index d28f3d1e3..ae07bfd48 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/BoundListOperations.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/BoundListOperations.java @@ -16,6 +16,7 @@ package org.springframework.data.keyvalue.redis.core; import java.util.List; +import java.util.concurrent.TimeUnit; /** * List operations bound to a certain key. @@ -38,11 +39,16 @@ public interface BoundListOperations extends KeyBound { V leftPop(); + V leftPop(long timeout, TimeUnit unit); + V rightPop(); + V rightPop(long timeout, TimeUnit unit); + Long remove(long i, Object value); V index(long index); void set(long index, V value); + } diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/DefaultBoundListOperations.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/DefaultBoundListOperations.java index 2d8a4e4bd..c8f9fe552 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/DefaultBoundListOperations.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/DefaultBoundListOperations.java @@ -16,6 +16,7 @@ package org.springframework.data.keyvalue.redis.core; import java.util.List; +import java.util.concurrent.TimeUnit; /** @@ -54,6 +55,11 @@ class DefaultBoundListOperations extends DefaultKeyBound implements Bou return ops.leftPop(getKey()); } + @Override + public V leftPop(long timeout, TimeUnit unit) { + return ops.leftPop(getKey(), timeout, unit); + } + @Override public Long leftPush(V value) { return ops.leftPush(getKey(), value); @@ -79,6 +85,12 @@ class DefaultBoundListOperations extends DefaultKeyBound implements Bou return ops.rightPop(getKey()); } + @Override + public V rightPop(long timeout, TimeUnit unit) { + return ops.rightPop(getKey(), timeout, unit); + } + + @Override public Long rightPush(V value) { return ops.rightPush(getKey(), value); diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/ListOperations.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/ListOperations.java index 31c644940..a9a610d3a 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/ListOperations.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/ListOperations.java @@ -16,6 +16,7 @@ package org.springframework.data.keyvalue.redis.core; import java.util.List; +import java.util.concurrent.TimeUnit; /** * Redis, list specific operations. @@ -42,11 +43,11 @@ public interface ListOperations { V leftPop(K key); + V leftPop(K key, long timeout, TimeUnit unit); + V rightPop(K key); - List blockingLeftPop(int timeout, K... keys); - - List blockingRightPop(int timeout, K... keys); + V rightPop(K key, long timeout, TimeUnit unit); RedisOperations getOperations(); } diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisTemplate.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisTemplate.java index cbe4ed605..7caef2f65 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisTemplate.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisTemplate.java @@ -707,29 +707,6 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation private class DefaultListOperations implements ListOperations { - @Override - public List blockingLeftPop(final int timeout, K... keys) { - final byte[][] rawKeys = rawKeys(keys); - - return execute(new RedisCallback>() { - @Override - public List doInRedis(RedisConnection connection) { - return values(connection.bLPop(timeout, rawKeys), List.class); - } - }, true); - } - - @Override - public List blockingRightPop(final int timeout, K... keys) { - final byte[][] rawKeys = rawKeys(keys); - return execute(new RedisCallback>() { - @Override - public List doInRedis(RedisConnection connection) { - return values(connection.bRPop(timeout, rawKeys), List.class); - } - }, true); - } - @Override public V index(K key, final long index) { return execute(new ValueDeserializingRedisCallback(key) { @@ -750,6 +727,20 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation }, true); } + @Override + public V leftPop(K key, long timeout, TimeUnit unit) { + final int tm = (int) unit.toSeconds(timeout); + + return execute(new ValueDeserializingRedisCallback(key) { + @Override + protected byte[] inRedis(byte[] rawKey, RedisConnection connection) { + return connection.bLPop(tm, rawKey).get(0); + } + }, true); + } + + + @Override public Long leftPush(K key, V value) { final byte[] rawKey = rawKey(key); @@ -806,6 +797,18 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation }, true); } + @Override + public V rightPop(K key, long timeout, TimeUnit unit) { + final int tm = (int) unit.toSeconds(timeout); + + return execute(new ValueDeserializingRedisCallback(key) { + @Override + protected byte[] inRedis(byte[] rawKey, RedisConnection connection) { + return connection.bRPop(tm, rawKey).get(0); + } + }, true); + } + @Override public Long rightPush(K key, V value) { final byte[] rawKey = rawKey(key); diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/util/DefaultRedisList.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/util/DefaultRedisList.java index cd5a7a1af..830899172 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/util/DefaultRedisList.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/util/DefaultRedisList.java @@ -20,24 +20,31 @@ import java.util.Iterator; import java.util.List; import java.util.ListIterator; import java.util.NoSuchElementException; +import java.util.concurrent.TimeUnit; import org.springframework.data.keyvalue.redis.core.BoundListOperations; import org.springframework.data.keyvalue.redis.core.RedisOperations; /** - * Default implementation for {@link RedisList}. Allows the maximum size (or the cap) to - * be specified to prevent the list from overgrowing. + * Default implementation for {@link RedisList}. * + * Allows the maximum size (or the cap) to be specified to prevent the list from over growing. + * + * Note that all write operations will execute immediately, whether a cap is specified or not - the list + * will always accept new items (trimming the tail after each insert in case of capped collections). + * * @author Costin Leau */ public class DefaultRedisList extends AbstractRedisCollection implements RedisList { private final BoundListOperations listOps; - private volatile long maxSize = 0; + private volatile int maxSize = 0; private volatile boolean capped = false; + private volatile long defaultWait = 0; + private class DefaultRedisListIterator extends RedisIterator { public DefaultRedisListIterator(Iterator delegate) { @@ -75,7 +82,7 @@ public class DefaultRedisList extends AbstractRedisCollection implements R * @param boundOps * @param maxSize */ - public DefaultRedisList(BoundListOperations boundOps, long maxSize) { + public DefaultRedisList(BoundListOperations boundOps, int maxSize) { super(boundOps.getKey(), boundOps.getOperations()); listOps = boundOps; setMaxSize(maxSize); @@ -86,7 +93,7 @@ public class DefaultRedisList extends AbstractRedisCollection implements R * * @param maxSize list maximum size */ - public void setMaxSize(long maxSize) { + public void setMaxSize(int maxSize) { this.maxSize = maxSize; capped = (maxSize > 0); } @@ -241,6 +248,9 @@ public class DefaultRedisList extends AbstractRedisCollection implements R throw new UnsupportedOperationException(); } + // + // Queue methods + // @Override public E element() { @@ -254,7 +264,7 @@ public class DefaultRedisList extends AbstractRedisCollection implements R @Override public boolean offer(E e) { - listOps.leftPush(e); + listOps.rightPush(e); cap(); return true; } @@ -282,4 +292,57 @@ public class DefaultRedisList extends AbstractRedisCollection implements R return value; } + + // + // BlockingQueue + // + + @Override + public int drainTo(Collection c, int maxElements) { + if (this.equals(c)) { + throw new IllegalArgumentException("Cannot drain a queue to itself"); + } + + int size = size(); + int loop = (size >= maxElements ? maxElements : size); + + for (int index = 0; index < loop; index++) { + c.add(poll()); + } + + return loop; + } + + @Override + public int drainTo(Collection c) { + return drainTo(c, size()); + } + + @Override + public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException { + return offer(e); + } + + @Override + public E poll(long timeout, TimeUnit unit) throws InterruptedException { + E element = listOps.leftPop(timeout, unit); + return (element == null ? null : element); + } + + @Override + public void put(E e) throws InterruptedException { + offer(e); + } + + @Override + public int remainingCapacity() { + return Integer.MAX_VALUE; + } + + @Override + public E take() throws InterruptedException { + return poll(0, TimeUnit.SECONDS); + } + + } \ No newline at end of file diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/util/RedisList.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/util/RedisList.java index cd0c54222..abc99369a 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/util/RedisList.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/util/RedisList.java @@ -17,6 +17,7 @@ package org.springframework.data.keyvalue.redis.util; import java.util.List; import java.util.Queue; +import java.util.concurrent.BlockingQueue; /** * Redis extension for the {@link List} contract. Supports {@link List} and {@link Queue} specific @@ -24,7 +25,7 @@ import java.util.Queue; * * @author Costin Leau */ -public interface RedisList extends RedisStore, List, Queue { +public interface RedisList extends RedisStore, List, BlockingQueue { List range(long start, long end);