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

+ fixed problem caused by message refactoring

Browse Source
This commit is contained in:
Costin Leau
2011-01-21 19:34:08 +02:00
parent 6e59acd62a
commit 4ce7d1fb45
3 changed files with 31 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnection.java
View File

@@ -763,8 +763,9 @@ public class JedisConnection implements RedisConnection {
public Boolean getBit(byte[] key, long offset) {
try {
if (isQueueing()) {
transaction.getbit(key, (int) offset);
return null;
// transaction.getbit(key, (int) offset);
// return null;
throw new UnsupportedOperationException();
}
return (jedis.getbit(key, (int) offset) == 0 ? Boolean.FALSE : Boolean.TRUE);
} catch (Exception ex) {
@@ -776,8 +777,9 @@ public class JedisConnection implements RedisConnection {
public void setBit(byte[] key, long offset, boolean value) {
try {
if (isQueueing()) {
transaction.setbit(key, (int) offset, JedisUtils.asBit(value));
return;
// transaction.setbit(key, (int) offset, JedisUtils.asBit(value));
// return;
throw new UnsupportedOperationException();
}
jedis.setbit(key, (int) offset, JedisUtils.asBit(value));
} catch (Exception ex) {
@@ -873,8 +875,9 @@ public class JedisConnection implements RedisConnection {
public Long lInsert(byte[] key, POSITION where, byte[] pivot, byte[] value) {
try {
if (isQueueing()) {
transaction.linsert(key, JedisUtils.convertPosition(where), pivot, value);
return null;
// transaction.linsert(key, JedisUtils.convertPosition(where), pivot, value);
// return null;
throw new UnsupportedOperationException();
}
return jedis.linsert(key, JedisUtils.convertPosition(where), pivot, value);
} catch (Exception ex) {
@@ -992,7 +995,7 @@ public class JedisConnection implements RedisConnection {
if (isQueueing()) {
throw new UnsupportedOperationException();
}
return jedis.brpoplpush(srcKey, dstKey, timeout);
return jedis.brpoplpush(srcKey, dstKey, timeout).getBytes();
} catch (Exception ex) {
throw convertJedisAccessException(ex);
}

4
spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisMessageListener.java
View File

@@ -37,12 +37,12 @@ class JedisMessageListener extends JedisPubSub {
@Override
public void onMessage(String channel, String message) {
listener.onMessage(new DefaultMessage(message.getBytes(), channel.getBytes()), null);
listener.onMessage(new DefaultMessage(channel.getBytes(), message.getBytes()), null);
}
@Override
public void onPMessage(String pattern, String channel, String message) {
listener.onMessage(new DefaultMessage(message.getBytes(), channel.getBytes()), pattern.getBytes());
listener.onMessage(new DefaultMessage(channel.getBytes(), message.getBytes()), pattern.getBytes());
}
@Override

20
spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/listener/PubSubTests.java
View File

@@ -53,7 +53,7 @@ public class PubSubTests<T> {
protected RedisTemplate template;
private static Set<RedisConnectionFactory> connFactories = new LinkedHashSet<RedisConnectionFactory>();
private final BlockingDeque<String> bag = new LinkedBlockingDeque<String>(4);
private final BlockingDeque<String> bag = new LinkedBlockingDeque<String>(99);
private final Object handler = new Object() {
void handleMessage(String message) {
@@ -130,7 +130,25 @@ public class PubSubTests<T> {
set.add(bag.poll(1, TimeUnit.SECONDS));
set.add(bag.poll(1, TimeUnit.SECONDS));
assertTrue(set.contains(payload1));
assertTrue(set.contains(payload2));
}
@Test
public void testMessageBatch() throws Exception {
container.addMessageListener(adapter, Arrays.asList(new ChannelTopic(CHANNEL)));
// wait for the container to start the registration
int COUNT = 10;
Thread.sleep(500);
for (int i = 0; i < COUNT; i++) {
template.convertAndSend(CHANNEL, "message=" + i);
}
Thread.sleep(1000);
assertEquals(COUNT, bag.size());
}
}