Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAKV-10

Browse Source 
+ RedisList implements BlockingQueue
This commit is contained in:
Costin Leau
2010-12-06 11:04:48 +02:00
parent be2a482880
commit 85e138e79c
6 changed files with 119 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/BoundListOperations.java
View File

@@ -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<K, V> extends KeyBound<K> {
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);
}

12
spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/DefaultBoundListOperations.java
View File

@@ -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<K, V> extends DefaultKeyBound<K> 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<K, V> extends DefaultKeyBound<K> 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);

7
spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/ListOperations.java
View File

@@ -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<K, V> {
V leftPop(K key);
V leftPop(K key, long timeout, TimeUnit unit);
V rightPop(K key);
List<V> blockingLeftPop(int timeout, K... keys);
List<V> blockingRightPop(int timeout, K... keys);
V rightPop(K key, long timeout, TimeUnit unit);
RedisOperations<K, V> getOperations();
}

49
spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisTemplate.java
View File

@@ -707,29 +707,6 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
private class DefaultListOperations implements ListOperations<K, V> {
@Override
public List<V> blockingLeftPop(final int timeout, K... keys) {
final byte[][] rawKeys = rawKeys(keys);
return execute(new RedisCallback<List<V>>() {
@Override
public List<V> doInRedis(RedisConnection connection) {
return values(connection.bLPop(timeout, rawKeys), List.class);
}
}, true);
}
@Override
public List<V> blockingRightPop(final int timeout, K... keys) {
final byte[][] rawKeys = rawKeys(keys);
return execute(new RedisCallback<List<V>>() {
@Override
public List<V> 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<K, V> 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<K, V> 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);

75
spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/util/DefaultRedisList.java
View File

@@ -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<E> extends AbstractRedisCollection<E> implements RedisList<E> {
private final BoundListOperations<String, E> listOps;
private volatile long maxSize = 0;
private volatile int maxSize = 0;
private volatile boolean capped = false;
private volatile long defaultWait = 0;
private class DefaultRedisListIterator<E> extends RedisIterator<E> {
public DefaultRedisListIterator(Iterator<E> delegate) {
@@ -75,7 +82,7 @@ public class DefaultRedisList<E> extends AbstractRedisCollection<E> implements R
* @param boundOps
* @param maxSize
*/
public DefaultRedisList(BoundListOperations<String, E> boundOps, long maxSize) {
public DefaultRedisList(BoundListOperations<String, E> boundOps, int maxSize) {
super(boundOps.getKey(), boundOps.getOperations());
listOps = boundOps;
setMaxSize(maxSize);
@@ -86,7 +93,7 @@ public class DefaultRedisList<E> extends AbstractRedisCollection<E> 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<E> extends AbstractRedisCollection<E> implements R
throw new UnsupportedOperationException();
}
//
// Queue methods
//
@Override
public E element() {
@@ -254,7 +264,7 @@ public class DefaultRedisList<E> extends AbstractRedisCollection<E> implements R
@Override
public boolean offer(E e) {
listOps.leftPush(e);
listOps.rightPush(e);
cap();
return true;
}
@@ -282,4 +292,57 @@ public class DefaultRedisList<E> extends AbstractRedisCollection<E> implements R
return value;
}
//
// BlockingQueue
//
@Override
public int drainTo(Collection<? super E> 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<? super E> 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);
}
}

3
spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/util/RedisList.java
View File

@@ -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<E> extends RedisStore<String>, List<E>, Queue<E> {
public interface RedisList<E> extends RedisStore<String>, List<E>, BlockingQueue<E> {
List<E> range(long start, long end);