From e934acacd9d7e0ba3727fab593d5cd063de3890f Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Fri, 30 Mar 2012 14:42:34 +0300 Subject: [PATCH 1/9] first draft for SPRP driver --- .classpath | 31 +- .settings/org.eclipse.jdt.core.prefs | 2 +- build.gradle | 1 + gradle.properties | 6 +- .../connection/sredis/SRedisConnection.java | 1797 +++++++++++++++++ .../sredis/SRedisConnectionFactory.java | 130 ++ .../connection/sredis/SRedisSubscription.java | 71 + .../redis/connection/sredis/SRedisUtils.java | 214 ++ .../redis/connection/sredis/package-info.java | 5 + 9 files changed, 2241 insertions(+), 16 deletions(-) create mode 100644 src/main/java/org/springframework/data/redis/connection/sredis/SRedisConnection.java create mode 100644 src/main/java/org/springframework/data/redis/connection/sredis/SRedisConnectionFactory.java create mode 100644 src/main/java/org/springframework/data/redis/connection/sredis/SRedisSubscription.java create mode 100644 src/main/java/org/springframework/data/redis/connection/sredis/SRedisUtils.java create mode 100644 src/main/java/org/springframework/data/redis/connection/sredis/package-info.java diff --git a/.classpath b/.classpath index 4d236507f..bbf9837aa 100644 --- a/.classpath +++ b/.classpath @@ -8,31 +8,36 @@ + - + + - - + + - - - + + + + + + + + - - - - - - + + + + + - diff --git a/.settings/org.eclipse.jdt.core.prefs b/.settings/org.eclipse.jdt.core.prefs index be590e07c..b69a99a38 100644 --- a/.settings/org.eclipse.jdt.core.prefs +++ b/.settings/org.eclipse.jdt.core.prefs @@ -1,5 +1,5 @@ # -#Tue Dec 13 20:42:40 EET 2011 +#Fri Mar 30 10:10:50 EEST 2012 org.eclipse.jdt.core.compiler.debug.localVariable=generate org.eclipse.jdt.core.compiler.compliance=1.5 org.eclipse.jdt.core.compiler.codegen.unusedLocal=preserve diff --git a/build.gradle b/build.gradle index f0ab280eb..066c12d59 100644 --- a/build.gradle +++ b/build.gradle @@ -82,6 +82,7 @@ dependencies { // Redis Drivers compile "redis.clients:jedis:$jedisVersion" + compile "com.github.spullara.redis:client:$sredisVersion" compile("org.jredis:jredis-anthonylauzon:$jredisVersion") { optional = true } compile("org.idevlab:rjc:$rjcVersion") { optional = true } diff --git a/gradle.properties b/gradle.properties index f2e892a26..9faebb341 100644 --- a/gradle.properties +++ b/gradle.properties @@ -5,7 +5,7 @@ log4jVersion = 1.2.16 slf4jVersion = 1.6.4 # Common libraries -springVersion = 3.1.0.RELEASE +springVersion = 3.1.1.RELEASE jacksonVersion = 1.8.6 # Testing @@ -15,7 +15,8 @@ mockitoVersion = 1.8.5 # Drivers jedisVersion = 2.0.0 jredisVersion = 03122010 -rjcVersion= 0.6.4 +rjcVersion = 0.6.4 +sredisVersion = 0.1 # Manifest properties @@ -24,6 +25,7 @@ spring.range = "[3.1.0, 4.0.0)" jedis.range = "[2.0.0, 2.0.0]" jackson.range = "[1.6, 2.0.0)" rjc.range = "[0.6.4, 0.6.4]" +sredis.range = "[0.1, 1.0)" # -------------------- # Project wide version diff --git a/src/main/java/org/springframework/data/redis/connection/sredis/SRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/sredis/SRedisConnection.java new file mode 100644 index 000000000..aa07fb1bd --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/sredis/SRedisConnection.java @@ -0,0 +1,1797 @@ +/* + * Copyright 2011 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.sredis; + +import java.io.IOException; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Set; + +import org.springframework.dao.DataAccessException; +import org.springframework.data.redis.RedisConnectionFailureException; +import org.springframework.data.redis.RedisSystemException; +import org.springframework.data.redis.connection.DataType; +import org.springframework.data.redis.connection.MessageListener; +import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.connection.RedisSubscribedConnectionException; +import org.springframework.data.redis.connection.SortParameters; +import org.springframework.data.redis.connection.Subscription; + +import redis.client.RedisClient; +import redis.client.RedisClient.Pipeline; +import redis.client.RedisException; +import redis.client.SocketPool; +import redis.reply.MultiBulkReply; + +import com.google.common.util.concurrent.ListenableFuture; + +/** + * {@code RedisConnection} implementation on top of spullara Redis Protocol library. + * + * @author Costin Leau + */ +public class SRedisConnection implements RedisConnection { + + private boolean isClosed = false; + private boolean isMulti = false; + private final RedisClient client; + private Pipeline pipeline; + private volatile SRedisSubscription subscription; + + public SRedisConnection(SocketPool pool) { + try { + this.client = new RedisClient(pool); + } catch (IOException e) { + throw new RedisConnectionFailureException("Could not connect", e); + } + } + + protected DataAccessException convertSRAccessException(Exception ex) { + if (ex instanceof RedisException) { + return SRedisUtils.convertSRedisAccessException((RedisException) ex); + } + if (ex instanceof IOException) { + return new RedisConnectionFailureException("Redis connection failed", (IOException) ex); + } + + return new RedisSystemException("Unknown SRedis exception", ex); + } + + + public void close() throws DataAccessException { + isClosed = true; + + try { + client.close(); + } catch (IOException ex) { + throw convertSRAccessException(ex); + } + } + + public boolean isClosed() { + return isClosed; + } + + public RedisClient getNativeConnection() { + return client; + } + + + public boolean isQueueing() { + return isMulti; + } + + public boolean isPipelined() { + return (pipeline != null); + } + + + public void openPipeline() { + if (pipeline == null) { + pipeline = client.pipeline(); + } + } + + @SuppressWarnings("unchecked") + public List closePipeline() { + if (pipeline != null) { + ListenableFuture reply = pipeline.exec(); + pipeline = null; + if (reply != null) { + try { + return SRedisUtils.toList(reply.get().byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + } + return Collections.emptyList(); + } + + + public List sort(byte[] key, SortParameters params) { + + byte[] sort = SRedisUtils.sort(params); + + try { + if (isPipelined()) { + pipeline.sort(key, sort, null, null); + return null; + } + return SRedisUtils.toBytesList(client.sort(key, sort, null, null).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + public Long sort(byte[] key, SortParameters params, byte[] sortKey) { + + byte[] sort = SRedisUtils.sort(params, sortKey); + + try { + if (isPipelined()) { + pipeline.sort(key, sort, null, null); + return null; + } + return SRedisUtils.toLong(client.sort(key, sort, null, null).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + public Long dbSize() { + try { + if (isPipelined()) { + pipeline.dbsize(); + return null; + } + return client.dbsize().integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + + public void flushDb() { + try { + if (isPipelined()) { + pipeline.flushdb(); + return; + } + client.flushdb(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void flushAll() { + try { + if (isPipelined()) { + pipeline.flushall(); + return; + } + client.flushall(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void bgSave() { + try { + if (isPipelined()) { + pipeline.bgsave(); + return; + } + client.bgsave(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void bgWriteAof() { + try { + if (isPipelined()) { + pipeline.bgrewriteaof(); + return; + } + client.bgrewriteaof(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void save() { + try { + if (isPipelined()) { + pipeline.save(); + return; + } + client.save(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public List getConfig(String param) { + try { + if (isPipelined()) { + pipeline.config_get(param); + return null; + } + return Collections.singletonList(client.config_get(param).toString()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Properties info() { + try { + if (isPipelined()) { + pipeline.info(); + return null; + } + return SRedisUtils.info(client.info()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long lastSave() { + try { + if (isPipelined()) { + pipeline.lastsave(); + return null; + } + return client.lastsave().integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void setConfig(String param, String value) { + try { + if (isPipelined()) { + pipeline.config_set(param, value); + return; + } + client.config_set(param, value); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + + public void resetConfigStats() { + try { + if (isPipelined()) { + pipeline.config_resetstat(); + return; + } + client.config_resetstat(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void shutdown() { + try { + if (isPipelined()) { + pipeline.shutdown(); + return; + } + client.shutdown(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] echo(byte[] message) { + try { + if (isPipelined()) { + pipeline.echo(message); + return null; + } + return client.echo(message).bytes; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public String ping() { + try { + if (isPipelined()) { + pipeline.ping(); + } + return client.ping().status; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long del(byte[]... keys) { + try { + if (isPipelined()) { + pipeline.del(keys); + return null; + } + return client.del(keys).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void discard() { + isMulti = false; + try { + if (isPipelined()) { + pipeline.discard(); + return; + } + + client.discard(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public List exec() { + isMulti = false; + try { + if (isPipelined()) { + pipeline.exec(); + return null; + } + return SRedisUtils.toList(client.exec().byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean exists(byte[] key) { + try { + if (isPipelined()) { + pipeline.exists(key); + return null; + } + return client.exists(key).integer == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean expire(byte[] key, long seconds) { + try { + if (isPipelined()) { + pipeline.expire(key, seconds); + return null; + } + return client.expire(key, seconds).integer == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean expireAt(byte[] key, long unixTime) { + try { + if (isPipelined()) { + pipeline.expireat(key, unixTime); + return null; + } + return client.expireat(key, unixTime).integer == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set keys(byte[] pattern) { + try { + if (isPipelined()) { + pipeline.keys(pattern); + return null; + } + return SRedisUtils.toSet(client.keys(pattern).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void multi() { + if (isQueueing()) { + return; + } + isMulti = true; + try { + if (isPipelined()) { + pipeline.multi(); + return; + } + client.multi(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean persist(byte[] key) { + try { + if (isPipelined()) { + pipeline.persist(key); + return null; + } + return client.persist(key).integer == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean move(byte[] key, int dbIndex) { + try { + if (isPipelined()) { + pipeline.move(key, dbIndex); + return null; + } + return client.move(key, dbIndex).integer == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] randomKey() { + try { + if (isPipelined()) { + pipeline.randomkey(); + return null; + } + return client.randomkey().bytes; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void rename(byte[] oldName, byte[] newName) { + try { + if (isPipelined()) { + pipeline.rename(oldName, newName); + return; + } + client.rename(oldName, newName); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean renameNX(byte[] oldName, byte[] newName) { + try { + if (isPipelined()) { + pipeline.renamenx(oldName, newName); + return null; + } + return (client.renamenx(oldName, newName).integer == 1); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void select(int dbIndex) { + try { + if (isPipelined()) { + throw new UnsupportedOperationException(); + } + client.select(dbIndex); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long ttl(byte[] key) { + try { + if (isPipelined()) { + pipeline.ttl(key); + return null; + } + return client.ttl(key).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public DataType type(byte[] key) { + try { + if (isPipelined()) { + pipeline.type(key); + return null; + } + return DataType.fromCode(client.type(key).status); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void unwatch() { + try { + client.unwatch(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void watch(byte[]... keys) { + if (isQueueing()) { + throw new UnsupportedOperationException(); + } + try { + for (byte[] key : keys) { + if (isPipelined()) { + pipeline.watch(key); + } + else { + client.watch(key); + } + } + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + // + // String commands + // + + + public byte[] get(byte[] key) { + try { + if (isPipelined()) { + pipeline.get(key); + return null; + } + + return client.get(key).bytes; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void set(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.set(key, value); + return; + } + client.set(key, value); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + + public byte[] getSet(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.getset(key, value); + return null; + } + return client.getset(key, value).bytes; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long append(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.append(key, value); + return null; + } + return client.append(key, value).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public List mGet(byte[]... keys) { + try { + if (isPipelined()) { + pipeline.mget(keys); + return null; + } + return SRedisUtils.toBytesList(client.mget(keys).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void mSet(Map tuples) { + try { + if (isPipelined()) { + pipeline.mset(SRedisUtils.convert(tuples)); + return; + } + client.mset(SRedisUtils.convert(tuples)); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void mSetNX(Map tuples) { + try { + if (isPipelined()) { + pipeline.msetnx(SRedisUtils.convert(tuples)); + return; + } + client.msetnx(SRedisUtils.convert(tuples)); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void setEx(byte[] key, long time, byte[] value) { + try { + if (isPipelined()) { + pipeline.setex(key, time, value); + return; + } + client.setex(key, time, value); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean setNX(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.setnx(key, value); + return null; + } + return client.setnx(key, value).integer == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] getRange(byte[] key, long start, long end) { + try { + if (isPipelined()) { + pipeline.getrange(key, start, end); + return null; + } + return client.getrange(key, start, end).bytes; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long decr(byte[] key) { + try { + if (isPipelined()) { + pipeline.decr(key); + return null; + } + return client.decr(key).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long decrBy(byte[] key, long value) { + try { + if (isPipelined()) { + pipeline.decrby(key, value); + return null; + } + return client.decrby(key, value).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long incr(byte[] key) { + try { + if (isPipelined()) { + pipeline.incr(key); + return null; + } + return client.incr(key).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long incrBy(byte[] key, long value) { + try { + if (isPipelined()) { + pipeline.incrby(key, value); + return null; + } + return client.incrby(key, value).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean getBit(byte[] key, long offset) { + try { + if (isQueueing()) { + throw new UnsupportedOperationException(); + } + if (isPipelined()) { + throw new UnsupportedOperationException(); + } + return (client.getbit(key, offset).integer == 1); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void setBit(byte[] key, long offset, boolean value) { + try { + if (isQueueing()) { + throw new UnsupportedOperationException(); + } + if (isPipelined()) { + throw new UnsupportedOperationException(); + } + client.setbit(key, offset, SRedisUtils.asBit(value)); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void setRange(byte[] key, byte[] value, long start) { + try { + if (isQueueing()) { + throw new UnsupportedOperationException(); + } + if (isPipelined()) { + throw new UnsupportedOperationException(); + } + client.setrange(key, start, value); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long strLen(byte[] key) { + try { + if (isPipelined()) { + pipeline.strlen(key); + return null; + } + return client.strlen(key).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + // + // List commands + // + + + public Long lPush(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.lpush(key, value); + return null; + } + return client.lpush(key, value).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long rPush(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.rpush(key, value); + return null; + } + return client.rpush(key, value).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public List bLPop(int timeout, byte[]... keys) { + try { + if (isPipelined()) { + pipeline.blpop(timeout, keys); + return null; + } + return SRedisUtils.toBytesList(client.blpop(timeout, keys).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public List bRPop(int timeout, byte[]... keys) { + try { + if (isPipelined()) { + pipeline.brpop(timeout, keys); + return null; + } + return SRedisUtils.toBytesList(client.brpop(timeout, keys).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] lIndex(byte[] key, long index) { + try { + if (isPipelined()) { + pipeline.lindex(key, index); + return null; + } + return client.lindex(key, index).bytes; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long lInsert(byte[] key, Position where, byte[] pivot, byte[] value) { + try { + if (isPipelined()) { + pipeline.linsert(key, SRedisUtils.convertPosition(where), pivot, value); + return null; + } + return client.linsert(key, SRedisUtils.convertPosition(where), pivot, value).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long lLen(byte[] key) { + try { + if (isPipelined()) { + pipeline.llen(key); + return null; + } + return client.llen(key).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] lPop(byte[] key) { + try { + if (isPipelined()) { + pipeline.lpop(key); + return null; + } + return client.lpop(key).bytes; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public List lRange(byte[] key, long start, long end) { + try { + if (isPipelined()) { + pipeline.lrange(key, start, end); + return null; + } + return SRedisUtils.toBytesList(client.lrange(key, start, end).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long lRem(byte[] key, long count, byte[] value) { + try { + if (isPipelined()) { + pipeline.lrem(key, count, value); + return null; + } + return client.lrem(key, count, value).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void lSet(byte[] key, long index, byte[] value) { + try { + if (isPipelined()) { + pipeline.lset(key, index, value); + return; + } + client.lset(key, index, value); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void lTrim(byte[] key, long start, long end) { + try { + if (isPipelined()) { + pipeline.ltrim(key, start, end); + return; + } + client.ltrim(key, start, end); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] rPop(byte[] key) { + try { + if (isPipelined()) { + pipeline.rpop(key); + return null; + } + return client.rpop(key).bytes; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] rPopLPush(byte[] srcKey, byte[] dstKey) { + try { + if (isPipelined()) { + pipeline.rpoplpush(srcKey, dstKey); + return null; + } + return client.rpoplpush(srcKey, dstKey).bytes; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] bRPopLPush(int timeout, byte[] srcKey, byte[] dstKey) { + try { + if (isPipelined()) { + pipeline.brpoplpush(srcKey, dstKey, timeout); + return null; + } + return client.brpoplpush(srcKey, dstKey, timeout).bytes; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long lPushX(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.lpushx(key, value); + return null; + } + return client.lpushx(key, value).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long rPushX(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.rpushx(key, value); + return null; + } + return client.rpushx(key, value).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + // + // Set commands + // + + + public Boolean sAdd(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.sadd(key, value); + return null; + } + return (client.sadd(key, value).integer == 1); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long sCard(byte[] key) { + try { + if (isPipelined()) { + pipeline.scard(key); + return null; + } + return client.scard(key).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set sDiff(byte[]... keys) { + try { + if (isPipelined()) { + pipeline.sdiff(keys); + return null; + } + return SRedisUtils.toSet(client.sdiff(keys).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void sDiffStore(byte[] destKey, byte[]... keys) { + try { + if (isPipelined()) { + pipeline.sdiffstore(destKey, keys); + return; + } + client.sdiffstore(destKey, keys); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set sInter(byte[]... keys) { + try { + if (isPipelined()) { + pipeline.sinter(keys); + return null; + } + return SRedisUtils.toSet(client.sinter(keys).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void sInterStore(byte[] destKey, byte[]... keys) { + try { + if (isPipelined()) { + pipeline.sinterstore(destKey, keys); + return; + } + client.sinterstore(destKey, keys); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean sIsMember(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.sismember(key, value); + return null; + } + return client.sismember(key, value).integer == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set sMembers(byte[] key) { + try { + if (isPipelined()) { + pipeline.smembers(key); + return null; + } + return SRedisUtils.toSet(client.smembers(key).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean sMove(byte[] srcKey, byte[] destKey, byte[] value) { + try { + if (isPipelined()) { + pipeline.smove(srcKey, destKey, value); + return null; + } + return client.smove(srcKey, destKey, value).integer == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] sPop(byte[] key) { + try { + if (isPipelined()) { + pipeline.spop(key); + return null; + } + return client.spop(key).bytes; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] sRandMember(byte[] key) { + try { + if (isPipelined()) { + pipeline.srandmember(key); + return null; + } + return client.srandmember(key).bytes; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean sRem(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.srem(key, value); + return null; + } + return client.srem(key, value).integer == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set sUnion(byte[]... keys) { + try { + if (isPipelined()) { + pipeline.sunion(keys); + return null; + } + return SRedisUtils.toSet(client.sunion(keys).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void sUnionStore(byte[] destKey, byte[]... keys) { + try { + if (isPipelined()) { + pipeline.sunionstore(destKey, keys); + return; + } + client.sunionstore(destKey, keys); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + // + // ZSet commands + // + + + public Boolean zAdd(byte[] key, double score, byte[] value) { + try { + if (isPipelined()) { + pipeline.zadd(key, score, value, null, null); + return null; + } + return client.zadd(key, score, value, null, null).integer == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long zCard(byte[] key) { + try { + if (isPipelined()) { + pipeline.zcard(key); + return null; + } + return client.zcard(key).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long zCount(byte[] key, double min, double max) { + try { + if (isQueueing()) { + pipeline.zcount(key, min, max); + return null; + } + return client.zcount(key, min, max).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Double zIncrBy(byte[] key, double increment, byte[] value) { + try { + if (isPipelined()) { + pipeline.zincrby(key, increment, value); + return null; + } + return SRedisUtils.toDouble(client.zincrby(key, increment, value).bytes); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long zInterStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets) { + throw new UnsupportedOperationException(); + } + + + public Long zInterStore(byte[] destKey, byte[]... sets) { + try { + if (isQueueing()) { + pipeline.zinterstore(destKey, sets); + return null; + } + return client.zinterstore(destKey, sets).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRange(byte[] key, long start, long end) { + try { + if (isPipelined()) { + pipeline.zrange(key, start, end, null); + return null; + } + return SRedisUtils.toSet(client.zrange(key, start, end, null).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRangeWithScores(byte[] key, long start, long end) { + try { + if (isPipelined()) { + pipeline.zrange(key, start, end, SRedisUtils.WITHSCORES); + return null; + } + return SRedisUtils.convertTuple(client.zrange(key, start, end, SRedisUtils.WITHSCORES)); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRangeByScore(byte[] key, double min, double max) { + try { + if (isPipelined()) { + pipeline.zrangebyscore(key, min, max, null, null); + return null; + } + return SRedisUtils.toSet(client.zrangebyscore(key, min, max, null, null).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRangeByScoreWithScores(byte[] key, double min, double max) { + try { + if (isPipelined()) { + pipeline.zrangebyscore(key, min, max, SRedisUtils.WITHSCORES, null); + return null; + } + return SRedisUtils.convertTuple(client.zrangebyscore(key, min, max, SRedisUtils.WITHSCORES, null)); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRevRangeWithScores(byte[] key, long start, long end) { + try { + if (isPipelined()) { + pipeline.zrevrange(key, start, end, SRedisUtils.WITHSCORES); + return null; + } + return SRedisUtils.convertTuple(client.zrevrange(key, start, end, SRedisUtils.WITHSCORES)); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRangeByScore(byte[] key, double min, double max, long offset, long count) { + try { + byte[] limit = SRedisUtils.limit(offset, count); + if (isPipelined()) { + pipeline.zrangebyscore(key, min, max, null, limit); + return null; + } + return SRedisUtils.toSet(client.zrangebyscore(key, min, max, null, limit).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRangeByScoreWithScores(byte[] key, double min, double max, long offset, long count) { + try { + byte[] limit = SRedisUtils.limit(offset, count); + if (isPipelined()) { + pipeline.zrangebyscore(key, min, max, SRedisUtils.WITHSCORES, limit); + return null; + } + return SRedisUtils.convertTuple(client.zrangebyscore(key, min, max, SRedisUtils.WITHSCORES, limit)); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRevRangeByScore(byte[] key, double min, double max, long offset, long count) { + try { + byte[] limit = SRedisUtils.limit(offset, count); + if (isPipelined()) { + client.zrevrangebyscore(key, min, max, null, limit); + } + return SRedisUtils.toSet(client.zrevrangebyscore(key, min, max, null, limit).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRevRangeByScore(byte[] key, double min, double max) { + try { + if (isPipelined()) { + client.zrevrangebyscore(key, min, max, null, null); + } + return SRedisUtils.toSet(client.zrevrangebyscore(key, min, max, null, null).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRevRangeByScoreWithScores(byte[] key, double min, double max, long offset, long count) { + try { + byte[] limit = SRedisUtils.limit(offset, count); + if (isPipelined()) { + client.zrevrangebyscore(key, min, max, SRedisUtils.WITHSCORES, limit); + } + return SRedisUtils.convertTuple(client.zrevrangebyscore(key, min, max, SRedisUtils.WITHSCORES, limit)); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRevRangeByScoreWithScores(byte[] key, double min, double max) { + try { + if (isPipelined()) { + client.zrevrangebyscore(key, min, max, SRedisUtils.WITHSCORES, null); + } + return SRedisUtils.convertTuple(client.zrevrangebyscore(key, min, max, SRedisUtils.WITHSCORES, null)); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long zRank(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.zrank(key, value); + return null; + } + return client.zrank(key, value).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean zRem(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.zrem(key, value); + return null; + } + return client.zrem(key, value).integer == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long zRemRange(byte[] key, long start, long end) { + try { + if (isPipelined()) { + pipeline.zremrangebyrank(key, start, end); + return null; + } + return client.zremrangebyrank(key, start, end).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long zRemRangeByScore(byte[] key, double min, double max) { + try { + if (isPipelined()) { + pipeline.zremrangebyscore(key, min, max); + return null; + } + return client.zremrangebyscore(key, min, max).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRevRange(byte[] key, long start, long end) { + try { + if (isPipelined()) { + pipeline.zrevrange(key, start, end, null); + return null; + } + return SRedisUtils.toSet(client.zrevrange(key, start, end, null).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long zRevRank(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.zrevrank(key, value); + return null; + } + return client.zrevrank(key, value).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Double zScore(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.zscore(key, value); + return null; + } + return SRedisUtils.toDouble(client.zscore(key, value).bytes); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long zUnionStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets) { + throw new UnsupportedOperationException(); + } + + + public Long zUnionStore(byte[] destKey, byte[]... sets) { + try { + if (isPipelined()) { + pipeline.zunionstore(destKey, sets); + return null; + } + return client.zunionstore(destKey, sets).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + // + // Hash commands + // + + + public Boolean hSet(byte[] key, byte[] field, byte[] value) { + try { + if (isPipelined()) { + pipeline.hset(key, field, value); + return null; + } + return client.hset(key, field, value).integer == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean hSetNX(byte[] key, byte[] field, byte[] value) { + try { + if (isPipelined()) { + pipeline.hsetnx(key, field, value); + return null; + } + return client.hsetnx(key, field, value).integer == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean hDel(byte[] key, byte[] field) { + try { + if (isPipelined()) { + pipeline.hdel(key, field); + return null; + } + return client.hdel(key, field).integer == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean hExists(byte[] key, byte[] field) { + try { + if (isPipelined()) { + pipeline.hexists(key, field); + return null; + } + return client.hexists(key, field).integer == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] hGet(byte[] key, byte[] field) { + try { + if (isPipelined()) { + pipeline.hget(key, field); + return null; + } + return client.hget(key, field).bytes; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Map hGetAll(byte[] key) { + try { + if (isPipelined()) { + pipeline.hgetall(key); + return null; + } + return SRedisUtils.toMap(client.hgetall(key).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long hIncrBy(byte[] key, byte[] field, long delta) { + try { + if (isPipelined()) { + pipeline.hincrby(key, field, delta); + return null; + } + return client.hincrby(key, field, delta).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set hKeys(byte[] key) { + try { + if (isPipelined()) { + pipeline.hkeys(key); + return null; + } + return SRedisUtils.toSet(client.hkeys(key).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long hLen(byte[] key) { + try { + if (isPipelined()) { + pipeline.hlen(key); + return null; + } + return client.hlen(key).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public List hMGet(byte[] key, byte[]... fields) { + try { + if (isPipelined()) { + pipeline.hmget(key, fields); + return null; + } + return SRedisUtils.toBytesList(client.hmget(key, fields).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void hMSet(byte[] key, Map tuple) { + try { + if (isPipelined()) { + pipeline.hmset(key, tuple); + return; + } + client.hmset(key, tuple); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public List hVals(byte[] key) { + try { + if (isPipelined()) { + pipeline.hvals(key); + return null; + } + return SRedisUtils.toBytesList(client.hvals(key).byteArrays); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + // + // Pub/Sub functionality + // + + public Long publish(byte[] channel, byte[] message) { + try { + if (isQueueing()) { + throw new UnsupportedOperationException(); + } + if (isPipelined()) { + pipeline.publish(channel, message); + return null; + } + return client.publish(channel, message).integer; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Subscription getSubscription() { + return subscription; + } + + + public boolean isSubscribed() { + return (subscription != null && subscription.isAlive()); + } + + + public void pSubscribe(MessageListener listener, byte[]... patterns) { + checkSubscription(); + + try { + if (isQueueing()) { + throw new UnsupportedOperationException(); + } + if (isPipelined()) { + throw new UnsupportedOperationException(); + } + + subscription = new SRedisSubscription(listener, client); + subscription.pSubscribe(patterns); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void subscribe(MessageListener listener, byte[]... channels) { + checkSubscription(); + + try { + if (isPipelined()) { + throw new UnsupportedOperationException(); + } + + subscription = new SRedisSubscription(listener, client); + subscription.subscribe(channels); + + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + private void checkSubscription() { + if (isSubscribed()) { + throw new RedisSubscribedConnectionException( + "Connection already subscribed; use the connection Subscription to cancel or add new channels"); + } + } +} \ No newline at end of file diff --git a/src/main/java/org/springframework/data/redis/connection/sredis/SRedisConnectionFactory.java b/src/main/java/org/springframework/data/redis/connection/sredis/SRedisConnectionFactory.java new file mode 100644 index 000000000..09f6935c9 --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/sredis/SRedisConnectionFactory.java @@ -0,0 +1,130 @@ +/* + * Copyright 2011-2012 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.sredis; + +import java.io.IOException; +import java.lang.reflect.Field; +import java.net.Socket; +import java.util.Queue; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.dao.DataAccessException; +import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; +import org.springframework.util.ReflectionUtils; + +import redis.client.SocketPool; + +/** + * Connection factory creating Redis Protocol based connections. + * + * @author Costin Leau + */ +public class SRedisConnectionFactory implements InitializingBean, DisposableBean, RedisConnectionFactory { + + private final static Log log = LogFactory.getLog(JedisConnectionFactory.class); + + private String hostName = "localhost"; + private int port = 6379; + private SocketPool pool; + + /** + * Constructs a new SRedisConnectionFactory instance + * with default settings. + */ + public SRedisConnectionFactory() { + } + + /** + * Constructs a new SRedisConnectionFactory instance + * with default settings. + */ + public SRedisConnectionFactory(String host, int port) { + this.hostName = host; + this.port = port; + } + + public void afterPropertiesSet() { + pool = new SocketPool(hostName, port); + } + + public void destroy() { + Field f = ReflectionUtils.findField(SocketPool.class, "queue"); + ReflectionUtils.makeAccessible(f); + Queue queue = (Queue) ReflectionUtils.getField(f, pool); + Socket s = null; + do { + s = queue.poll(); + if (s != null) { + try { + s.close(); + } catch (IOException ex) { + // ignore + } + } + } while (s != null); + pool = null; + } + + public RedisConnection getConnection() { + return new SRedisConnection(pool); + } + + public DataAccessException translateExceptionIfPossible(RuntimeException ex) { + return SRedisUtils.convertSRedisAccessException(ex); + } + + /** + * Returns the current host. + * + * @return the host + */ + public String getHostName() { + return hostName; + } + + /** + * Sets the host. + * + * @param host the host to set + */ + public void setHostName(String host) { + this.hostName = host; + } + + /** + * Returns the current port. + * + * @return the port + */ + public int getPort() { + return port; + } + + /** + * Sets the port. + * + * @param port the port to set + */ + public void setPort(int port) { + this.port = port; + } +} \ No newline at end of file diff --git a/src/main/java/org/springframework/data/redis/connection/sredis/SRedisSubscription.java b/src/main/java/org/springframework/data/redis/connection/sredis/SRedisSubscription.java new file mode 100644 index 000000000..06a2db247 --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/sredis/SRedisSubscription.java @@ -0,0 +1,71 @@ +/* + * Copyright 2011-2012 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.sredis; + +import org.springframework.data.redis.connection.MessageListener; +import org.springframework.data.redis.connection.util.AbstractSubscription; + +import redis.client.RedisClient; + +/** + * Message subscription on top of SRP. + * + * @author Costin Leau + */ +class SRedisSubscription extends AbstractSubscription { + + private final RedisClient client; + + SRedisSubscription(MessageListener listener, RedisClient client) { + super(listener); + this.client = client; + } + + protected void doClose() { + client.unsubscribe(null); + client.punsubscribe(null); + } + + + protected void doPsubscribe(byte[]... patterns) { + client.psubscribe(patterns); + } + + + protected void doPUnsubscribe(boolean all, byte[]... patterns) { + if (all) { + client.punsubscribe(null); + } + else { + client.punsubscribe(patterns); + } + } + + protected void doSubscribe(byte[]... channels) { + client.subscribe(channels); + } + + + protected void doUnsubscribe(boolean all, byte[]... channels) { + if (all) { + client.unsubscribe(null); + } + else { + client.unsubscribe(channels); + } + } +} \ No newline at end of file diff --git a/src/main/java/org/springframework/data/redis/connection/sredis/SRedisUtils.java b/src/main/java/org/springframework/data/redis/connection/sredis/SRedisUtils.java new file mode 100644 index 000000000..1a95852fb --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/sredis/SRedisUtils.java @@ -0,0 +1,214 @@ +/* + * Copyright 2011 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.sredis; + +import java.io.StringReader; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Set; + +import org.springframework.dao.DataAccessException; +import org.springframework.data.redis.RedisSystemException; +import org.springframework.data.redis.connection.DefaultTuple; +import org.springframework.data.redis.connection.RedisListCommands.Position; +import org.springframework.data.redis.connection.RedisZSetCommands.Tuple; +import org.springframework.data.redis.connection.SortParameters; +import org.springframework.util.Assert; + +import redis.client.RedisException; +import redis.reply.BulkReply; +import redis.reply.MultiBulkReply; + +import com.google.common.base.Charsets; + +/** + * Helper class featuring methods for SRedis connection handling, providing support for exception translation. + * + * @author Costin Leau + */ +abstract class SRedisUtils { + + private static final byte[] ONE = new byte[] { 1 }; + private static final byte[] ZERO = new byte[] { 0 }; + private static final byte[] BEFORE = "BEFORE".getBytes(Charsets.UTF_8); + private static final byte[] AFTER = "AFTER".getBytes(Charsets.UTF_8); + static final byte[] WITHSCORES = "WITHSCORES".getBytes(Charsets.UTF_8); + private static final byte[] SPACE = "".getBytes(Charsets.UTF_8); + private static final byte[] BY = "BY ".getBytes(Charsets.UTF_8); + private static final byte[] GET = "GET ".getBytes(Charsets.UTF_8); + private static final byte[] ALPHA = "ALPHA ".getBytes(Charsets.UTF_8); + private static final byte[] STORE = "STORE ".getBytes(Charsets.UTF_8); + + + static DataAccessException convertSRedisAccessException(RuntimeException ex) { + if (ex instanceof RedisException) { + return new RedisSystemException("redis exception", ex); + } + return null; + } + + static Properties info(BulkReply reply) { + Properties info = new Properties(); + // use the same charset as the library + StringReader stringReader = new StringReader(new String(reply.bytes, Charsets.UTF_8)); + try { + info.load(stringReader); + } catch (Exception ex) { + throw new RedisSystemException("Cannot read Redis info", ex); + } finally { + stringReader.close(); + } + return info; + } + + static List toBytesList(Object[] byteArrays) { + List list = new ArrayList(byteArrays.length); + for (Object obj : byteArrays) { + if (obj instanceof byte[]) + list.add((byte[]) obj); + else + throw new IllegalArgumentException("array contains more then just bytes" + obj); + } + + return list; + } + + static List toList(T[] byteArrays) { + return Arrays.asList(byteArrays); + } + + static Set toSet(Object[] byteArrays) { + return new LinkedHashSet(toBytesList(byteArrays)); + } + + static byte[][] convert(Map hgetAll) { + byte[][] result = new byte[hgetAll.size() * 2][]; + + int index = 0; + for (Map.Entry entry : hgetAll.entrySet()) { + result[index++] = entry.getKey(); + result[index++] = entry.getValue(); + } + return result; + } + + static byte[] asBit(boolean value) { + return (value ? ONE : ZERO); + } + + static byte[] convertPosition(Position where) { + Assert.notNull("list positions are mandatory"); + return (Position.AFTER.equals(where) ? AFTER : BEFORE); + } + + static Double toDouble(byte[] bytes) { + return Double.valueOf(new String(bytes, Charsets.UTF_8)); + } + + static Long toLong(Object[] byteArrays) { + return Long.valueOf(new String((byte[]) byteArrays[0], Charsets.UTF_8)); + } + + static Set convertTuple(MultiBulkReply zrange) { + Object[] byteArrays = zrange.byteArrays; + Set tuples = new LinkedHashSet(byteArrays.length / 2 + 1); + + for (int i = 0; i < byteArrays.length; i++) { + byte[] value = (byte[]) byteArrays[i]; + i++; + Double score = toDouble((byte[]) byteArrays[i]); + tuples.add(new DefaultTuple(value, score)); + } + + return tuples; + } + + static Map toMap(Object[] byteArrays) { + Map map = new LinkedHashMap(byteArrays.length / 2); + for (int i = 0; i < byteArrays.length; i++) { + map.put((byte[]) byteArrays[i++], (byte[]) byteArrays[i]); + } + return map; + } + + static byte[] limit(long offset, long count) { + return ("LIMIT " + offset + " " + count).getBytes(Charsets.UTF_8); + } + + static byte[] sort(SortParameters params) { + return sort(params, null); + } + + static byte[] sort(SortParameters params, byte[] sortKey) { + List arrays = new ArrayList(); + + if (params.getByPattern() != null) { + arrays.add(BY); + arrays.add(params.getByPattern()); + arrays.add(SPACE); + } + + if (params.getLimit() != null) { + arrays.add(limit(params.getLimit().getStart(), params.getLimit().getCount())); + arrays.add(SPACE); + } + + if (params.getGetPattern() != null) { + byte[][] pattern = params.getGetPattern(); + for (byte[] bs : pattern) { + arrays.add(GET); + arrays.add(bs); + arrays.add(SPACE); + } + } + + if (params.getOrder() != null) { + arrays.add(params.getOrder().name().getBytes(Charsets.UTF_8)); + arrays.add(SPACE); + } + + if (params.isAlphabetic()) { + arrays.add(ALPHA); + } + + if (sortKey != null) { + arrays.add(STORE); + arrays.add(sortKey); + } + + // concatenate array + int size = 0; + + for (byte[] bs : arrays) { + size += bs.length; + } + byte[] result = new byte[size]; + + int index = 0; + for (byte[] bs : arrays) { + System.arraycopy(bs, 0, result, index, bs.length); + index += bs.length; + } + + return result; + } +} \ No newline at end of file diff --git a/src/main/java/org/springframework/data/redis/connection/sredis/package-info.java b/src/main/java/org/springframework/data/redis/connection/sredis/package-info.java new file mode 100644 index 000000000..e151a4393 --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/sredis/package-info.java @@ -0,0 +1,5 @@ +/** + * Connection package for spullara Redis Protocol library. + */ +package org.springframework.data.redis.connection.sredis; + From 9fc7c32117c361d3f6c4646126ae5d0963c4e0d6 Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Fri, 30 Mar 2012 15:34:43 +0300 Subject: [PATCH 2/9] fix casting problems with varargs add integration tests --- .../connection/sredis/SRedisConnection.java | 53 +++++++++--------- .../connection/sredis/SRedisSubscription.java | 16 +++--- .../redis/connection/sredis/SRedisUtils.java | 5 ++ .../srp/SrpConnectionIntegrationTests.java | 55 +++++++++++++++++++ .../collections/CollectionTestParams.java | 33 ++++++++++- 5 files changed, 126 insertions(+), 36 deletions(-) create mode 100644 src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionIntegrationTests.java diff --git a/src/main/java/org/springframework/data/redis/connection/sredis/SRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/sredis/SRedisConnection.java index aa07fb1bd..dace9c2ce 100644 --- a/src/main/java/org/springframework/data/redis/connection/sredis/SRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/sredis/SRedisConnection.java @@ -107,7 +107,6 @@ public class SRedisConnection implements RedisConnection { } } - @SuppressWarnings("unchecked") public List closePipeline() { if (pipeline != null) { ListenableFuture reply = pipeline.exec(); @@ -130,10 +129,10 @@ public class SRedisConnection implements RedisConnection { try { if (isPipelined()) { - pipeline.sort(key, sort, null, null); + pipeline.sort(key, sort, null, (Object[]) null); return null; } - return SRedisUtils.toBytesList(client.sort(key, sort, null, null).byteArrays); + return SRedisUtils.toBytesList(client.sort(key, sort, null, (Object[]) null).byteArrays); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -145,10 +144,10 @@ public class SRedisConnection implements RedisConnection { try { if (isPipelined()) { - pipeline.sort(key, sort, null, null); + pipeline.sort(key, sort, null, (Object[]) null); return null; } - return SRedisUtils.toLong(client.sort(key, sort, null, null).byteArrays); + return SRedisUtils.toLong(client.sort(key, sort, null, (Object[]) null).byteArrays); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -340,10 +339,10 @@ public class SRedisConnection implements RedisConnection { public Long del(byte[]... keys) { try { if (isPipelined()) { - pipeline.del(keys); + pipeline.del((Object[]) keys); return null; } - return client.del(keys).integer; + return client.del((Object[]) keys).integer; } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -640,10 +639,10 @@ public class SRedisConnection implements RedisConnection { public List mGet(byte[]... keys) { try { if (isPipelined()) { - pipeline.mget(keys); + pipeline.mget((Object[]) keys); return null; } - return SRedisUtils.toBytesList(client.mget(keys).byteArrays); + return SRedisUtils.toBytesList(client.mget((Object) keys).byteArrays); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -653,10 +652,10 @@ public class SRedisConnection implements RedisConnection { public void mSet(Map tuples) { try { if (isPipelined()) { - pipeline.mset(SRedisUtils.convert(tuples)); + pipeline.mset((Object[]) SRedisUtils.convert(tuples)); return; } - client.mset(SRedisUtils.convert(tuples)); + client.mset((Object[]) SRedisUtils.convert(tuples)); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -666,10 +665,10 @@ public class SRedisConnection implements RedisConnection { public void mSetNX(Map tuples) { try { if (isPipelined()) { - pipeline.msetnx(SRedisUtils.convert(tuples)); + pipeline.msetnx((Object[]) SRedisUtils.convert(tuples)); return; } - client.msetnx(SRedisUtils.convert(tuples)); + client.msetnx((Object[]) SRedisUtils.convert(tuples)); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1084,10 +1083,10 @@ public class SRedisConnection implements RedisConnection { public Set sDiff(byte[]... keys) { try { if (isPipelined()) { - pipeline.sdiff(keys); + pipeline.sdiff((Object[]) keys); return null; } - return SRedisUtils.toSet(client.sdiff(keys).byteArrays); + return SRedisUtils.toSet(client.sdiff((Object[]) keys).byteArrays); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1097,10 +1096,10 @@ public class SRedisConnection implements RedisConnection { public void sDiffStore(byte[] destKey, byte[]... keys) { try { if (isPipelined()) { - pipeline.sdiffstore(destKey, keys); + pipeline.sdiffstore(destKey, (Object[]) keys); return; } - client.sdiffstore(destKey, keys); + client.sdiffstore(destKey, (Object[]) keys); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1110,10 +1109,10 @@ public class SRedisConnection implements RedisConnection { public Set sInter(byte[]... keys) { try { if (isPipelined()) { - pipeline.sinter(keys); + pipeline.sinter((Object[]) keys); return null; } - return SRedisUtils.toSet(client.sinter(keys).byteArrays); + return SRedisUtils.toSet(client.sinter((Object[]) keys).byteArrays); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1123,10 +1122,10 @@ public class SRedisConnection implements RedisConnection { public void sInterStore(byte[] destKey, byte[]... keys) { try { if (isPipelined()) { - pipeline.sinterstore(destKey, keys); + pipeline.sinterstore(destKey, (Object[]) keys); return; } - client.sinterstore(destKey, keys); + client.sinterstore(destKey, (Object[]) keys); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1214,10 +1213,10 @@ public class SRedisConnection implements RedisConnection { public Set sUnion(byte[]... keys) { try { if (isPipelined()) { - pipeline.sunion(keys); + pipeline.sunion((Object[]) keys); return null; } - return SRedisUtils.toSet(client.sunion(keys).byteArrays); + return SRedisUtils.toSet(client.sunion((Object[]) keys).byteArrays); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1227,10 +1226,10 @@ public class SRedisConnection implements RedisConnection { public void sUnionStore(byte[] destKey, byte[]... keys) { try { if (isPipelined()) { - pipeline.sunionstore(destKey, keys); + pipeline.sunionstore(destKey, (Object[]) keys); return; } - client.sunionstore(destKey, keys); + client.sunionstore(destKey, (Object[]) keys); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1687,10 +1686,10 @@ public class SRedisConnection implements RedisConnection { public List hMGet(byte[] key, byte[]... fields) { try { if (isPipelined()) { - pipeline.hmget(key, fields); + pipeline.hmget(key, (Object[]) fields); return null; } - return SRedisUtils.toBytesList(client.hmget(key, fields).byteArrays); + return SRedisUtils.toBytesList(client.hmget(key, (Object[]) fields).byteArrays); } catch (Exception ex) { throw convertSRAccessException(ex); } diff --git a/src/main/java/org/springframework/data/redis/connection/sredis/SRedisSubscription.java b/src/main/java/org/springframework/data/redis/connection/sredis/SRedisSubscription.java index 06a2db247..fd7ce5977 100644 --- a/src/main/java/org/springframework/data/redis/connection/sredis/SRedisSubscription.java +++ b/src/main/java/org/springframework/data/redis/connection/sredis/SRedisSubscription.java @@ -36,36 +36,36 @@ class SRedisSubscription extends AbstractSubscription { } protected void doClose() { - client.unsubscribe(null); - client.punsubscribe(null); + client.unsubscribe((Object[]) null); + client.punsubscribe((Object[]) null); } protected void doPsubscribe(byte[]... patterns) { - client.psubscribe(patterns); + client.psubscribe((Object[]) patterns); } protected void doPUnsubscribe(boolean all, byte[]... patterns) { if (all) { - client.punsubscribe(null); + client.punsubscribe((Object[]) null); } else { - client.punsubscribe(patterns); + client.punsubscribe((Object[]) patterns); } } protected void doSubscribe(byte[]... channels) { - client.subscribe(channels); + client.subscribe((Object[]) channels); } protected void doUnsubscribe(boolean all, byte[]... channels) { if (all) { - client.unsubscribe(null); + client.unsubscribe((Object[]) null); } else { - client.unsubscribe(channels); + client.unsubscribe((Object[]) channels); } } } \ No newline at end of file diff --git a/src/main/java/org/springframework/data/redis/connection/sredis/SRedisUtils.java b/src/main/java/org/springframework/data/redis/connection/sredis/SRedisUtils.java index 1a95852fb..5019a7a89 100644 --- a/src/main/java/org/springframework/data/redis/connection/sredis/SRedisUtils.java +++ b/src/main/java/org/springframework/data/redis/connection/sredis/SRedisUtils.java @@ -19,6 +19,7 @@ package org.springframework.data.redis.connection.sredis; import java.io.StringReader; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.List; @@ -82,6 +83,10 @@ abstract class SRedisUtils { static List toBytesList(Object[] byteArrays) { List list = new ArrayList(byteArrays.length); + if (byteArrays.length == 1 && byteArrays[0] == null) { + return Collections.emptyList(); + } + for (Object obj : byteArrays) { if (obj instanceof byte[]) list.add((byte[]) obj); diff --git a/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionIntegrationTests.java new file mode 100644 index 000000000..0bf3a8414 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionIntegrationTests.java @@ -0,0 +1,55 @@ +/* + * Copyright 2011-2012 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.srp; + +import org.junit.Test; +import org.springframework.data.redis.SettingsUtils; +import org.springframework.data.redis.connection.AbstractConnectionIntegrationTests; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.connection.sredis.SRedisConnectionFactory; + +import redis.client.RedisClient; + +/** + * @author Costin Leau + */ +public class SrpConnectionIntegrationTests extends AbstractConnectionIntegrationTests { + SRedisConnectionFactory factory; + + public SrpConnectionIntegrationTests() { + factory = new SRedisConnectionFactory(); + factory.setPort(SettingsUtils.getPort()); + factory.setHostName(SettingsUtils.getHost()); + + factory.afterPropertiesSet(); + } + + + protected RedisConnectionFactory getConnectionFactory() { + return factory; + } + + @Test + public void testRaw() throws Exception { + RedisClient rc = (RedisClient) factory.getConnection().getNativeConnection(); + + System.out.println(rc.dbsize()); + System.out.println(rc.exists("foobar")); + rc.set("foobar", "barfoo"); + System.out.println(rc.get("foobar")); + } +} diff --git a/src/test/java/org/springframework/data/redis/support/collections/CollectionTestParams.java b/src/test/java/org/springframework/data/redis/support/collections/CollectionTestParams.java index 0108cda85..c35cc0e55 100644 --- a/src/test/java/org/springframework/data/redis/support/collections/CollectionTestParams.java +++ b/src/test/java/org/springframework/data/redis/support/collections/CollectionTestParams.java @@ -23,6 +23,7 @@ import org.springframework.data.redis.SettingsUtils; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; import org.springframework.data.redis.connection.jredis.JredisConnectionFactory; import org.springframework.data.redis.connection.rjc.RjcConnectionFactory; +import org.springframework.data.redis.connection.sredis.SRedisConnectionFactory; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.serializer.JacksonJsonRedisSerializer; @@ -135,6 +136,32 @@ public abstract class CollectionTestParams { jsonPersonTemplateRJC.setConnectionFactory(rjcConnFactory); jsonPersonTemplateRJC.afterPropertiesSet(); + // SRP + SRedisConnectionFactory srConnFactory = new SRedisConnectionFactory(); + srConnFactory.setPort(SettingsUtils.getPort()); + srConnFactory.setHostName(SettingsUtils.getHost()); + srConnFactory.afterPropertiesSet(); + + RedisTemplate stringTemplateSRP = new StringRedisTemplate(srConnFactory); + RedisTemplate personTemplateSRP = new RedisTemplate(); + personTemplateSRP.setConnectionFactory(srConnFactory); + personTemplateSRP.afterPropertiesSet(); + + RedisTemplate xstreamStringTemplateSRP = new RedisTemplate(); + xstreamStringTemplateSRP.setConnectionFactory(srConnFactory); + xstreamStringTemplateSRP.setDefaultSerializer(serializer); + xstreamStringTemplateSRP.afterPropertiesSet(); + + RedisTemplate xstreamPersonTemplateSRP = new RedisTemplate(); + xstreamPersonTemplateSRP.setValueSerializer(serializer); + xstreamPersonTemplateSRP.setConnectionFactory(srConnFactory); + xstreamPersonTemplateSRP.afterPropertiesSet(); + + RedisTemplate jsonPersonTemplateSRP = new RedisTemplate(); + jsonPersonTemplateSRP.setValueSerializer(jsonSerializer); + jsonPersonTemplateSRP.setConnectionFactory(srConnFactory); + jsonPersonTemplateSRP.afterPropertiesSet(); + return Arrays.asList(new Object[][] { { stringFactory, stringTemplate }, { stringFactory, stringTemplateRJC }, { personFactory, personTemplateRJC }, //{ stringFactory, stringTemplateJR }, @@ -146,6 +173,10 @@ public abstract class CollectionTestParams { { personFactory, jsonPersonTemplate }, //{ personFactory, jsonPersonTemplateJR }, { stringFactory, xstreamStringTemplateRJC }, { personFactory, xstreamPersonTemplateRJC }, - { personFactory, jsonPersonTemplateRJC } }); + { personFactory, jsonPersonTemplateRJC }, + { stringFactory, stringTemplateSRP },{ personFactory, personTemplateSRP }, + { stringFactory, xstreamStringTemplateSRP }, { personFactory, xstreamPersonTemplateSRP }, + { personFactory, jsonPersonTemplateSRP } + }); } } From 5ca56328992f9cf67d8dcadf80ac474b3fe5e40f Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Sun, 1 Apr 2012 19:53:28 +0300 Subject: [PATCH 3/9] update to srp 0.2 snapshot except for zset and pubsub ops, the rest of the tests seem to be running --- .classpath | 4 +- .settings/org.eclipse.jdt.core.prefs | 2 +- gradle.properties | 2 +- .../SrpConnection.java} | 333 +++++++++--------- .../SrpConnectionFactory.java} | 41 +-- .../SrpSubscription.java} | 6 +- .../SRedisUtils.java => srp/SrpUtils.java} | 42 +-- .../{sredis => srp}/package-info.java | 2 +- .../AbstractConnectionIntegrationTests.java | 13 +- .../srp/SrpConnectionIntegrationTests.java | 8 +- .../collections/CollectionTestParams.java | 4 +- 11 files changed, 229 insertions(+), 228 deletions(-) rename src/main/java/org/springframework/data/redis/connection/{sredis/SRedisConnection.java => srp/SrpConnection.java} (77%) rename src/main/java/org/springframework/data/redis/connection/{sredis/SRedisConnectionFactory.java => srp/SrpConnectionFactory.java} (71%) rename src/main/java/org/springframework/data/redis/connection/{sredis/SRedisSubscription.java => srp/SrpSubscription.java} (89%) rename src/main/java/org/springframework/data/redis/connection/{sredis/SRedisUtils.java => srp/SrpUtils.java} (85%) rename src/main/java/org/springframework/data/redis/connection/{sredis => srp}/package-info.java (68%) diff --git a/.classpath b/.classpath index bbf9837aa..a316b6684 100644 --- a/.classpath +++ b/.classpath @@ -12,7 +12,6 @@ - @@ -20,8 +19,9 @@ - + + diff --git a/.settings/org.eclipse.jdt.core.prefs b/.settings/org.eclipse.jdt.core.prefs index b69a99a38..9377866ca 100644 --- a/.settings/org.eclipse.jdt.core.prefs +++ b/.settings/org.eclipse.jdt.core.prefs @@ -1,5 +1,5 @@ # -#Fri Mar 30 10:10:50 EEST 2012 +#Sun Apr 01 10:03:27 EEST 2012 org.eclipse.jdt.core.compiler.debug.localVariable=generate org.eclipse.jdt.core.compiler.compliance=1.5 org.eclipse.jdt.core.compiler.codegen.unusedLocal=preserve diff --git a/gradle.properties b/gradle.properties index 9faebb341..07999da81 100644 --- a/gradle.properties +++ b/gradle.properties @@ -16,7 +16,7 @@ mockitoVersion = 1.8.5 jedisVersion = 2.0.0 jredisVersion = 03122010 rjcVersion = 0.6.4 -sredisVersion = 0.1 +sredisVersion = 0.2-SNAPSHOT # Manifest properties diff --git a/src/main/java/org/springframework/data/redis/connection/sredis/SRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java similarity index 77% rename from src/main/java/org/springframework/data/redis/connection/sredis/SRedisConnection.java rename to src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java index dace9c2ce..bd07a475d 100644 --- a/src/main/java/org/springframework/data/redis/connection/sredis/SRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.data.redis.connection.sredis; +package org.springframework.data.redis.connection.srp; import java.io.IOException; import java.util.Collections; @@ -21,6 +21,7 @@ import java.util.List; import java.util.Map; import java.util.Properties; import java.util.Set; +import java.util.concurrent.BlockingQueue; import org.springframework.dao.DataAccessException; import org.springframework.data.redis.RedisConnectionFailureException; @@ -35,27 +36,28 @@ import org.springframework.data.redis.connection.Subscription; import redis.client.RedisClient; import redis.client.RedisClient.Pipeline; import redis.client.RedisException; -import redis.client.SocketPool; -import redis.reply.MultiBulkReply; -import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.base.Charsets; /** * {@code RedisConnection} implementation on top of spullara Redis Protocol library. * * @author Costin Leau */ -public class SRedisConnection implements RedisConnection { +public class SrpConnection implements RedisConnection { + + private final RedisClient client; + private final BlockingQueue queue; private boolean isClosed = false; private boolean isMulti = false; - private final RedisClient client; private Pipeline pipeline; - private volatile SRedisSubscription subscription; + private volatile SrpSubscription subscription; - public SRedisConnection(SocketPool pool) { + public SrpConnection(String host, int port, BlockingQueue queue) { try { - this.client = new RedisClient(pool); + this.client = new RedisClient(host, port); + this.queue = queue; } catch (IOException e) { throw new RedisConnectionFailureException("Could not connect", e); } @@ -63,7 +65,7 @@ public class SRedisConnection implements RedisConnection { protected DataAccessException convertSRAccessException(Exception ex) { if (ex instanceof RedisException) { - return SRedisUtils.convertSRedisAccessException((RedisException) ex); + return SrpUtils.convertSRedisAccessException((RedisException) ex); } if (ex instanceof IOException) { return new RedisConnectionFailureException("Redis connection failed", (IOException) ex); @@ -75,6 +77,7 @@ public class SRedisConnection implements RedisConnection { public void close() throws DataAccessException { isClosed = true; + queue.remove(this); try { client.close(); @@ -108,31 +111,32 @@ public class SRedisConnection implements RedisConnection { } public List closePipeline() { - if (pipeline != null) { - ListenableFuture reply = pipeline.exec(); - pipeline = null; - if (reply != null) { - try { - return SRedisUtils.toList(reply.get().byteArrays); - } catch (Exception ex) { - throw convertSRAccessException(ex); - } - } - } - return Collections.emptyList(); + // if (pipeline != null) { + // //ListenableFuture reply = pipeline.exec(); + // pipeline = null; + // if (reply != null) { + // try { + // return SrpUtils.toList(reply.get().data()); + // } catch (Exception ex) { + // throw convertSRAccessException(ex); + // } + // } + // } + throw new UnsupportedOperationException(); + //return Collections.emptyList(); } public List sort(byte[] key, SortParameters params) { - byte[] sort = SRedisUtils.sort(params); + byte[] sort = SrpUtils.sort(params); try { if (isPipelined()) { pipeline.sort(key, sort, null, (Object[]) null); return null; } - return SRedisUtils.toBytesList(client.sort(key, sort, null, (Object[]) null).byteArrays); + return SrpUtils.toBytesList(client.sort(key, sort, null, (Object[]) null).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -140,14 +144,14 @@ public class SRedisConnection implements RedisConnection { public Long sort(byte[] key, SortParameters params, byte[] sortKey) { - byte[] sort = SRedisUtils.sort(params, sortKey); + byte[] sort = SrpUtils.sort(params, sortKey); try { if (isPipelined()) { pipeline.sort(key, sort, null, (Object[]) null); return null; } - return SRedisUtils.toLong(client.sort(key, sort, null, (Object[]) null).byteArrays); + return SrpUtils.toLong(client.sort(key, sort, null, (Object[]) null).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -159,7 +163,7 @@ public class SRedisConnection implements RedisConnection { pipeline.dbsize(); return null; } - return client.dbsize().integer; + return client.dbsize().data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -251,7 +255,7 @@ public class SRedisConnection implements RedisConnection { pipeline.info(); return null; } - return SRedisUtils.info(client.info()); + return SrpUtils.info(client.info()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -264,7 +268,7 @@ public class SRedisConnection implements RedisConnection { pipeline.lastsave(); return null; } - return client.lastsave().integer; + return client.lastsave().data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -299,12 +303,13 @@ public class SRedisConnection implements RedisConnection { public void shutdown() { + byte[] save = "SAVE".getBytes(Charsets.UTF_8); try { if (isPipelined()) { - pipeline.shutdown(); + pipeline.shutdown(save, null); return; } - client.shutdown(); + client.shutdown(save, null); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -317,7 +322,7 @@ public class SRedisConnection implements RedisConnection { pipeline.echo(message); return null; } - return client.echo(message).bytes; + return client.echo(message).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -329,7 +334,7 @@ public class SRedisConnection implements RedisConnection { if (isPipelined()) { pipeline.ping(); } - return client.ping().status; + return client.ping().data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -342,7 +347,7 @@ public class SRedisConnection implements RedisConnection { pipeline.del((Object[]) keys); return null; } - return client.del((Object[]) keys).integer; + return client.del((Object[]) keys).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -353,8 +358,9 @@ public class SRedisConnection implements RedisConnection { isMulti = false; try { if (isPipelined()) { - pipeline.discard(); - return; + //pipeline.discard(); + //return; + throw new UnsupportedOperationException(); } client.discard(); @@ -368,10 +374,12 @@ public class SRedisConnection implements RedisConnection { isMulti = false; try { if (isPipelined()) { - pipeline.exec(); - return null; + // pipeline.exec(); + // return null; + throw new UnsupportedOperationException(); } - return SRedisUtils.toList(client.exec().byteArrays); + // return SrpUtils.toList(client.exec()); + throw new UnsupportedOperationException(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -384,7 +392,7 @@ public class SRedisConnection implements RedisConnection { pipeline.exists(key); return null; } - return client.exists(key).integer == 1; + return client.exists(key).data() == 1; } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -397,7 +405,7 @@ public class SRedisConnection implements RedisConnection { pipeline.expire(key, seconds); return null; } - return client.expire(key, seconds).integer == 1; + return client.expire(key, seconds).data() == 1; } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -410,7 +418,7 @@ public class SRedisConnection implements RedisConnection { pipeline.expireat(key, unixTime); return null; } - return client.expireat(key, unixTime).integer == 1; + return client.expireat(key, unixTime).data() == 1; } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -423,7 +431,7 @@ public class SRedisConnection implements RedisConnection { pipeline.keys(pattern); return null; } - return SRedisUtils.toSet(client.keys(pattern).byteArrays); + return SrpUtils.toSet(client.keys(pattern).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -437,8 +445,9 @@ public class SRedisConnection implements RedisConnection { isMulti = true; try { if (isPipelined()) { - pipeline.multi(); - return; + // pipeline.multi(); + // return; + throw new UnsupportedOperationException(); } client.multi(); } catch (Exception ex) { @@ -453,7 +462,7 @@ public class SRedisConnection implements RedisConnection { pipeline.persist(key); return null; } - return client.persist(key).integer == 1; + return client.persist(key).data() == 1; } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -466,7 +475,7 @@ public class SRedisConnection implements RedisConnection { pipeline.move(key, dbIndex); return null; } - return client.move(key, dbIndex).integer == 1; + return client.move(key, dbIndex).data() == 1; } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -479,7 +488,7 @@ public class SRedisConnection implements RedisConnection { pipeline.randomkey(); return null; } - return client.randomkey().bytes; + return client.randomkey().data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -505,7 +514,7 @@ public class SRedisConnection implements RedisConnection { pipeline.renamenx(oldName, newName); return null; } - return (client.renamenx(oldName, newName).integer == 1); + return (client.renamenx(oldName, newName).data() == 1); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -530,7 +539,7 @@ public class SRedisConnection implements RedisConnection { pipeline.ttl(key); return null; } - return client.ttl(key).integer; + return client.ttl(key).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -543,7 +552,7 @@ public class SRedisConnection implements RedisConnection { pipeline.type(key); return null; } - return DataType.fromCode(client.type(key).status); + return DataType.fromCode(client.type(key).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -564,13 +573,11 @@ public class SRedisConnection implements RedisConnection { throw new UnsupportedOperationException(); } try { - for (byte[] key : keys) { - if (isPipelined()) { - pipeline.watch(key); - } - else { - client.watch(key); - } + if (isPipelined()) { + pipeline.watch((Object[]) keys); + } + else { + client.watch((Object[]) keys); } } catch (Exception ex) { throw convertSRAccessException(ex); @@ -589,7 +596,7 @@ public class SRedisConnection implements RedisConnection { return null; } - return client.get(key).bytes; + return client.get(key).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -616,7 +623,7 @@ public class SRedisConnection implements RedisConnection { pipeline.getset(key, value); return null; } - return client.getset(key, value).bytes; + return client.getset(key, value).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -629,7 +636,7 @@ public class SRedisConnection implements RedisConnection { pipeline.append(key, value); return null; } - return client.append(key, value).integer; + return client.append(key, value).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -642,7 +649,7 @@ public class SRedisConnection implements RedisConnection { pipeline.mget((Object[]) keys); return null; } - return SRedisUtils.toBytesList(client.mget((Object) keys).byteArrays); + return SrpUtils.toBytesList(client.mget((Object[]) keys).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -652,10 +659,10 @@ public class SRedisConnection implements RedisConnection { public void mSet(Map tuples) { try { if (isPipelined()) { - pipeline.mset((Object[]) SRedisUtils.convert(tuples)); + pipeline.mset((Object[]) SrpUtils.convert(tuples)); return; } - client.mset((Object[]) SRedisUtils.convert(tuples)); + client.mset((Object[]) SrpUtils.convert(tuples)); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -665,10 +672,10 @@ public class SRedisConnection implements RedisConnection { public void mSetNX(Map tuples) { try { if (isPipelined()) { - pipeline.msetnx((Object[]) SRedisUtils.convert(tuples)); + pipeline.msetnx((Object[]) SrpUtils.convert(tuples)); return; } - client.msetnx((Object[]) SRedisUtils.convert(tuples)); + client.msetnx((Object[]) SrpUtils.convert(tuples)); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -694,7 +701,7 @@ public class SRedisConnection implements RedisConnection { pipeline.setnx(key, value); return null; } - return client.setnx(key, value).integer == 1; + return client.setnx(key, value).data() == 1; } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -707,7 +714,7 @@ public class SRedisConnection implements RedisConnection { pipeline.getrange(key, start, end); return null; } - return client.getrange(key, start, end).bytes; + return client.getrange(key, start, end).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -720,7 +727,7 @@ public class SRedisConnection implements RedisConnection { pipeline.decr(key); return null; } - return client.decr(key).integer; + return client.decr(key).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -733,7 +740,7 @@ public class SRedisConnection implements RedisConnection { pipeline.decrby(key, value); return null; } - return client.decrby(key, value).integer; + return client.decrby(key, value).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -746,7 +753,7 @@ public class SRedisConnection implements RedisConnection { pipeline.incr(key); return null; } - return client.incr(key).integer; + return client.incr(key).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -759,7 +766,7 @@ public class SRedisConnection implements RedisConnection { pipeline.incrby(key, value); return null; } - return client.incrby(key, value).integer; + return client.incrby(key, value).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -774,7 +781,7 @@ public class SRedisConnection implements RedisConnection { if (isPipelined()) { throw new UnsupportedOperationException(); } - return (client.getbit(key, offset).integer == 1); + return (client.getbit(key, offset).data() == 1); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -789,7 +796,7 @@ public class SRedisConnection implements RedisConnection { if (isPipelined()) { throw new UnsupportedOperationException(); } - client.setbit(key, offset, SRedisUtils.asBit(value)); + client.setbit(key, offset, SrpUtils.asBit(value)); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -817,7 +824,7 @@ public class SRedisConnection implements RedisConnection { pipeline.strlen(key); return null; } - return client.strlen(key).integer; + return client.strlen(key).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -831,10 +838,10 @@ public class SRedisConnection implements RedisConnection { public Long lPush(byte[] key, byte[] value) { try { if (isPipelined()) { - pipeline.lpush(key, value); + pipeline.lpush(key, new Object[] { value }); return null; } - return client.lpush(key, value).integer; + return client.lpush(key, new Object[] { value }).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -844,10 +851,10 @@ public class SRedisConnection implements RedisConnection { public Long rPush(byte[] key, byte[] value) { try { if (isPipelined()) { - pipeline.rpush(key, value); + pipeline.rpush(key, new Object[] { value }); return null; } - return client.rpush(key, value).integer; + return client.rpush(key, new Object[] { value }).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -857,10 +864,11 @@ public class SRedisConnection implements RedisConnection { public List bLPop(int timeout, byte[]... keys) { try { if (isPipelined()) { - pipeline.blpop(timeout, keys); + // pipeline.blpop(timeout, keys); return null; } - return SRedisUtils.toBytesList(client.blpop(timeout, keys).byteArrays); + // return SrpUtils.toBytesList(client.blpop(timeout, keys).data()); + throw new UnsupportedOperationException(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -870,10 +878,11 @@ public class SRedisConnection implements RedisConnection { public List bRPop(int timeout, byte[]... keys) { try { if (isPipelined()) { - pipeline.brpop(timeout, keys); + // pipeline.brpop(timeout, keys); return null; } - return SRedisUtils.toBytesList(client.brpop(timeout, keys).byteArrays); + // return SrpUtils.toBytesList(client.brpop(timeout, keys).data()); + throw new UnsupportedOperationException(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -886,7 +895,7 @@ public class SRedisConnection implements RedisConnection { pipeline.lindex(key, index); return null; } - return client.lindex(key, index).bytes; + return client.lindex(key, index).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -896,10 +905,10 @@ public class SRedisConnection implements RedisConnection { public Long lInsert(byte[] key, Position where, byte[] pivot, byte[] value) { try { if (isPipelined()) { - pipeline.linsert(key, SRedisUtils.convertPosition(where), pivot, value); + pipeline.linsert(key, SrpUtils.convertPosition(where), pivot, value); return null; } - return client.linsert(key, SRedisUtils.convertPosition(where), pivot, value).integer; + return client.linsert(key, SrpUtils.convertPosition(where), pivot, value).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -912,7 +921,7 @@ public class SRedisConnection implements RedisConnection { pipeline.llen(key); return null; } - return client.llen(key).integer; + return client.llen(key).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -925,7 +934,7 @@ public class SRedisConnection implements RedisConnection { pipeline.lpop(key); return null; } - return client.lpop(key).bytes; + return client.lpop(key).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -938,7 +947,7 @@ public class SRedisConnection implements RedisConnection { pipeline.lrange(key, start, end); return null; } - return SRedisUtils.toBytesList(client.lrange(key, start, end).byteArrays); + return SrpUtils.toBytesList(client.lrange(key, start, end).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -951,7 +960,7 @@ public class SRedisConnection implements RedisConnection { pipeline.lrem(key, count, value); return null; } - return client.lrem(key, count, value).integer; + return client.lrem(key, count, value).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -990,7 +999,7 @@ public class SRedisConnection implements RedisConnection { pipeline.rpop(key); return null; } - return client.rpop(key).bytes; + return client.rpop(key).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1003,7 +1012,7 @@ public class SRedisConnection implements RedisConnection { pipeline.rpoplpush(srcKey, dstKey); return null; } - return client.rpoplpush(srcKey, dstKey).bytes; + return client.rpoplpush(srcKey, dstKey).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1016,7 +1025,7 @@ public class SRedisConnection implements RedisConnection { pipeline.brpoplpush(srcKey, dstKey, timeout); return null; } - return client.brpoplpush(srcKey, dstKey, timeout).bytes; + return client.brpoplpush(srcKey, dstKey, timeout).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1029,7 +1038,7 @@ public class SRedisConnection implements RedisConnection { pipeline.lpushx(key, value); return null; } - return client.lpushx(key, value).integer; + return client.lpushx(key, value).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1042,7 +1051,7 @@ public class SRedisConnection implements RedisConnection { pipeline.rpushx(key, value); return null; } - return client.rpushx(key, value).integer; + return client.rpushx(key, value).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1057,10 +1066,10 @@ public class SRedisConnection implements RedisConnection { public Boolean sAdd(byte[] key, byte[] value) { try { if (isPipelined()) { - pipeline.sadd(key, value); + pipeline.sadd(key, new Object[] { value }); return null; } - return (client.sadd(key, value).integer == 1); + return (client.sadd(key, new Object[] { value }).data() == 1); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1073,7 +1082,7 @@ public class SRedisConnection implements RedisConnection { pipeline.scard(key); return null; } - return client.scard(key).integer; + return client.scard(key).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1086,7 +1095,7 @@ public class SRedisConnection implements RedisConnection { pipeline.sdiff((Object[]) keys); return null; } - return SRedisUtils.toSet(client.sdiff((Object[]) keys).byteArrays); + return SrpUtils.toSet(client.sdiff((Object[]) keys).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1112,7 +1121,7 @@ public class SRedisConnection implements RedisConnection { pipeline.sinter((Object[]) keys); return null; } - return SRedisUtils.toSet(client.sinter((Object[]) keys).byteArrays); + return SrpUtils.toSet(client.sinter((Object[]) keys).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1138,7 +1147,7 @@ public class SRedisConnection implements RedisConnection { pipeline.sismember(key, value); return null; } - return client.sismember(key, value).integer == 1; + return client.sismember(key, value).data() == 1; } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1151,7 +1160,7 @@ public class SRedisConnection implements RedisConnection { pipeline.smembers(key); return null; } - return SRedisUtils.toSet(client.smembers(key).byteArrays); + return SrpUtils.toSet(client.smembers(key).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1164,7 +1173,7 @@ public class SRedisConnection implements RedisConnection { pipeline.smove(srcKey, destKey, value); return null; } - return client.smove(srcKey, destKey, value).integer == 1; + return client.smove(srcKey, destKey, value).data() == 1; } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1177,7 +1186,7 @@ public class SRedisConnection implements RedisConnection { pipeline.spop(key); return null; } - return client.spop(key).bytes; + return client.spop(key).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1190,7 +1199,7 @@ public class SRedisConnection implements RedisConnection { pipeline.srandmember(key); return null; } - return client.srandmember(key).bytes; + return client.srandmember(key).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1200,10 +1209,10 @@ public class SRedisConnection implements RedisConnection { public Boolean sRem(byte[] key, byte[] value) { try { if (isPipelined()) { - pipeline.srem(key, value); + pipeline.srem(key, new Object[] { value }); return null; } - return client.srem(key, value).integer == 1; + return client.srem(key, new Object[] { value }).data() == 1; } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1216,7 +1225,7 @@ public class SRedisConnection implements RedisConnection { pipeline.sunion((Object[]) keys); return null; } - return SRedisUtils.toSet(client.sunion((Object[]) keys).byteArrays); + return SrpUtils.toSet(client.sunion((Object[]) keys).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1246,7 +1255,7 @@ public class SRedisConnection implements RedisConnection { pipeline.zadd(key, score, value, null, null); return null; } - return client.zadd(key, score, value, null, null).integer == 1; + return client.zadd(key, score, value, null, null).data() == 1; } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1259,7 +1268,7 @@ public class SRedisConnection implements RedisConnection { pipeline.zcard(key); return null; } - return client.zcard(key).integer; + return client.zcard(key).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1272,7 +1281,7 @@ public class SRedisConnection implements RedisConnection { pipeline.zcount(key, min, max); return null; } - return client.zcount(key, min, max).integer; + return client.zcount(key, min, max).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1285,7 +1294,7 @@ public class SRedisConnection implements RedisConnection { pipeline.zincrby(key, increment, value); return null; } - return SRedisUtils.toDouble(client.zincrby(key, increment, value).bytes); + return SrpUtils.toDouble(client.zincrby(key, increment, value).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1300,10 +1309,10 @@ public class SRedisConnection implements RedisConnection { public Long zInterStore(byte[] destKey, byte[]... sets) { try { if (isQueueing()) { - pipeline.zinterstore(destKey, sets); + pipeline.zinterstore(destKey, sets.length, (Object[]) sets); return null; } - return client.zinterstore(destKey, sets).integer; + return client.zinterstore(destKey, sets.length, (Object[]) sets).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1316,7 +1325,7 @@ public class SRedisConnection implements RedisConnection { pipeline.zrange(key, start, end, null); return null; } - return SRedisUtils.toSet(client.zrange(key, start, end, null).byteArrays); + return SrpUtils.toSet(client.zrange(key, start, end, null).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1326,10 +1335,10 @@ public class SRedisConnection implements RedisConnection { public Set zRangeWithScores(byte[] key, long start, long end) { try { if (isPipelined()) { - pipeline.zrange(key, start, end, SRedisUtils.WITHSCORES); + pipeline.zrange(key, start, end, SrpUtils.WITHSCORES); return null; } - return SRedisUtils.convertTuple(client.zrange(key, start, end, SRedisUtils.WITHSCORES)); + return SrpUtils.convertTuple(client.zrange(key, start, end, SrpUtils.WITHSCORES)); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1342,7 +1351,7 @@ public class SRedisConnection implements RedisConnection { pipeline.zrangebyscore(key, min, max, null, null); return null; } - return SRedisUtils.toSet(client.zrangebyscore(key, min, max, null, null).byteArrays); + return SrpUtils.toSet(client.zrangebyscore(key, min, max, null, null).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1352,10 +1361,10 @@ public class SRedisConnection implements RedisConnection { public Set zRangeByScoreWithScores(byte[] key, double min, double max) { try { if (isPipelined()) { - pipeline.zrangebyscore(key, min, max, SRedisUtils.WITHSCORES, null); + pipeline.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, null); return null; } - return SRedisUtils.convertTuple(client.zrangebyscore(key, min, max, SRedisUtils.WITHSCORES, null)); + return SrpUtils.convertTuple(client.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, null)); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1365,10 +1374,10 @@ public class SRedisConnection implements RedisConnection { public Set zRevRangeWithScores(byte[] key, long start, long end) { try { if (isPipelined()) { - pipeline.zrevrange(key, start, end, SRedisUtils.WITHSCORES); + pipeline.zrevrange(key, start, end, SrpUtils.WITHSCORES); return null; } - return SRedisUtils.convertTuple(client.zrevrange(key, start, end, SRedisUtils.WITHSCORES)); + return SrpUtils.convertTuple(client.zrevrange(key, start, end, SrpUtils.WITHSCORES)); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1377,12 +1386,12 @@ public class SRedisConnection implements RedisConnection { public Set zRangeByScore(byte[] key, double min, double max, long offset, long count) { try { - byte[] limit = SRedisUtils.limit(offset, count); + byte[] limit = SrpUtils.limit(offset, count); if (isPipelined()) { pipeline.zrangebyscore(key, min, max, null, limit); return null; } - return SRedisUtils.toSet(client.zrangebyscore(key, min, max, null, limit).byteArrays); + return SrpUtils.toSet(client.zrangebyscore(key, min, max, null, limit).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1391,12 +1400,12 @@ public class SRedisConnection implements RedisConnection { public Set zRangeByScoreWithScores(byte[] key, double min, double max, long offset, long count) { try { - byte[] limit = SRedisUtils.limit(offset, count); + byte[] limit = SrpUtils.limit(offset, count); if (isPipelined()) { - pipeline.zrangebyscore(key, min, max, SRedisUtils.WITHSCORES, limit); + pipeline.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, limit); return null; } - return SRedisUtils.convertTuple(client.zrangebyscore(key, min, max, SRedisUtils.WITHSCORES, limit)); + return SrpUtils.convertTuple(client.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, limit)); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1405,11 +1414,11 @@ public class SRedisConnection implements RedisConnection { public Set zRevRangeByScore(byte[] key, double min, double max, long offset, long count) { try { - byte[] limit = SRedisUtils.limit(offset, count); + byte[] limit = SrpUtils.limit(offset, count); if (isPipelined()) { client.zrevrangebyscore(key, min, max, null, limit); } - return SRedisUtils.toSet(client.zrevrangebyscore(key, min, max, null, limit).byteArrays); + return SrpUtils.toSet(client.zrevrangebyscore(key, min, max, null, limit).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1421,7 +1430,7 @@ public class SRedisConnection implements RedisConnection { if (isPipelined()) { client.zrevrangebyscore(key, min, max, null, null); } - return SRedisUtils.toSet(client.zrevrangebyscore(key, min, max, null, null).byteArrays); + return SrpUtils.toSet(client.zrevrangebyscore(key, min, max, null, null).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1430,11 +1439,11 @@ public class SRedisConnection implements RedisConnection { public Set zRevRangeByScoreWithScores(byte[] key, double min, double max, long offset, long count) { try { - byte[] limit = SRedisUtils.limit(offset, count); + byte[] limit = SrpUtils.limit(offset, count); if (isPipelined()) { - client.zrevrangebyscore(key, min, max, SRedisUtils.WITHSCORES, limit); + client.zrevrangebyscore(key, min, max, SrpUtils.WITHSCORES, limit); } - return SRedisUtils.convertTuple(client.zrevrangebyscore(key, min, max, SRedisUtils.WITHSCORES, limit)); + return SrpUtils.convertTuple(client.zrevrangebyscore(key, min, max, SrpUtils.WITHSCORES, limit)); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1444,9 +1453,9 @@ public class SRedisConnection implements RedisConnection { public Set zRevRangeByScoreWithScores(byte[] key, double min, double max) { try { if (isPipelined()) { - client.zrevrangebyscore(key, min, max, SRedisUtils.WITHSCORES, null); + client.zrevrangebyscore(key, min, max, SrpUtils.WITHSCORES, null); } - return SRedisUtils.convertTuple(client.zrevrangebyscore(key, min, max, SRedisUtils.WITHSCORES, null)); + return SrpUtils.convertTuple(client.zrevrangebyscore(key, min, max, SrpUtils.WITHSCORES, null)); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1459,7 +1468,7 @@ public class SRedisConnection implements RedisConnection { pipeline.zrank(key, value); return null; } - return client.zrank(key, value).integer; + return client.zrank(key, value).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1469,10 +1478,10 @@ public class SRedisConnection implements RedisConnection { public Boolean zRem(byte[] key, byte[] value) { try { if (isPipelined()) { - pipeline.zrem(key, value); + pipeline.zrem(key, new Object[] { value }); return null; } - return client.zrem(key, value).integer == 1; + return client.zrem(key, new Object[] { value }).data() == 1; } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1485,7 +1494,7 @@ public class SRedisConnection implements RedisConnection { pipeline.zremrangebyrank(key, start, end); return null; } - return client.zremrangebyrank(key, start, end).integer; + return client.zremrangebyrank(key, start, end).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1498,7 +1507,7 @@ public class SRedisConnection implements RedisConnection { pipeline.zremrangebyscore(key, min, max); return null; } - return client.zremrangebyscore(key, min, max).integer; + return client.zremrangebyscore(key, min, max).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1511,7 +1520,7 @@ public class SRedisConnection implements RedisConnection { pipeline.zrevrange(key, start, end, null); return null; } - return SRedisUtils.toSet(client.zrevrange(key, start, end, null).byteArrays); + return SrpUtils.toSet(client.zrevrange(key, start, end, null).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1524,7 +1533,7 @@ public class SRedisConnection implements RedisConnection { pipeline.zrevrank(key, value); return null; } - return client.zrevrank(key, value).integer; + return client.zrevrank(key, value).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1537,7 +1546,7 @@ public class SRedisConnection implements RedisConnection { pipeline.zscore(key, value); return null; } - return SRedisUtils.toDouble(client.zscore(key, value).bytes); + return SrpUtils.toDouble(client.zscore(key, value).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1552,10 +1561,10 @@ public class SRedisConnection implements RedisConnection { public Long zUnionStore(byte[] destKey, byte[]... sets) { try { if (isPipelined()) { - pipeline.zunionstore(destKey, sets); + pipeline.zunionstore(destKey, sets.length, (Object[]) sets); return null; } - return client.zunionstore(destKey, sets).integer; + return client.zunionstore(destKey, sets.length, (Object[]) sets).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1572,7 +1581,7 @@ public class SRedisConnection implements RedisConnection { pipeline.hset(key, field, value); return null; } - return client.hset(key, field, value).integer == 1; + return client.hset(key, field, value).data() == 1; } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1585,7 +1594,7 @@ public class SRedisConnection implements RedisConnection { pipeline.hsetnx(key, field, value); return null; } - return client.hsetnx(key, field, value).integer == 1; + return client.hsetnx(key, field, value).data() == 1; } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1595,10 +1604,10 @@ public class SRedisConnection implements RedisConnection { public Boolean hDel(byte[] key, byte[] field) { try { if (isPipelined()) { - pipeline.hdel(key, field); + pipeline.hdel(key, new Object[] { field }); return null; } - return client.hdel(key, field).integer == 1; + return client.hdel(key, new Object[] { field }).data() == 1; } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1611,7 +1620,7 @@ public class SRedisConnection implements RedisConnection { pipeline.hexists(key, field); return null; } - return client.hexists(key, field).integer == 1; + return client.hexists(key, field).data() == 1; } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1624,7 +1633,7 @@ public class SRedisConnection implements RedisConnection { pipeline.hget(key, field); return null; } - return client.hget(key, field).bytes; + return client.hget(key, field).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1637,7 +1646,7 @@ public class SRedisConnection implements RedisConnection { pipeline.hgetall(key); return null; } - return SRedisUtils.toMap(client.hgetall(key).byteArrays); + return SrpUtils.toMap(client.hgetall(key).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1650,7 +1659,7 @@ public class SRedisConnection implements RedisConnection { pipeline.hincrby(key, field, delta); return null; } - return client.hincrby(key, field, delta).integer; + return client.hincrby(key, field, delta).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1663,7 +1672,7 @@ public class SRedisConnection implements RedisConnection { pipeline.hkeys(key); return null; } - return SRedisUtils.toSet(client.hkeys(key).byteArrays); + return SrpUtils.toSet(client.hkeys(key).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1676,7 +1685,7 @@ public class SRedisConnection implements RedisConnection { pipeline.hlen(key); return null; } - return client.hlen(key).integer; + return client.hlen(key).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1689,7 +1698,7 @@ public class SRedisConnection implements RedisConnection { pipeline.hmget(key, (Object[]) fields); return null; } - return SRedisUtils.toBytesList(client.hmget(key, (Object[]) fields).byteArrays); + return SrpUtils.toBytesList(client.hmget(key, (Object[]) fields).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1699,10 +1708,10 @@ public class SRedisConnection implements RedisConnection { public void hMSet(byte[] key, Map tuple) { try { if (isPipelined()) { - pipeline.hmset(key, tuple); + pipeline.hmset(key, SrpUtils.convert(tuple)); return; } - client.hmset(key, tuple); + client.hmset(key, SrpUtils.convert(tuple)); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1715,7 +1724,7 @@ public class SRedisConnection implements RedisConnection { pipeline.hvals(key); return null; } - return SRedisUtils.toBytesList(client.hvals(key).byteArrays); + return SrpUtils.toBytesList(client.hvals(key).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1735,7 +1744,7 @@ public class SRedisConnection implements RedisConnection { pipeline.publish(channel, message); return null; } - return client.publish(channel, message).integer; + return client.publish(channel, message).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1763,7 +1772,7 @@ public class SRedisConnection implements RedisConnection { throw new UnsupportedOperationException(); } - subscription = new SRedisSubscription(listener, client); + subscription = new SrpSubscription(listener, client); subscription.pSubscribe(patterns); } catch (Exception ex) { throw convertSRAccessException(ex); @@ -1779,7 +1788,7 @@ public class SRedisConnection implements RedisConnection { throw new UnsupportedOperationException(); } - subscription = new SRedisSubscription(listener, client); + subscription = new SrpSubscription(listener, client); subscription.subscribe(channels); } catch (Exception ex) { diff --git a/src/main/java/org/springframework/data/redis/connection/sredis/SRedisConnectionFactory.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnectionFactory.java similarity index 71% rename from src/main/java/org/springframework/data/redis/connection/sredis/SRedisConnectionFactory.java rename to src/main/java/org/springframework/data/redis/connection/srp/SrpConnectionFactory.java index 09f6935c9..edd0d0879 100644 --- a/src/main/java/org/springframework/data/redis/connection/sredis/SRedisConnectionFactory.java +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnectionFactory.java @@ -14,12 +14,10 @@ * limitations under the License. */ -package org.springframework.data.redis.connection.sredis; +package org.springframework.data.redis.connection.srp; -import java.io.IOException; -import java.lang.reflect.Field; -import java.net.Socket; -import java.util.Queue; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -29,67 +27,60 @@ import org.springframework.dao.DataAccessException; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; -import org.springframework.util.ReflectionUtils; - -import redis.client.SocketPool; /** * Connection factory creating Redis Protocol based connections. * * @author Costin Leau */ -public class SRedisConnectionFactory implements InitializingBean, DisposableBean, RedisConnectionFactory { +public class SrpConnectionFactory implements InitializingBean, DisposableBean, RedisConnectionFactory { private final static Log log = LogFactory.getLog(JedisConnectionFactory.class); private String hostName = "localhost"; private int port = 6379; - private SocketPool pool; + private BlockingQueue trackedConnections = new ArrayBlockingQueue(50); + /** * Constructs a new SRedisConnectionFactory instance * with default settings. */ - public SRedisConnectionFactory() { + public SrpConnectionFactory() { } /** * Constructs a new SRedisConnectionFactory instance * with default settings. */ - public SRedisConnectionFactory(String host, int port) { + public SrpConnectionFactory(String host, int port) { this.hostName = host; this.port = port; } public void afterPropertiesSet() { - pool = new SocketPool(hostName, port); } public void destroy() { - Field f = ReflectionUtils.findField(SocketPool.class, "queue"); - ReflectionUtils.makeAccessible(f); - Queue queue = (Queue) ReflectionUtils.getField(f, pool); - Socket s = null; + SrpConnection con; do { - s = queue.poll(); - if (s != null) { + con = trackedConnections.poll(); + if (con != null && !con.isClosed()) { try { - s.close(); - } catch (IOException ex) { + con.close(); + } catch (Exception ex) { // ignore } } - } while (s != null); - pool = null; + } while (con != null); } public RedisConnection getConnection() { - return new SRedisConnection(pool); + return new SrpConnection(hostName, port, trackedConnections); } public DataAccessException translateExceptionIfPossible(RuntimeException ex) { - return SRedisUtils.convertSRedisAccessException(ex); + return SrpUtils.convertSRedisAccessException(ex); } /** diff --git a/src/main/java/org/springframework/data/redis/connection/sredis/SRedisSubscription.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpSubscription.java similarity index 89% rename from src/main/java/org/springframework/data/redis/connection/sredis/SRedisSubscription.java rename to src/main/java/org/springframework/data/redis/connection/srp/SrpSubscription.java index fd7ce5977..2ac193382 100644 --- a/src/main/java/org/springframework/data/redis/connection/sredis/SRedisSubscription.java +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpSubscription.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.data.redis.connection.sredis; +package org.springframework.data.redis.connection.srp; import org.springframework.data.redis.connection.MessageListener; import org.springframework.data.redis.connection.util.AbstractSubscription; @@ -26,11 +26,11 @@ import redis.client.RedisClient; * * @author Costin Leau */ -class SRedisSubscription extends AbstractSubscription { +class SrpSubscription extends AbstractSubscription { private final RedisClient client; - SRedisSubscription(MessageListener listener, RedisClient client) { + SrpSubscription(MessageListener listener, RedisClient client) { super(listener); this.client = client; } diff --git a/src/main/java/org/springframework/data/redis/connection/sredis/SRedisUtils.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpUtils.java similarity index 85% rename from src/main/java/org/springframework/data/redis/connection/sredis/SRedisUtils.java rename to src/main/java/org/springframework/data/redis/connection/srp/SrpUtils.java index 5019a7a89..946f3cefb 100644 --- a/src/main/java/org/springframework/data/redis/connection/sredis/SRedisUtils.java +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpUtils.java @@ -14,12 +14,11 @@ * limitations under the License. */ -package org.springframework.data.redis.connection.sredis; +package org.springframework.data.redis.connection.srp; import java.io.StringReader; import java.util.ArrayList; import java.util.Arrays; -import java.util.Collections; import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.List; @@ -38,6 +37,7 @@ import org.springframework.util.Assert; import redis.client.RedisException; import redis.reply.BulkReply; import redis.reply.MultiBulkReply; +import redis.reply.Reply; import com.google.common.base.Charsets; @@ -46,7 +46,7 @@ import com.google.common.base.Charsets; * * @author Costin Leau */ -abstract class SRedisUtils { +abstract class SrpUtils { private static final byte[] ONE = new byte[] { 1 }; private static final byte[] ZERO = new byte[] { 0 }; @@ -70,7 +70,7 @@ abstract class SRedisUtils { static Properties info(BulkReply reply) { Properties info = new Properties(); // use the same charset as the library - StringReader stringReader = new StringReader(new String(reply.bytes, Charsets.UTF_8)); + StringReader stringReader = new StringReader(new String(reply.data(), Charsets.UTF_8)); try { info.load(stringReader); } catch (Exception ex) { @@ -81,17 +81,17 @@ abstract class SRedisUtils { return info; } - static List toBytesList(Object[] byteArrays) { - List list = new ArrayList(byteArrays.length); - if (byteArrays.length == 1 && byteArrays[0] == null) { - return Collections.emptyList(); - } - - for (Object obj : byteArrays) { - if (obj instanceof byte[]) - list.add((byte[]) obj); + static List toBytesList(Reply[] replies) { + List list = new ArrayList(replies.length); + for (Reply reply : replies) { + Object data = reply.data(); + if (data == null) { + list.add(null); + } + else if (data instanceof byte[]) + list.add((byte[]) data); else - throw new IllegalArgumentException("array contains more then just bytes" + obj); + throw new IllegalArgumentException("array contains more then just nulls and bytes -> " + data); } return list; @@ -101,7 +101,7 @@ abstract class SRedisUtils { return Arrays.asList(byteArrays); } - static Set toSet(Object[] byteArrays) { + static Set toSet(Reply[] byteArrays) { return new LinkedHashSet(toBytesList(byteArrays)); } @@ -126,21 +126,21 @@ abstract class SRedisUtils { } static Double toDouble(byte[] bytes) { - return Double.valueOf(new String(bytes, Charsets.UTF_8)); + return (bytes == null || bytes.length == 0 ? null : Double.valueOf(new String(bytes, Charsets.UTF_8))); } - static Long toLong(Object[] byteArrays) { - return Long.valueOf(new String((byte[]) byteArrays[0], Charsets.UTF_8)); + static Long toLong(Object[] bytes) { + return (bytes == null || bytes.length == 0 ? null : Long.valueOf(new String((byte[]) bytes[0], Charsets.UTF_8))); } static Set convertTuple(MultiBulkReply zrange) { - Object[] byteArrays = zrange.byteArrays; + Reply[] byteArrays = zrange.data(); Set tuples = new LinkedHashSet(byteArrays.length / 2 + 1); for (int i = 0; i < byteArrays.length; i++) { - byte[] value = (byte[]) byteArrays[i]; + byte[] value = (byte[]) byteArrays[i].data(); i++; - Double score = toDouble((byte[]) byteArrays[i]); + Double score = toDouble((byte[]) byteArrays[i].data()); tuples.add(new DefaultTuple(value, score)); } diff --git a/src/main/java/org/springframework/data/redis/connection/sredis/package-info.java b/src/main/java/org/springframework/data/redis/connection/srp/package-info.java similarity index 68% rename from src/main/java/org/springframework/data/redis/connection/sredis/package-info.java rename to src/main/java/org/springframework/data/redis/connection/srp/package-info.java index e151a4393..d3b6d524f 100644 --- a/src/main/java/org/springframework/data/redis/connection/sredis/package-info.java +++ b/src/main/java/org/springframework/data/redis/connection/srp/package-info.java @@ -1,5 +1,5 @@ /** * Connection package for spullara Redis Protocol library. */ -package org.springframework.data.redis.connection.sredis; +package org.springframework.data.redis.connection.srp; 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 216be63be..972c00c8a 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -16,7 +16,12 @@ package org.springframework.data.redis.connection; -import static org.junit.Assert.*; +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; import java.util.Arrays; import java.util.List; @@ -34,12 +39,6 @@ import org.springframework.dao.DataAccessException; import org.springframework.data.redis.Address; import org.springframework.data.redis.ConnectionFactoryTracker; import org.springframework.data.redis.Person; -import org.springframework.data.redis.connection.DefaultStringRedisConnection; -import org.springframework.data.redis.connection.Message; -import org.springframework.data.redis.connection.MessageListener; -import org.springframework.data.redis.connection.RedisConnection; -import org.springframework.data.redis.connection.RedisConnectionFactory; -import org.springframework.data.redis.connection.StringRedisConnection; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer; import org.springframework.data.redis.serializer.RedisSerializer; diff --git a/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionIntegrationTests.java index 0bf3a8414..b6fc07350 100644 --- a/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionIntegrationTests.java @@ -20,7 +20,6 @@ import org.junit.Test; import org.springframework.data.redis.SettingsUtils; import org.springframework.data.redis.connection.AbstractConnectionIntegrationTests; import org.springframework.data.redis.connection.RedisConnectionFactory; -import org.springframework.data.redis.connection.sredis.SRedisConnectionFactory; import redis.client.RedisClient; @@ -28,10 +27,10 @@ import redis.client.RedisClient; * @author Costin Leau */ public class SrpConnectionIntegrationTests extends AbstractConnectionIntegrationTests { - SRedisConnectionFactory factory; + SrpConnectionFactory factory; public SrpConnectionIntegrationTests() { - factory = new SRedisConnectionFactory(); + factory = new SrpConnectionFactory(); factory.setPort(SettingsUtils.getPort()); factory.setHostName(SettingsUtils.getHost()); @@ -52,4 +51,7 @@ public class SrpConnectionIntegrationTests extends AbstractConnectionIntegration rc.set("foobar", "barfoo"); System.out.println(rc.get("foobar")); } + + public void testNullCollections() throws Exception { + } } diff --git a/src/test/java/org/springframework/data/redis/support/collections/CollectionTestParams.java b/src/test/java/org/springframework/data/redis/support/collections/CollectionTestParams.java index c35cc0e55..16839b639 100644 --- a/src/test/java/org/springframework/data/redis/support/collections/CollectionTestParams.java +++ b/src/test/java/org/springframework/data/redis/support/collections/CollectionTestParams.java @@ -23,7 +23,7 @@ import org.springframework.data.redis.SettingsUtils; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; import org.springframework.data.redis.connection.jredis.JredisConnectionFactory; import org.springframework.data.redis.connection.rjc.RjcConnectionFactory; -import org.springframework.data.redis.connection.sredis.SRedisConnectionFactory; +import org.springframework.data.redis.connection.srp.SrpConnectionFactory; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.serializer.JacksonJsonRedisSerializer; @@ -137,7 +137,7 @@ public abstract class CollectionTestParams { jsonPersonTemplateRJC.afterPropertiesSet(); // SRP - SRedisConnectionFactory srConnFactory = new SRedisConnectionFactory(); + SrpConnectionFactory srConnFactory = new SrpConnectionFactory(); srConnFactory.setPort(SettingsUtils.getPort()); srConnFactory.setHostName(SettingsUtils.getHost()); srConnFactory.afterPropertiesSet(); From 3cf59c224155c318a08882fc03143c5699cc6b9b Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Mon, 2 Apr 2012 10:38:21 +0300 Subject: [PATCH 4/9] integrate the latest srp updates add pub/sub listening support now that it's supported by srp --- .../redis/connection/srp/SrpConnection.java | 7 +-- .../connection/srp/SrpMessageListener.java | 54 +++++++++++++++++++ .../redis/connection/srp/SrpSubscription.java | 6 ++- 3 files changed, 62 insertions(+), 5 deletions(-) create mode 100644 src/main/java/org/springframework/data/redis/connection/srp/SrpMessageListener.java 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 bd07a475d..46fa598da 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 @@ -1465,10 +1465,11 @@ public class SrpConnection implements RedisConnection { public Long zRank(byte[] key, byte[] value) { try { if (isPipelined()) { - pipeline.zrank(key, value); - return null; + // (Long) pipeline .zrank(key, value).data; + // return null; + throw new UnsupportedOperationException(); } - return client.zrank(key, value).data(); + return (Long) client.zrank(key, value).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpMessageListener.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpMessageListener.java new file mode 100644 index 000000000..619438adc --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpMessageListener.java @@ -0,0 +1,54 @@ +/* + * Copyright 2011-2012 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.srp; + +import org.springframework.data.redis.connection.DefaultMessage; +import org.springframework.data.redis.connection.MessageListener; +import org.springframework.util.Assert; + +import redis.client.ReplyListener; + +/** + * MessageListener wrapper around SRP {@link ReplyListener}. + * + * @author Costin Leau + */ +class SrpMessageListener implements ReplyListener { + + private final MessageListener listener; + + SrpMessageListener(MessageListener listener) { + Assert.notNull(listener, "message listener is required"); + this.listener = listener; + } + + public void message(byte[] channel, byte[] message) { + listener.onMessage(new DefaultMessage(channel, message), null); + } + + public void psubscribed(byte[] arg0, int arg1) { + } + + public void punsubscribed(byte[] arg0, int arg1) { + } + + public void subscribed(byte[] arg0, int arg1) { + } + + public void unsubscribed(byte[] arg0, int arg1) { + } +} diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpSubscription.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpSubscription.java index 2ac193382..16b44803a 100644 --- a/src/main/java/org/springframework/data/redis/connection/srp/SrpSubscription.java +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpSubscription.java @@ -20,6 +20,7 @@ import org.springframework.data.redis.connection.MessageListener; import org.springframework.data.redis.connection.util.AbstractSubscription; import redis.client.RedisClient; +import redis.client.ReplyListener; /** * Message subscription on top of SRP. @@ -29,10 +30,13 @@ import redis.client.RedisClient; class SrpSubscription extends AbstractSubscription { private final RedisClient client; + private final ReplyListener listener; SrpSubscription(MessageListener listener, RedisClient client) { super(listener); this.client = client; + this.listener = new SrpMessageListener(listener); + client.addListener(this.listener); } protected void doClose() { @@ -45,7 +49,6 @@ class SrpSubscription extends AbstractSubscription { client.psubscribe((Object[]) patterns); } - protected void doPUnsubscribe(boolean all, byte[]... patterns) { if (all) { client.punsubscribe((Object[]) null); @@ -59,7 +62,6 @@ class SrpSubscription extends AbstractSubscription { client.subscribe((Object[]) channels); } - protected void doUnsubscribe(boolean all, byte[]... channels) { if (all) { client.unsubscribe((Object[]) null); From e5b638315ad39fb34007bd90028be28013b7e405 Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Tue, 3 Apr 2012 16:04:38 +0300 Subject: [PATCH 5/9] update to latest SRP version --- .gitignore | 3 ++- .../redis/connection/srp/SrpConnection.java | 26 +++++++++++++------ .../connection/srp/SrpMessageListener.java | 4 +++ .../redis/connection/srp/SrpSubscription.java | 1 + 4 files changed, 25 insertions(+), 9 deletions(-) diff --git a/.gitignore b/.gitignore index 863410265..140bc3589 100644 --- a/.gitignore +++ b/.gitignore @@ -10,4 +10,5 @@ pom.xml *.iws *.log .classpath -.project \ No newline at end of file +.project +.settings 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 46fa598da..31bc92d55 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 @@ -36,6 +36,7 @@ import org.springframework.data.redis.connection.Subscription; import redis.client.RedisClient; import redis.client.RedisClient.Pipeline; import redis.client.RedisException; +import redis.reply.Reply; import com.google.common.base.Charsets; @@ -136,7 +137,7 @@ public class SrpConnection implements RedisConnection { pipeline.sort(key, sort, null, (Object[]) null); return null; } - return SrpUtils.toBytesList(client.sort(key, sort, null, (Object[]) null).data()); + return SrpUtils.toBytesList((Reply[]) client.sort(key, sort, null, (Object[]) null).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -151,7 +152,7 @@ public class SrpConnection implements RedisConnection { pipeline.sort(key, sort, null, (Object[]) null); return null; } - return SrpUtils.toLong(client.sort(key, sort, null, (Object[]) null).data()); + return ((Long) client.sort(key, sort, null, (Object[]) null).data()); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1252,10 +1253,10 @@ public class SrpConnection implements RedisConnection { public Boolean zAdd(byte[] key, double score, byte[] value) { try { if (isPipelined()) { - pipeline.zadd(key, score, value, null, null); + pipeline.zadd(new Object[] { key, score, value }); return null; } - return client.zadd(key, score, value, null, null).data() == 1; + return client.zadd(new Object[] { key, score, value }).data() == 1; } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1307,18 +1308,27 @@ public class SrpConnection implements RedisConnection { public Long zInterStore(byte[] destKey, byte[]... sets) { + + Object[] args = new Object[2+sets.length]; + + args[0] = destKey; + args[1] = sets.length; + int i = 2; + for (byte[] set : sets) { + args[i++] = set; + } + try { if (isQueueing()) { - pipeline.zinterstore(destKey, sets.length, (Object[]) sets); + pipeline.zinterstore(args); return null; } - return client.zinterstore(destKey, sets.length, (Object[]) sets).data(); + return client.zinterstore(args).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } } - public Set zRange(byte[] key, long start, long end) { try { if (isPipelined()) { @@ -1534,7 +1544,7 @@ public class SrpConnection implements RedisConnection { pipeline.zrevrank(key, value); return null; } - return client.zrevrank(key, value).data(); + return (Long) client.zrevrank(key, value).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpMessageListener.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpMessageListener.java index 619438adc..0534ceb81 100644 --- a/src/main/java/org/springframework/data/redis/connection/srp/SrpMessageListener.java +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpMessageListener.java @@ -40,6 +40,10 @@ class SrpMessageListener implements ReplyListener { listener.onMessage(new DefaultMessage(channel, message), null); } + public void pmessage(byte[] pattern, byte[] channel, byte[] message) { + listener.onMessage(new DefaultMessage(channel, message), pattern); + } + public void psubscribed(byte[] arg0, int arg1) { } diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpSubscription.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpSubscription.java index 16b44803a..473122936 100644 --- a/src/main/java/org/springframework/data/redis/connection/srp/SrpSubscription.java +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpSubscription.java @@ -42,6 +42,7 @@ class SrpSubscription extends AbstractSubscription { protected void doClose() { client.unsubscribe((Object[]) null); client.punsubscribe((Object[]) null); + client.removeListener(this.listener); } From cac41148e489d875a97094532341f04984413ccb Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Tue, 3 Apr 2012 22:50:47 +0300 Subject: [PATCH 6/9] update client driver - pretty much working... --- .../redis/connection/srp/SrpConnection.java | 28 ++++++++----------- 1 file changed, 11 insertions(+), 17 deletions(-) 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 31bc92d55..b1aed3cc6 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 @@ -359,9 +359,7 @@ public class SrpConnection implements RedisConnection { isMulti = false; try { if (isPipelined()) { - //pipeline.discard(); - //return; - throw new UnsupportedOperationException(); + client.discard(); } client.discard(); @@ -374,13 +372,9 @@ public class SrpConnection implements RedisConnection { public List exec() { isMulti = false; try { - if (isPipelined()) { - // pipeline.exec(); - // return null; - throw new UnsupportedOperationException(); - } - // return SrpUtils.toList(client.exec()); - throw new UnsupportedOperationException(); + // if (isPipelined()) { + return Collections.singletonList((Object) client.exec()); + // } } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -444,11 +438,11 @@ public class SrpConnection implements RedisConnection { return; } isMulti = true; + openPipeline(); try { if (isPipelined()) { - // pipeline.multi(); - // return; - throw new UnsupportedOperationException(); + client.multi(); + return; } client.multi(); } catch (Exception ex) { @@ -1308,16 +1302,16 @@ public class SrpConnection implements RedisConnection { public Long zInterStore(byte[] destKey, byte[]... sets) { - - Object[] args = new Object[2+sets.length]; - + + Object[] args = new Object[2 + sets.length]; + args[0] = destKey; args[1] = sets.length; int i = 2; for (byte[] set : sets) { args[i++] = set; } - + try { if (isQueueing()) { pipeline.zinterstore(args); From 3f6227bbfa23547c818226031344135425fb17d9 Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Wed, 4 Apr 2012 09:25:58 +0300 Subject: [PATCH 7/9] build against SRP 0.2 version - all tests pass --- build.gradle | 2 +- gradle.properties | 4 ++-- template.mf | 5 ++++- 3 files changed, 7 insertions(+), 4 deletions(-) diff --git a/build.gradle b/build.gradle index 066c12d59..6f137f0a7 100644 --- a/build.gradle +++ b/build.gradle @@ -82,7 +82,7 @@ dependencies { // Redis Drivers compile "redis.clients:jedis:$jedisVersion" - compile "com.github.spullara.redis:client:$sredisVersion" + compile "com.github.spullara.redis:client:$srpVersion" compile("org.jredis:jredis-anthonylauzon:$jredisVersion") { optional = true } compile("org.idevlab:rjc:$rjcVersion") { optional = true } diff --git a/gradle.properties b/gradle.properties index 07999da81..700fbc276 100644 --- a/gradle.properties +++ b/gradle.properties @@ -16,7 +16,7 @@ mockitoVersion = 1.8.5 jedisVersion = 2.0.0 jredisVersion = 03122010 rjcVersion = 0.6.4 -sredisVersion = 0.2-SNAPSHOT +srpVersion = 0.2 # Manifest properties @@ -25,7 +25,7 @@ spring.range = "[3.1.0, 4.0.0)" jedis.range = "[2.0.0, 2.0.0]" jackson.range = "[1.6, 2.0.0)" rjc.range = "[0.6.4, 0.6.4]" -sredis.range = "[0.1, 1.0)" +srp.range = "[0.2, 1.0)" # -------------------- # Project wide version diff --git a/template.mf b/template.mf index ed33cbcaf..182e1c84c 100644 --- a/template.mf +++ b/template.mf @@ -26,4 +26,7 @@ Import-Template: org.idevlab.rjc.*;resolution:="optional";version=${rjc.range}, org.apache.commons.pool.impl.*;resolution:="optional";version="[1.0.0, 3.0.0)", org.codehaus.jackson.*;resolution:="optional";version=${jackson.range}, - org.apache.commons.beanutils.*;resolution:="optional";version=1.8.5 \ No newline at end of file + org.apache.commons.beanutils.*;resolution:="optional";version=1.8.5, + redis.client.*;resolution:="optional";version=${srp.range}, + redis.reply.*;resolution:="optional";version=${srp.range}, + com.google.common.*;resolution:="optional";version="[11.0.0, 20.0.0)" \ No newline at end of file From 79d9582c4e961eb18007e1007528ce7a3dc4091c Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Wed, 23 May 2012 18:04:06 +0300 Subject: [PATCH 8/9] fix connection.setBit() DATAREDIS-89 --- .classpath | 4 ++-- .settings/org.eclipse.jdt.core.prefs | 2 +- .../springframework/data/redis/connection/srp/SrpUtils.java | 4 ++-- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/.classpath b/.classpath index a316b6684..06c221d7f 100644 --- a/.classpath +++ b/.classpath @@ -19,9 +19,8 @@ - + - @@ -34,6 +33,7 @@ + diff --git a/.settings/org.eclipse.jdt.core.prefs b/.settings/org.eclipse.jdt.core.prefs index 9377866ca..8c9e1f31e 100644 --- a/.settings/org.eclipse.jdt.core.prefs +++ b/.settings/org.eclipse.jdt.core.prefs @@ -1,5 +1,5 @@ # -#Sun Apr 01 10:03:27 EEST 2012 +#Wed Apr 04 09:18:47 EEST 2012 org.eclipse.jdt.core.compiler.debug.localVariable=generate org.eclipse.jdt.core.compiler.compliance=1.5 org.eclipse.jdt.core.compiler.codegen.unusedLocal=preserve diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpUtils.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpUtils.java index 946f3cefb..808f33812 100644 --- a/src/main/java/org/springframework/data/redis/connection/srp/SrpUtils.java +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpUtils.java @@ -48,8 +48,8 @@ import com.google.common.base.Charsets; */ abstract class SrpUtils { - private static final byte[] ONE = new byte[] { 1 }; - private static final byte[] ZERO = new byte[] { 0 }; + private static final byte[] ONE = new byte[] { '1' }; + private static final byte[] ZERO = new byte[] { '0' }; private static final byte[] BEFORE = "BEFORE".getBytes(Charsets.UTF_8); private static final byte[] AFTER = "AFTER".getBytes(Charsets.UTF_8); static final byte[] WITHSCORES = "WITHSCORES".getBytes(Charsets.UTF_8); From d9d954642cc1b8923c426adc31ae72cda8448867 Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Thu, 21 Jun 2012 20:40:36 +0300 Subject: [PATCH 9/9] add native execution for Srp connection --- .../redis/connection/srp/SrpConnection.java | 32 +++++++++++++------ 1 file changed, 23 insertions(+), 9 deletions(-) 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 b1aed3cc6..ae56e207b 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 @@ -32,7 +32,9 @@ import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisSubscribedConnectionException; import org.springframework.data.redis.connection.SortParameters; import org.springframework.data.redis.connection.Subscription; +import org.springframework.util.Assert; +import redis.Command; import redis.client.RedisClient; import redis.client.RedisClient.Pipeline; import redis.client.RedisException; @@ -75,6 +77,18 @@ public class SrpConnection implements RedisConnection { return new RedisSystemException("Unknown SRedis exception", ex); } + public Object execute(String command, byte[]... args) { + Assert.hasText(command, "a valid command needs to be specified"); + String name = command.trim().toUpperCase(); + Command cmd = new Command(name.getBytes(Charsets.UTF_8), args); + if (isPipelined()) { + client.pipeline(name, cmd); + return null; + } + else { + return client.execute(name, cmd); + } + } public void close() throws DataAccessException { isClosed = true; @@ -1097,13 +1111,13 @@ public class SrpConnection implements RedisConnection { } - public void sDiffStore(byte[] destKey, byte[]... keys) { + public Long sDiffStore(byte[] destKey, byte[]... keys) { try { if (isPipelined()) { pipeline.sdiffstore(destKey, (Object[]) keys); - return; + return null; } - client.sdiffstore(destKey, (Object[]) keys); + return client.sdiffstore(destKey, (Object[]) keys).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1123,13 +1137,13 @@ public class SrpConnection implements RedisConnection { } - public void sInterStore(byte[] destKey, byte[]... keys) { + public Long sInterStore(byte[] destKey, byte[]... keys) { try { if (isPipelined()) { pipeline.sinterstore(destKey, (Object[]) keys); - return; + return null; } - client.sinterstore(destKey, (Object[]) keys); + return client.sinterstore(destKey, (Object[]) keys).data(); } catch (Exception ex) { throw convertSRAccessException(ex); } @@ -1227,13 +1241,13 @@ public class SrpConnection implements RedisConnection { } - public void sUnionStore(byte[] destKey, byte[]... keys) { + public Long sUnionStore(byte[] destKey, byte[]... keys) { try { if (isPipelined()) { pipeline.sunionstore(destKey, (Object[]) keys); - return; + return null; } - client.sunionstore(destKey, (Object[]) keys); + return client.sunionstore(destKey, (Object[]) keys).data(); } catch (Exception ex) { throw convertSRAccessException(ex); }