From 850560f2237f5d8964479ba2ba16e25c7b752c49 Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Wed, 16 Mar 2011 09:49:07 +0200 Subject: [PATCH 1/4] DATAKV-46 + add initial Rjc connection/connection factory support --- spring-data-redis/pom.xml | 23 +- .../jedis/JedisConnectionFactory.java | 2 +- .../connection/jredis/JredisConnection.java | 6 +- .../redis/connection/rjc/RjcConnection.java | 709 ++++++++++++++++++ .../connection/rjc/RjcConnectionFactory.java | 209 ++++++ .../redis/connection/rjc/RjcUtils.java | 41 + .../connection/rjc/SingleDataSource.java | 38 + 7 files changed, 1011 insertions(+), 17 deletions(-) create mode 100644 spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java create mode 100644 spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnectionFactory.java create mode 100644 spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcUtils.java create mode 100644 spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/SingleDataSource.java diff --git a/spring-data-redis/pom.xml b/spring-data-redis/pom.xml index 2f683aefe..c0fe69d97 100644 --- a/spring-data-redis/pom.xml +++ b/spring-data-redis/pom.xml @@ -16,8 +16,10 @@ "[3.0.0, 4.0.0)" 03122010 1.5.2 + 0.6.2 "[1.0.0,2.0.0)" "[1.6, 2.0.0)" + "[0.6.2, 0.6.2]" @@ -135,27 +137,18 @@ compile - org.jredis jredis-anthonylauzon ${jredis.ver} compile + + org.idevlab + rjc + ${rjc.ver} + compile + diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnectionFactory.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnectionFactory.java index 099067bbd..51f326a39 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnectionFactory.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnectionFactory.java @@ -34,7 +34,7 @@ import redis.clients.jedis.JedisShardInfo; import redis.clients.jedis.Protocol; /** - * Connection factory using creating Jedis based connections. + * Connection factory creating Jedis based connections. * * @author Costin Leau */ diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisConnection.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisConnection.java index dca7e829d..6357ff522 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisConnection.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisConnection.java @@ -81,7 +81,11 @@ public class JredisConnection implements RedisConnection { // don't actually close the connection // if a pool is used if (!isPool) { - jredis.quit(); + try { + jredis.quit(); + } catch (Exception ex) { + throw convertJredisAccessException(ex); + } } } diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java new file mode 100644 index 000000000..1983555d5 --- /dev/null +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java @@ -0,0 +1,709 @@ +/* + * 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.keyvalue.redis.connection.rjc; + +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Set; + +import org.idevlab.rjc.RedisException; +import org.idevlab.rjc.Session; +import org.idevlab.rjc.SessionFactoryImpl; +import org.springframework.dao.DataAccessException; +import org.springframework.data.keyvalue.UncategorizedKeyvalueStoreException; +import org.springframework.data.keyvalue.redis.connection.DataType; +import org.springframework.data.keyvalue.redis.connection.MessageListener; +import org.springframework.data.keyvalue.redis.connection.RedisConnection; +import org.springframework.data.keyvalue.redis.connection.SortParameters; +import org.springframework.data.keyvalue.redis.connection.Subscription; + +/** + * {@code RedisConnection} implementation on top of rjc library. + * + * @author Costin Leau + */ +public class RjcConnection implements RedisConnection { + + private final int dbIndex; + private final Session session; + private boolean isClosed = false; + + public RjcConnection(org.idevlab.rjc.ds.RedisConnection connection, int dbIndex) { + session = new SessionFactoryImpl(new SingleDataSource(connection)).create(); + this.dbIndex = dbIndex; + + // select the db + if (dbIndex > 0) { + select(dbIndex); + } + } + + protected DataAccessException convertRjcAccessException(Exception ex) { + if (ex instanceof RedisException) { + return RjcUtils.convertRjcAccessException((RedisException) ex); + } + return new UncategorizedKeyvalueStoreException("Unknown rjc exception", ex); + } + + @Override + public void close() throws DataAccessException { + isClosed = true; + try { + session.close(); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + + @Override + public boolean isClosed() { + return isClosed; + } + + @Override + public Session getNativeConnection() { + return session; + } + + @Override + public List closePipeline() { + throw new UnsupportedOperationException(); + } + + + @Override + public boolean isPipelined() { + throw new UnsupportedOperationException(); + } + + @Override + public boolean isQueueing() { + throw new UnsupportedOperationException(); + } + + @Override + public void openPipeline() { + throw new UnsupportedOperationException(); + } + + @Override + public Long del(byte[]... keys) { + throw new UnsupportedOperationException(); + } + + @Override + public byte[] echo(byte[] message) { + throw new UnsupportedOperationException(); + } + + @Override + public Boolean exists(byte[] key) { + throw new UnsupportedOperationException(); + } + + @Override + public Boolean expire(byte[] key, long seconds) { + throw new UnsupportedOperationException(); + } + + @Override + public Boolean expireAt(byte[] key, long unixTime) { + throw new UnsupportedOperationException(); + } + + @Override + public Set keys(byte[] pattern) { + throw new UnsupportedOperationException(); + } + + @Override + public Boolean persist(byte[] key) { + throw new UnsupportedOperationException(); + } + + @Override + public String ping() { + throw new UnsupportedOperationException(); + } + + @Override + public byte[] randomKey() { + throw new UnsupportedOperationException(); + } + + @Override + public void rename(byte[] oldName, byte[] newName) { + throw new UnsupportedOperationException(); + } + + @Override + public Boolean renameNX(byte[] oldName, byte[] newName) { + throw new UnsupportedOperationException(); + } + + @Override + public void select(int dbIndex) { + throw new UnsupportedOperationException(); + } + + @Override + public List sort(byte[] key, SortParameters params) { + throw new UnsupportedOperationException(); + } + + @Override + public Long sort(byte[] key, SortParameters params, byte[] storeKey) { + throw new UnsupportedOperationException(); + } + + @Override + public Long ttl(byte[] key) { + throw new UnsupportedOperationException(); + } + + @Override + public DataType type(byte[] key) { + throw new UnsupportedOperationException(); + } + + @Override + public void discard() { + throw new UnsupportedOperationException(); + } + + @Override + public List exec() { + throw new UnsupportedOperationException(); + } + + @Override + public void multi() { + throw new UnsupportedOperationException(); + } + + @Override + public void unwatch() { + throw new UnsupportedOperationException(); + } + + @Override + public void watch(byte[]... keys) { + throw new UnsupportedOperationException(); + } + + @Override + public Long append(byte[] key, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public Long decr(byte[] key) { + throw new UnsupportedOperationException(); + } + + @Override + public Long decrBy(byte[] key, long value) { + throw new UnsupportedOperationException(); + } + + @Override + public byte[] get(byte[] key) { + throw new UnsupportedOperationException(); + } + + @Override + public Boolean getBit(byte[] key, long offset) { + throw new UnsupportedOperationException(); + } + + @Override + public byte[] getRange(byte[] key, int begin, int end) { + throw new UnsupportedOperationException(); + } + + @Override + public byte[] getSet(byte[] key, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public Long incr(byte[] key) { + throw new UnsupportedOperationException(); + } + + @Override + public Long incrBy(byte[] key, long value) { + throw new UnsupportedOperationException(); + } + + @Override + public List mGet(byte[]... keys) { + throw new UnsupportedOperationException(); + } + + @Override + public void mSet(Map tuple) { + throw new UnsupportedOperationException(); + } + + @Override + public void mSetNX(Map tuple) { + throw new UnsupportedOperationException(); + } + + @Override + public void set(byte[] key, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public void setBit(byte[] key, long offset, boolean value) { + throw new UnsupportedOperationException(); + } + + @Override + public void setEx(byte[] key, long seconds, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public Boolean setNX(byte[] key, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public void setRange(byte[] key, int begin, int end) { + throw new UnsupportedOperationException(); + } + + @Override + public Long strLen(byte[] key) { + throw new UnsupportedOperationException(); + } + + @Override + public List bLPop(int timeout, byte[]... keys) { + throw new UnsupportedOperationException(); + } + + @Override + public List bRPop(int timeout, byte[]... keys) { + throw new UnsupportedOperationException(); + } + + @Override + public byte[] bRPopLPush(int timeout, byte[] srcKey, byte[] dstKey) { + throw new UnsupportedOperationException(); + } + + @Override + public byte[] lIndex(byte[] key, long index) { + throw new UnsupportedOperationException(); + } + + @Override + public Long lInsert(byte[] key, Position where, byte[] pivot, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public Long lLen(byte[] key) { + throw new UnsupportedOperationException(); + } + + @Override + public byte[] lPop(byte[] key) { + throw new UnsupportedOperationException(); + } + + @Override + public Long lPush(byte[] key, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public Long lPushX(byte[] key, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public List lRange(byte[] key, long begin, long end) { + throw new UnsupportedOperationException(); + } + + @Override + public Long lRem(byte[] key, long count, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public void lSet(byte[] key, long index, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public void lTrim(byte[] key, long begin, long end) { + throw new UnsupportedOperationException(); + } + + @Override + public byte[] rPop(byte[] key) { + throw new UnsupportedOperationException(); + } + + @Override + public byte[] rPopLPush(byte[] srcKey, byte[] dstKey) { + throw new UnsupportedOperationException(); + } + + @Override + public Long rPush(byte[] key, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public Long rPushX(byte[] key, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public Boolean sAdd(byte[] key, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public Long sCard(byte[] key) { + throw new UnsupportedOperationException(); + } + + @Override + public Set sDiff(byte[]... keys) { + throw new UnsupportedOperationException(); + } + + @Override + public void sDiffStore(byte[] destKey, byte[]... keys) { + throw new UnsupportedOperationException(); + } + + @Override + public Set sInter(byte[]... keys) { + throw new UnsupportedOperationException(); + } + + @Override + public void sInterStore(byte[] destKey, byte[]... keys) { + throw new UnsupportedOperationException(); + } + + @Override + public Boolean sIsMember(byte[] key, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public Set sMembers(byte[] key) { + throw new UnsupportedOperationException(); + } + + @Override + public Boolean sMove(byte[] srcKey, byte[] destKey, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public byte[] sPop(byte[] key) { + throw new UnsupportedOperationException(); + } + + @Override + public byte[] sRandMember(byte[] key) { + throw new UnsupportedOperationException(); + } + + @Override + public Boolean sRem(byte[] key, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public Set sUnion(byte[]... keys) { + throw new UnsupportedOperationException(); + } + + @Override + public void sUnionStore(byte[] destKey, byte[]... keys) { + throw new UnsupportedOperationException(); + } + + @Override + public Boolean zAdd(byte[] key, double score, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public Long zCard(byte[] key) { + throw new UnsupportedOperationException(); + } + + @Override + public Long zCount(byte[] key, double min, double max) { + throw new UnsupportedOperationException(); + } + + @Override + public Double zIncrBy(byte[] key, double increment, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public Long zInterStore(byte[] destKey, byte[]... sets) { + throw new UnsupportedOperationException(); + } + + @Override + public Long zInterStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets) { + throw new UnsupportedOperationException(); + } + + @Override + public Set zRange(byte[] key, long begin, long end) { + throw new UnsupportedOperationException(); + } + + @Override + public Set zRangeByScore(byte[] key, double min, double max) { + throw new UnsupportedOperationException(); + } + + @Override + public Set zRangeByScore(byte[] key, double min, double max, long offset, long count) { + throw new UnsupportedOperationException(); + } + + @Override + public Set zRangeByScoreWithScore(byte[] key, double min, double max) { + throw new UnsupportedOperationException(); + } + + @Override + public Set zRangeByScoreWithScore(byte[] key, double min, double max, long offset, long count) { + throw new UnsupportedOperationException(); + } + + @Override + public Set zRangeWithScore(byte[] key, long begin, long end) { + throw new UnsupportedOperationException(); + } + + @Override + public Long zRank(byte[] key, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public Boolean zRem(byte[] key, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public Long zRemRange(byte[] key, long begin, long end) { + throw new UnsupportedOperationException(); + } + + @Override + public Long zRemRangeByScore(byte[] key, double min, double max) { + throw new UnsupportedOperationException(); + } + + @Override + public Set zRevRange(byte[] key, long begin, long end) { + throw new UnsupportedOperationException(); + } + + @Override + public Set zRevRangeWithScore(byte[] key, long begin, long end) { + throw new UnsupportedOperationException(); + } + + @Override + public Long zRevRank(byte[] key, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public Double zScore(byte[] key, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public Long zUnionStore(byte[] destKey, byte[]... sets) { + throw new UnsupportedOperationException(); + } + + @Override + public Long zUnionStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets) { + throw new UnsupportedOperationException(); + } + + @Override + public Boolean hDel(byte[] key, byte[] field) { + throw new UnsupportedOperationException(); + } + + @Override + public Boolean hExists(byte[] key, byte[] field) { + throw new UnsupportedOperationException(); + } + + @Override + public byte[] hGet(byte[] key, byte[] field) { + throw new UnsupportedOperationException(); + } + + @Override + public Map hGetAll(byte[] key) { + throw new UnsupportedOperationException(); + } + + @Override + public Long hIncrBy(byte[] key, byte[] field, long delta) { + throw new UnsupportedOperationException(); + } + + @Override + public Set hKeys(byte[] key) { + throw new UnsupportedOperationException(); + } + + @Override + public Long hLen(byte[] key) { + throw new UnsupportedOperationException(); + } + + @Override + public List hMGet(byte[] key, byte[]... fields) { + throw new UnsupportedOperationException(); + } + + @Override + public void hMSet(byte[] key, Map hashes) { + throw new UnsupportedOperationException(); + } + + @Override + public Boolean hSet(byte[] key, byte[] field, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public Boolean hSetNX(byte[] key, byte[] field, byte[] value) { + throw new UnsupportedOperationException(); + } + + @Override + public List hVals(byte[] key) { + throw new UnsupportedOperationException(); + } + + @Override + public void bgSave() { + throw new UnsupportedOperationException(); + } + + @Override + public void bgWriteAof() { + throw new UnsupportedOperationException(); + } + + @Override + public Long dbSize() { + throw new UnsupportedOperationException(); + } + + @Override + public void flushAll() { + throw new UnsupportedOperationException(); + } + + @Override + public void flushDb() { + throw new UnsupportedOperationException(); + } + + @Override + public List getConfig(String pattern) { + throw new UnsupportedOperationException(); + } + + @Override + public Properties info() { + throw new UnsupportedOperationException(); + } + + @Override + public Long lastSave() { + throw new UnsupportedOperationException(); + } + + @Override + public void resetConfigStats() { + throw new UnsupportedOperationException(); + } + + @Override + public void save() { + throw new UnsupportedOperationException(); + } + + @Override + public void setConfig(String param, String value) { + throw new UnsupportedOperationException(); + } + + @Override + public void shutdown() { + throw new UnsupportedOperationException(); + } + + @Override + public Subscription getSubscription() { + throw new UnsupportedOperationException(); + } + + @Override + public boolean isSubscribed() { + throw new UnsupportedOperationException(); + } + + @Override + public void pSubscribe(MessageListener listener, byte[]... patterns) { + throw new UnsupportedOperationException(); + } + + @Override + public Long publish(byte[] channel, byte[] message) { + throw new UnsupportedOperationException(); + } + + @Override + public void subscribe(MessageListener listener, byte[]... channels) { + throw new UnsupportedOperationException(); + } + +} diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnectionFactory.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnectionFactory.java new file mode 100644 index 000000000..97f1c65cd --- /dev/null +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnectionFactory.java @@ -0,0 +1,209 @@ +/* + * 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.keyvalue.redis.connection.rjc; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.idevlab.rjc.ds.DataSource; +import org.idevlab.rjc.ds.PoolableDataSource; +import org.idevlab.rjc.ds.SimpleDataSource; +import org.idevlab.rjc.protocol.Protocol; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.dao.DataAccessException; +import org.springframework.data.keyvalue.redis.connection.RedisConnection; +import org.springframework.data.keyvalue.redis.connection.RedisConnectionFactory; +import org.springframework.data.keyvalue.redis.connection.jedis.JedisConnectionFactory; +import org.springframework.util.Assert; + +/** + * Connection factory creating rjc based connections. + * + * @author Costin Leau + */ +public class RjcConnectionFactory implements InitializingBean, DisposableBean, RedisConnectionFactory { + + private final static Log log = LogFactory.getLog(JedisConnectionFactory.class); + + private String hostName = "localhost"; + private int port = Protocol.DEFAULT_PORT; + private int timeout = Protocol.DEFAULT_TIMEOUT; + private String password; + + private boolean usePool = true; + private int dbIndex = 0; + private DataSource dataSource; + + + /** + * Constructs a new RjcConnectionFactory instance + * with default settings (default connection pooling, no shard information). + */ + public RjcConnectionFactory() { + } + + + public void afterPropertiesSet() { + if (usePool) { + PoolableDataSource pool = new PoolableDataSource(); + pool.setHost(hostName); + pool.setPort(port); + pool.setPassword(password); + pool.setTimeout(timeout); + + pool.init(); + + dataSource = pool; + + } + else { + dataSource = new SimpleDataSource(hostName, port, timeout, password); + } + } + + public void destroy() { + if (usePool && dataSource != null) { + try { + ((PoolableDataSource) dataSource).close(); + } catch (Exception ex) { + log.warn("Cannot properly close Rjc pool", ex); + } + dataSource = null; + } + } + + @Override + public RedisConnection getConnection() { + return postProcessConnection(new RjcConnection(dataSource.getConnection(), usePool, dbIndex)); + } + + /** + * Post process a newly retrieved connection. Useful for decorating or executing + * initialization commands on a new connection. + * This implementation simply returns the connection. + * + * @param connection + * @return processed connection + */ + protected RjcConnection postProcessConnection(RjcConnection connection) { + return connection; + } + + @Override + public DataAccessException translateExceptionIfPossible(RuntimeException ex) { + return RjcUtils.convertRjcAccessException(ex); + } + + + /** + * Returns the Redis hostName. + * + * @return Returns the hostName + */ + public String getHostName() { + return hostName; + } + + /** + * Sets the Redis hostName. + * + * @param hostName The hostName to set. + */ + public void setHostName(String hostName) { + this.hostName = hostName; + } + + /** + * Returns the password used for authenticating with the Redis server. + * + * @return password for authentication + */ + public String getPassword() { + return password; + } + + /** + * Sets the password used for authenticating with the Redis server. + * + * @param password the password to set + */ + public void setPassword(String password) { + this.password = password; + } + + /** + * Returns the port used to connect to the Redis instance. + * + * @return Redis port. + */ + public int getPort() { + return port; + + } + + /** + * Sets the port used to connect to the Redis instance. + * + * @param port Redis port + */ + public void setPort(int port) { + this.port = port; + } + /** + * Returns the timeout. + * + * @return Returns the timeout + */ + public int getTimeout() { + return timeout; + } + + /** + * @param timeout The timeout to set. + */ + public void setTimeout(int timeout) { + this.timeout = timeout; + } + + /** + * Indicates the use of a connection pool. + * + * @return Returns the use of connection pooling. + */ + public boolean getUsePool() { + return usePool; + } + + /** + * Turns on or off the use of connection pooling. + * + * @param usePool The usePool to set. + */ + public void setUsePool(boolean usePool) { + this.usePool = usePool; + } + + /** + * Sets the index of the database used by this connection factory. + * Can be between 0 (default) and 15. + * + * @param index database index + */ + public void setDatabase(int index) { + Assert.isTrue(index >= 0, "invalid DB index (a positive index required)"); + this.dbIndex = index; + } +} \ No newline at end of file diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcUtils.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcUtils.java new file mode 100644 index 000000000..9c0370bb0 --- /dev/null +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcUtils.java @@ -0,0 +1,41 @@ +/* + * 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.keyvalue.redis.connection.rjc; + +import org.idevlab.rjc.RedisException; +import org.springframework.dao.DataAccessException; +import org.springframework.dao.InvalidDataAccessApiUsageException; +import org.springframework.data.keyvalue.redis.UncategorizedRedisException; + +/** + * Helper class featuring methods for RJC connection handling, providing support for exception translation. + * + * @author Costin Leau + */ +public abstract class RjcUtils { + + public static DataAccessException convertRjcAccessException(RuntimeException ex) { + if (ex instanceof RedisException) { + return convertRjcAccessException((RedisException) ex); + } + + return new UncategorizedRedisException("Unknown exception", ex); + } + + public static DataAccessException convertRjcAccessException(RedisException ex) { + return new InvalidDataAccessApiUsageException(ex.getMessage(), ex); + } +} diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/SingleDataSource.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/SingleDataSource.java new file mode 100644 index 000000000..db152b72c --- /dev/null +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/SingleDataSource.java @@ -0,0 +1,38 @@ +/* + * 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.keyvalue.redis.connection.rjc; + +import org.idevlab.rjc.ds.DataSource; +import org.idevlab.rjc.ds.RedisConnection; + +/** + * Basic data source that always returns the same connection. + * + * @author Costin Leau + */ +class SingleDataSource implements DataSource { + + private final RedisConnection connection; + + SingleDataSource(RedisConnection connection) { + this.connection = connection; + } + + @Override + public RedisConnection getConnection() { + return connection; + } +} From 897122263a2f02dcd9052bf6ddeb7ff2136b11dc Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Wed, 16 Mar 2011 13:13:27 +0200 Subject: [PATCH 2/4] DATAKV-44 + almost done with the Rjc connection + arranged base64 a bit + fixed some jredis/jedis bug in the process + updated some of the RedisConnection methods --- .../DefaultStringRedisConnection.java | 6 +- .../redis/connection/RedisStringCommands.java | 4 +- .../connection/jedis/JedisConnection.java | 9 +- .../redis/connection/jedis/JedisUtils.java | 9 +- .../connection/jredis/JredisConnection.java | 8 +- .../redis/connection/jredis/JredisUtils.java | 43 +- .../redis/connection/rjc/RjcConnection.java | 2468 +++++++++++++---- .../connection/rjc/RjcConnectionFactory.java | 2 +- .../redis/connection/rjc/RjcUtils.java | 180 +- .../connection/{jredis => util}/Base64.java | 2 +- .../redis/connection/util/DecodeUtils.java | 82 + 11 files changed, 2196 insertions(+), 617 deletions(-) rename spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/{jredis => util}/Base64.java (99%) create mode 100644 spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/util/DecodeUtils.java diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/DefaultStringRedisConnection.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/DefaultStringRedisConnection.java index 3d857e3df..9db1d2af4 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/DefaultStringRedisConnection.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/DefaultStringRedisConnection.java @@ -156,7 +156,7 @@ public class DefaultStringRedisConnection implements StringRedisConnection { return delegate.getNativeConnection(); } - public byte[] getRange(byte[] key, int start, int end) { + public byte[] getRange(byte[] key, long start, long end) { return delegate.getRange(key, start, end); } @@ -396,8 +396,8 @@ public class DefaultStringRedisConnection implements StringRedisConnection { return delegate.setNX(key, value); } - public void setRange(byte[] key, int start, int end) { - delegate.setRange(key, start, end); + public void setRange(byte[] key, long start, byte[] value) { + delegate.setRange(key, start, value); } public void shutdown() { diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/RedisStringCommands.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/RedisStringCommands.java index ea96dfde6..d68acd0d6 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/RedisStringCommands.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/RedisStringCommands.java @@ -52,9 +52,9 @@ public interface RedisStringCommands { Long append(byte[] key, byte[] value); - byte[] getRange(byte[] key, int begin, int end); + byte[] getRange(byte[] key, long begin, long end); - void setRange(byte[] key, int begin, int end); + void setRange(byte[] key, long offset, byte[] value); Boolean getBit(byte[] key, long offset); diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnection.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnection.java index d3f898769..5fdd3beb2 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnection.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnection.java @@ -914,7 +914,7 @@ public class JedisConnection implements RedisConnection { } @Override - public byte[] getRange(byte[] key, int start, int end) { + public byte[] getRange(byte[] key, long start, long end) { try { if (isQueueing()) { transaction.substr(key, (int) start, (int) end); @@ -1033,7 +1033,7 @@ public class JedisConnection implements RedisConnection { } @Override - public void setRange(byte[] key, int start, int end) { + public void setRange(byte[] key, long start, byte[] value) { throw new UnsupportedOperationException(); } @@ -1770,11 +1770,10 @@ public class JedisConnection implements RedisConnection { public Set zRevRangeWithScore(byte[] key, long start, long end) { try { if (isQueueing()) { - transaction.zrangeWithScores(key, (int) start, (int) end); - return null; + throw new UnsupportedOperationException(); } if (isPipelined()) { - pipeline.zrangeWithScores(key, (int) start, (int) end); + pipeline.zrangeByScoreWithScores(key, (int) start, (int) end); return null; } return JedisUtils.convertJedisTuple(jedis.zrangeByScoreWithScores(key, (int) start, (int) end)); diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisUtils.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisUtils.java index bdfe315cd..b76e1d08a 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisUtils.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisUtils.java @@ -55,8 +55,8 @@ public abstract class JedisUtils { private static final String OK_CODE = "OK"; private static final String OK_MULTI_CODE = "+OK"; - private static final byte[] ONE = new byte[] { 0 }; - private static final byte[] ZERO = new byte[] { 1 }; + private static final byte[] ONE = new byte[] { 1 }; + private static final byte[] ZERO = new byte[] { 0 }; /** * Converts the given, native Jedis exception to Spring's DAO hierarchy. @@ -194,10 +194,13 @@ public abstract class JedisUtils { static Properties info(String string) { Properties info = new Properties(); + StringReader stringReader = new StringReader(string); try { - info.load(new StringReader(string)); + info.load(stringReader); } catch (Exception ex) { throw new UncategorizedRedisException("Cannot read Redis info", ex); + } finally { + stringReader.close(); } return info; } diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisConnection.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisConnection.java index 6357ff522..63072f3ff 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisConnection.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisConnection.java @@ -305,7 +305,7 @@ public class JredisConnection implements RedisConnection { @Override public Set keys(byte[] pattern) { try { - return JredisUtils.convertCollection(jredis.keys(JredisUtils.decode(pattern))); + return JredisUtils.convertToSet(jredis.keys(JredisUtils.decode(pattern))); } catch (Exception ex) { throw convertJredisAccessException(ex); } @@ -463,7 +463,7 @@ public class JredisConnection implements RedisConnection { } @Override - public byte[] getRange(byte[] key, int start, int end) { + public byte[] getRange(byte[] key, long start, long end) { try { return jredis.substr(JredisUtils.decode(key), start, end); } catch (Exception ex) { @@ -518,7 +518,7 @@ public class JredisConnection implements RedisConnection { } @Override - public void setRange(byte[] key, int start, int end) { + public void setRange(byte[] key, long start, byte[] value) { throw new UnsupportedOperationException(); } @@ -1032,7 +1032,7 @@ public class JredisConnection implements RedisConnection { @Override public Set hKeys(byte[] key) { try { - return new LinkedHashSet(JredisUtils.convertCollection(jredis.hkeys(JredisUtils.decode(key)))); + return new LinkedHashSet(JredisUtils.convertToSet(jredis.hkeys(JredisUtils.decode(key)))); } catch (Exception ex) { throw convertJredisAccessException(ex); } diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisUtils.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisUtils.java index 7820186db..9cb3dc146 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisUtils.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisUtils.java @@ -17,8 +17,6 @@ package org.springframework.data.keyvalue.redis.connection.jredis; import java.util.Collection; -import java.util.LinkedHashMap; -import java.util.LinkedHashSet; import java.util.Map; import java.util.Properties; import java.util.Set; @@ -34,6 +32,7 @@ import org.springframework.data.keyvalue.redis.connection.DataType; import org.springframework.data.keyvalue.redis.connection.SortParameters; import org.springframework.data.keyvalue.redis.connection.SortParameters.Order; import org.springframework.data.keyvalue.redis.connection.SortParameters.Range; +import org.springframework.data.keyvalue.redis.connection.util.DecodeUtils; /** * Helper class featuring methods for JRedis connection handling, providing support for exception translation. @@ -82,46 +81,28 @@ public abstract class JredisUtils { } static String decode(byte[] bytes) { - return Base64.encodeToString(bytes, false); - } - - static String[] decodeMultiple(byte[]... bytes) { - String[] result = new String[bytes.length]; - for (int i = 0; i < bytes.length; i++) { - result[i] = decode(bytes[i]); - } - return result; + return DecodeUtils.decode(bytes); } static byte[] encode(String string) { - return Base64.decode(string); + return DecodeUtils.encode(string); + } + + static String[] decodeMultiple(byte[]... bytes) { + return DecodeUtils.decodeMultiple(bytes); } static Map encodeMap(Map map) { - Map result = new LinkedHashMap(map.size()); - for (Map.Entry entry : map.entrySet()) { - result.put(encode(entry.getKey()), entry.getValue()); - } - return result; - } - - static Set convertCollection(Collection keys) { - Set set = new LinkedHashSet(keys.size()); - - for (String string : keys) { - set.add(Base64.decode(string)); - } - return set; + return DecodeUtils.encodeMap(map); } static Map decodeMap(Map tuple) { - Map result = new LinkedHashMap(tuple.size()); - for (Map.Entry entry : tuple.entrySet()) { - result.put(decode(entry.getKey()), entry.getValue()); - } - return result; + return DecodeUtils.decodeMap(tuple); } + static Set convertToSet(Collection keys) { + return DecodeUtils.convertToSet(keys); + } static Sort applySortingParams(Sort jredisSort, SortParameters params, byte[] storeKey) { if (params != null) { diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java index 1983555d5..a95e5a2d3 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java @@ -15,16 +15,21 @@ */ package org.springframework.data.keyvalue.redis.connection.rjc; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Properties; import java.util.Set; +import org.idevlab.rjc.Client; import org.idevlab.rjc.RedisException; import org.idevlab.rjc.Session; import org.idevlab.rjc.SessionFactoryImpl; +import org.idevlab.rjc.SortingParams; +import org.idevlab.rjc.ZParams; import org.springframework.dao.DataAccessException; import org.springframework.data.keyvalue.UncategorizedKeyvalueStoreException; +import org.springframework.data.keyvalue.redis.SubscribedRedisConnectionException; import org.springframework.data.keyvalue.redis.connection.DataType; import org.springframework.data.keyvalue.redis.connection.MessageListener; import org.springframework.data.keyvalue.redis.connection.RedisConnection; @@ -39,11 +44,16 @@ import org.springframework.data.keyvalue.redis.connection.Subscription; public class RjcConnection implements RedisConnection { private final int dbIndex; - private final Session session; private boolean isClosed = false; + private final Client client; + private final Session session; + private volatile Client pipeline; + public RjcConnection(org.idevlab.rjc.ds.RedisConnection connection, int dbIndex) { session = new SessionFactoryImpl(new SingleDataSource(connection)).create(); + client = new Client(connection); + this.dbIndex = dbIndex; // select the db @@ -81,629 +91,1955 @@ public class RjcConnection implements RedisConnection { } @Override - public List closePipeline() { - throw new UnsupportedOperationException(); + public boolean isQueueing() { + return client.isInMulti(); } - @Override public boolean isPipelined() { - throw new UnsupportedOperationException(); - } - - @Override - public boolean isQueueing() { - throw new UnsupportedOperationException(); + return (pipeline != null); } @Override public void openPipeline() { - throw new UnsupportedOperationException(); + if (pipeline == null) { + pipeline = client; + } } + @SuppressWarnings("unchecked") @Override - public Long del(byte[]... keys) { - throw new UnsupportedOperationException(); - } - - @Override - public byte[] echo(byte[] message) { - throw new UnsupportedOperationException(); - } - - @Override - public Boolean exists(byte[] key) { - throw new UnsupportedOperationException(); - } - - @Override - public Boolean expire(byte[] key, long seconds) { - throw new UnsupportedOperationException(); - } - - @Override - public Boolean expireAt(byte[] key, long unixTime) { - throw new UnsupportedOperationException(); - } - - @Override - public Set keys(byte[] pattern) { - throw new UnsupportedOperationException(); - } - - @Override - public Boolean persist(byte[] key) { - throw new UnsupportedOperationException(); - } - - @Override - public String ping() { - throw new UnsupportedOperationException(); - } - - @Override - public byte[] randomKey() { - throw new UnsupportedOperationException(); - } - - @Override - public void rename(byte[] oldName, byte[] newName) { - throw new UnsupportedOperationException(); - } - - @Override - public Boolean renameNX(byte[] oldName, byte[] newName) { - throw new UnsupportedOperationException(); - } - - @Override - public void select(int dbIndex) { - throw new UnsupportedOperationException(); + public List closePipeline() { + if (pipeline != null) { + List execute = client.getAll(); + if (execute != null && !execute.isEmpty()) { + return (List) execute; + } + } + return Collections.emptyList(); } @Override public List sort(byte[] key, SortParameters params) { - throw new UnsupportedOperationException(); - } - - @Override - public Long sort(byte[] key, SortParameters params, byte[] storeKey) { - throw new UnsupportedOperationException(); - } - - @Override - public Long ttl(byte[] key) { - throw new UnsupportedOperationException(); - } - - @Override - public DataType type(byte[] key) { - throw new UnsupportedOperationException(); - } - - @Override - public void discard() { - throw new UnsupportedOperationException(); - } - - @Override - public List exec() { - throw new UnsupportedOperationException(); - } - - @Override - public void multi() { - throw new UnsupportedOperationException(); - } - - @Override - public void unwatch() { - throw new UnsupportedOperationException(); - } - - @Override - public void watch(byte[]... keys) { - throw new UnsupportedOperationException(); - } - - @Override - public Long append(byte[] key, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public Long decr(byte[] key) { - throw new UnsupportedOperationException(); - } - - @Override - public Long decrBy(byte[] key, long value) { - throw new UnsupportedOperationException(); - } - - @Override - public byte[] get(byte[] key) { - throw new UnsupportedOperationException(); - } - - @Override - public Boolean getBit(byte[] key, long offset) { - throw new UnsupportedOperationException(); - } - - @Override - public byte[] getRange(byte[] key, int begin, int end) { - throw new UnsupportedOperationException(); - } - - @Override - public byte[] getSet(byte[] key, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public Long incr(byte[] key) { - throw new UnsupportedOperationException(); - } - - @Override - public Long incrBy(byte[] key, long value) { - throw new UnsupportedOperationException(); - } - - @Override - public List mGet(byte[]... keys) { - throw new UnsupportedOperationException(); - } - - @Override - public void mSet(Map tuple) { - throw new UnsupportedOperationException(); - } - - @Override - public void mSetNX(Map tuple) { - throw new UnsupportedOperationException(); - } - - @Override - public void set(byte[] key, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public void setBit(byte[] key, long offset, boolean value) { - throw new UnsupportedOperationException(); - } - - @Override - public void setEx(byte[] key, long seconds, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public Boolean setNX(byte[] key, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public void setRange(byte[] key, int begin, int end) { - throw new UnsupportedOperationException(); - } - - @Override - public Long strLen(byte[] key) { - throw new UnsupportedOperationException(); - } - - @Override - public List bLPop(int timeout, byte[]... keys) { - throw new UnsupportedOperationException(); - } - - @Override - public List bRPop(int timeout, byte[]... keys) { - throw new UnsupportedOperationException(); - } - - @Override - public byte[] bRPopLPush(int timeout, byte[] srcKey, byte[] dstKey) { - throw new UnsupportedOperationException(); - } - - @Override - public byte[] lIndex(byte[] key, long index) { - throw new UnsupportedOperationException(); - } - - @Override - public Long lInsert(byte[] key, Position where, byte[] pivot, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public Long lLen(byte[] key) { - throw new UnsupportedOperationException(); - } - - @Override - public byte[] lPop(byte[] key) { - throw new UnsupportedOperationException(); - } - - @Override - public Long lPush(byte[] key, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public Long lPushX(byte[] key, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public List lRange(byte[] key, long begin, long end) { - throw new UnsupportedOperationException(); - } - - @Override - public Long lRem(byte[] key, long count, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public void lSet(byte[] key, long index, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public void lTrim(byte[] key, long begin, long end) { - throw new UnsupportedOperationException(); - } - - @Override - public byte[] rPop(byte[] key) { - throw new UnsupportedOperationException(); - } - - @Override - public byte[] rPopLPush(byte[] srcKey, byte[] dstKey) { - throw new UnsupportedOperationException(); - } - - @Override - public Long rPush(byte[] key, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public Long rPushX(byte[] key, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public Boolean sAdd(byte[] key, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public Long sCard(byte[] key) { - throw new UnsupportedOperationException(); - } - - @Override - public Set sDiff(byte[]... keys) { - throw new UnsupportedOperationException(); - } - @Override - public void sDiffStore(byte[] destKey, byte[]... keys) { - throw new UnsupportedOperationException(); - } - - @Override - public Set sInter(byte[]... keys) { - throw new UnsupportedOperationException(); - } - - @Override - public void sInterStore(byte[] destKey, byte[]... keys) { - throw new UnsupportedOperationException(); - } - - @Override - public Boolean sIsMember(byte[] key, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public Set sMembers(byte[] key) { - throw new UnsupportedOperationException(); - } - - @Override - public Boolean sMove(byte[] srcKey, byte[] destKey, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public byte[] sPop(byte[] key) { - throw new UnsupportedOperationException(); - } - - @Override - public byte[] sRandMember(byte[] key) { - throw new UnsupportedOperationException(); - } - - @Override - public Boolean sRem(byte[] key, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public Set sUnion(byte[]... keys) { - throw new UnsupportedOperationException(); - } - - @Override - public void sUnionStore(byte[] destKey, byte[]... keys) { - throw new UnsupportedOperationException(); - } - - @Override - public Boolean zAdd(byte[] key, double score, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public Long zCard(byte[] key) { - throw new UnsupportedOperationException(); - } - - @Override - public Long zCount(byte[] key, double min, double max) { - throw new UnsupportedOperationException(); - } - - @Override - public Double zIncrBy(byte[] key, double increment, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public Long zInterStore(byte[] destKey, byte[]... sets) { - throw new UnsupportedOperationException(); - } - - @Override - public Long zInterStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets) { - throw new UnsupportedOperationException(); - } - - @Override - public Set zRange(byte[] key, long begin, long end) { - throw new UnsupportedOperationException(); - } - - @Override - public Set zRangeByScore(byte[] key, double min, double max) { - throw new UnsupportedOperationException(); - } - - @Override - public Set zRangeByScore(byte[] key, double min, double max, long offset, long count) { - throw new UnsupportedOperationException(); - } - - @Override - public Set zRangeByScoreWithScore(byte[] key, double min, double max) { - throw new UnsupportedOperationException(); - } - - @Override - public Set zRangeByScoreWithScore(byte[] key, double min, double max, long offset, long count) { - throw new UnsupportedOperationException(); - } - - @Override - public Set zRangeWithScore(byte[] key, long begin, long end) { - throw new UnsupportedOperationException(); - } - - @Override - public Long zRank(byte[] key, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public Boolean zRem(byte[] key, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public Long zRemRange(byte[] key, long begin, long end) { - throw new UnsupportedOperationException(); - } - - @Override - public Long zRemRangeByScore(byte[] key, double min, double max) { - throw new UnsupportedOperationException(); - } - - @Override - public Set zRevRange(byte[] key, long begin, long end) { - throw new UnsupportedOperationException(); - } - - @Override - public Set zRevRangeWithScore(byte[] key, long begin, long end) { - throw new UnsupportedOperationException(); - } - - @Override - public Long zRevRank(byte[] key, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public Double zScore(byte[] key, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public Long zUnionStore(byte[] destKey, byte[]... sets) { - throw new UnsupportedOperationException(); - } - - @Override - public Long zUnionStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets) { - throw new UnsupportedOperationException(); - } - - @Override - public Boolean hDel(byte[] key, byte[] field) { - throw new UnsupportedOperationException(); - } - - @Override - public Boolean hExists(byte[] key, byte[] field) { - throw new UnsupportedOperationException(); - } - - @Override - public byte[] hGet(byte[] key, byte[] field) { - throw new UnsupportedOperationException(); - } - - @Override - public Map hGetAll(byte[] key) { - throw new UnsupportedOperationException(); - } - - @Override - public Long hIncrBy(byte[] key, byte[] field, long delta) { - throw new UnsupportedOperationException(); - } - - @Override - public Set hKeys(byte[] key) { - throw new UnsupportedOperationException(); - } - - @Override - public Long hLen(byte[] key) { - throw new UnsupportedOperationException(); - } - - @Override - public List hMGet(byte[] key, byte[]... fields) { - throw new UnsupportedOperationException(); - } - - @Override - public void hMSet(byte[] key, Map hashes) { - throw new UnsupportedOperationException(); - } - - @Override - public Boolean hSet(byte[] key, byte[] field, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public Boolean hSetNX(byte[] key, byte[] field, byte[] value) { - throw new UnsupportedOperationException(); - } - - @Override - public List hVals(byte[] key) { - throw new UnsupportedOperationException(); - } - - @Override - public void bgSave() { - throw new UnsupportedOperationException(); - } - - @Override - public void bgWriteAof() { - throw new UnsupportedOperationException(); + SortingParams sortParams = RjcUtils.convertSortParams(params); + final String stringKey = RjcUtils.decode(key); + + try { + if (isPipelined()) { + if (sortParams != null) { + pipeline.sort(stringKey, sortParams); + } + else { + pipeline.sort(stringKey); + } + + return null; + } + return RjcUtils.convertToList((sortParams != null ? session.sort(stringKey, sortParams) + : session.sort(stringKey))); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long sort(byte[] key, SortParameters params, byte[] sortKey) { + + SortingParams sortParams = RjcUtils.convertSortParams(params); + final String stringKey = RjcUtils.decode(key); + final String stringSortKey = RjcUtils.decode(sortKey); + + try { + if (isPipelined()) { + if (sortParams != null) { + pipeline.sort(stringKey, sortParams, stringSortKey); + } + else { + pipeline.sort(stringKey, stringSortKey); + } + + return null; + } + return (sortParams != null ? session.sort(stringKey, sortParams, stringSortKey) : session.sort(stringKey, + stringSortKey)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } } @Override public Long dbSize() { - throw new UnsupportedOperationException(); + try { + if (isPipelined()) { + pipeline.dbSize(); + return null; + } + return session.dbSize(); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + + @Override + public void flushDb() { + try { + if (isPipelined()) { + pipeline.flushDB(); + return; + } + session.flushDB(); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } } @Override public void flushAll() { - throw new UnsupportedOperationException(); + try { + if (isPipelined()) { + pipeline.flushAll(); + return; + } + session.flushAll(); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } } @Override - public void flushDb() { - throw new UnsupportedOperationException(); + public void bgSave() { + try { + if (isPipelined()) { + pipeline.bgsave(); + return; + } + session.bgsave(); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } } @Override - public List getConfig(String pattern) { - throw new UnsupportedOperationException(); - } - - @Override - public Properties info() { - throw new UnsupportedOperationException(); - } - - @Override - public Long lastSave() { - throw new UnsupportedOperationException(); - } - - @Override - public void resetConfigStats() { - throw new UnsupportedOperationException(); + public void bgWriteAof() { + try { + if (isPipelined()) { + pipeline.bgrewriteaof(); + return; + } + session.bgrewriteaof(); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } } @Override public void save() { - throw new UnsupportedOperationException(); + try { + if (isPipelined()) { + pipeline.save(); + return; + } + session.save(); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public List getConfig(String param) { + try { + if (isPipelined()) { + pipeline.configGet(param); + return null; + } + return session.configGet(param); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Properties info() { + try { + if (isPipelined()) { + pipeline.info(); + return null; + } + return RjcUtils.info(session.info()); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long lastSave() { + try { + if (isPipelined()) { + pipeline.lastsave(); + return null; + } + return session.lastsave(); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } } @Override public void setConfig(String param, String value) { - throw new UnsupportedOperationException(); + try { + if (isPipelined()) { + pipeline.configSet(param, value); + return; + } + session.configSet(param, value); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + + @Override + public void resetConfigStats() { + try { + if (isPipelined()) { + pipeline.configResetStat(); + return; + } + client.configResetStat(); + client.getStatusCodeReply(); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } } @Override public void shutdown() { - throw new UnsupportedOperationException(); + try { + if (isPipelined()) { + pipeline.shutdown(); + return; + } + session.shutdown(); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public byte[] echo(byte[] message) { + String stringMsg = RjcUtils.decode(message); + try { + if (isPipelined()) { + pipeline.echo(stringMsg); + return null; + } + return RjcUtils.encode(session.echo(stringMsg)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public String ping() { + try { + if (isPipelined()) { + pipeline.ping(); + } + return session.ping(); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long del(byte[]... keys) { + String[] stringKeys = RjcUtils.decodeMultiple(keys); + + try { + if (isPipelined()) { + pipeline.del(stringKeys); + return null; + } + return session.del(stringKeys); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public void discard() { + try { + if (isPipelined()) { + pipeline.discard(); + return; + } + + session.discard(); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public List exec() { + try { + if (isPipelined()) { + pipeline.exec(); + return null; + } + return session.exec(); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Boolean exists(byte[] key) { + String stringKey = RjcUtils.decode(key); + + try { + if (isPipelined()) { + pipeline.exists(stringKey); + return null; + } + return session.exists(stringKey); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Boolean expire(byte[] key, long seconds) { + String stringKey = RjcUtils.decode(key); + + try { + if (isPipelined()) { + pipeline.expire(stringKey, (int) seconds); + return null; + } + return session.expire(stringKey, (int) seconds); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Boolean expireAt(byte[] key, long unixTime) { + String stringKey = RjcUtils.decode(key); + + try { + if (isPipelined()) { + pipeline.expireAt(stringKey, unixTime); + return null; + } + return session.expireAt(stringKey, unixTime); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Set keys(byte[] pattern) { + String stringKey = RjcUtils.decode(pattern); + + try { + if (isPipelined()) { + pipeline.keys(stringKey); + return null; + } + return RjcUtils.convertToSet(session.keys(stringKey)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public void multi() { + if (isQueueing()) { + return; + } + try { + if (isPipelined()) { + pipeline.multi(); + return; + } + session.multi(); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Boolean persist(byte[] key) { + String stringKey = RjcUtils.decode(key); + + try { + if (isPipelined()) { + pipeline.persist(stringKey); + return null; + } + return session.persist(stringKey); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public byte[] randomKey() { + try { + if (isPipelined()) { + pipeline.randomKey(); + return null; + } + return RjcUtils.encode(session.randomKey()); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public void rename(byte[] oldName, byte[] newName) { + String stringOldKey = RjcUtils.decode(oldName); + String stringNewKey = RjcUtils.decode(newName); + + try { + if (isPipelined()) { + pipeline.rename(stringOldKey, stringNewKey); + return; + } + session.rename(stringOldKey, stringNewKey); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Boolean renameNX(byte[] oldName, byte[] newName) { + String stringOldKey = RjcUtils.decode(oldName); + String stringNewKey = RjcUtils.decode(newName); + + try { + if (isPipelined()) { + pipeline.renamenx(stringOldKey, stringNewKey); + return null; + } + return session.renamenx(stringOldKey, stringNewKey); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public void select(int dbIndex) { + try { + if (isPipelined()) { + pipeline.select(dbIndex); + return; + } + session.select(dbIndex); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long ttl(byte[] key) { + String stringKey = RjcUtils.decode(key); + + try { + if (isPipelined()) { + pipeline.ttl(stringKey); + return null; + } + return session.ttl(stringKey); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public DataType type(byte[] key) { + String stringKey = RjcUtils.decode(key); + + try { + if (isPipelined()) { + pipeline.type(stringKey); + return null; + } + return DataType.fromCode(session.type(stringKey)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public void unwatch() { + try { + if (isPipelined()) { + pipeline.unwatch(); + return; + } + + session.unwatch(); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public void watch(byte[]... keys) { + String[] stringKeys = RjcUtils.decodeMultiple(keys); + + if (isQueueing()) { + return; + } + try { + if (isPipelined()) { + pipeline.watch(stringKeys); + return; + } + else { + session.watch(stringKeys); + } + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + // + // String commands + // + + @Override + public byte[] get(byte[] key) { + String stringKey = RjcUtils.decode(key); + + try { + if (isPipelined()) { + pipeline.get(stringKey); + return null; + } + + return RjcUtils.encode(session.get(stringKey)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public void set(byte[] key, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(key); + + try { + if (isPipelined()) { + pipeline.set(stringKey, stringValue); + return; + } + session.set(stringKey, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + + @Override + public byte[] getSet(byte[] key, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(key); + + try { + if (isPipelined()) { + pipeline.getSet(stringKey, stringValue); + return null; + } + return RjcUtils.encode(session.getSet(stringKey, stringValue)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long append(byte[] key, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(key); + + try { + if (isPipelined()) { + pipeline.append(stringKey, stringValue); + return null; + } + return session.append(stringKey, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public List mGet(byte[]... keys) { + String[] stringKeys = RjcUtils.decodeMultiple(keys); + + try { + if (isPipelined()) { + pipeline.mget(stringKeys); + return null; + } + return RjcUtils.convertToList(session.mget(stringKeys)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public void mSet(Map tuples) { + String[] decodeMap = RjcUtils.flatten(tuples); + + try { + if (isPipelined()) { + pipeline.mset(decodeMap); + return; + } + session.mset(decodeMap); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public void mSetNX(Map tuples) { + String[] decodeMap = RjcUtils.flatten(tuples); + + try { + + if (isPipelined()) { + pipeline.msetnx(decodeMap); + return; + } + session.msetnx(decodeMap); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public void setEx(byte[] key, long time, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(value); + + try { + if (isPipelined()) { + pipeline.setex(stringKey, (int) time, stringValue); + return; + } + session.setex(stringKey, (int) time, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Boolean setNX(byte[] key, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(value); + + try { + if (isPipelined()) { + pipeline.setnx(stringKey, stringValue); + return null; + } + return session.setnx(stringKey, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public byte[] getRange(byte[] key, long start, long end) { + String stringKey = RjcUtils.decode(key); + + try { + if (isPipelined()) { + pipeline.getRange(stringKey, (int) start, (int) end); + return null; + } + return RjcUtils.encode(session.getRange(stringKey, (int) start, (int) end)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long decr(byte[] key) { + String stringKey = RjcUtils.decode(key); + try { + + if (isPipelined()) { + pipeline.decr(stringKey); + return null; + } + return session.decr(stringKey); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long decrBy(byte[] key, long value) { + String stringKey = RjcUtils.decode(key); + try { + + if (isPipelined()) { + pipeline.decrBy(stringKey, (int) value); + return null; + } + return session.decrBy(stringKey, (int) value); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long incr(byte[] key) { + String stringKey = RjcUtils.decode(key); + + try { + + if (isPipelined()) { + pipeline.incr(stringKey); + return null; + } + return session.incr(stringKey); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long incrBy(byte[] key, long value) { + String stringKey = RjcUtils.decode(key); + + + try { + if (isPipelined()) { + pipeline.incrBy(stringKey, (int) value); + return null; + } + return session.incrBy(stringKey, (int) value); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Boolean getBit(byte[] key, long offset) { + String stringKey = RjcUtils.decode(key); + + try { + if (isPipelined()) { + pipeline.getbit(stringKey, (int) offset); + return null; + } + return (session.getBit(stringKey, (int) offset) == 0 ? Boolean.FALSE : Boolean.TRUE); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public void setBit(byte[] key, long offset, boolean value) { + String stringKey = RjcUtils.decode(key); + + try { + if (isPipelined()) { + pipeline.setbit(stringKey, (int) offset, RjcUtils.asBit(value)); + return; + } + session.setBit(stringKey, (int) offset, RjcUtils.asBit(value)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public void setRange(byte[] key, long offset, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(value); + + try { + if (isPipelined()) { + pipeline.setRange(stringKey, (int) offset, stringValue); + return; + } + session.setRange(stringKey, (int) offset, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long strLen(byte[] key) { + String stringKey = RjcUtils.decode(key); + + try { + if (isPipelined()) { + pipeline.strlen(stringKey); + return null; + } + return session.strlen(stringKey); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + // + // List commands + // + + @Override + public Long lPush(byte[] key, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(value); + + try { + if (isPipelined()) { + pipeline.lpush(stringKey, stringValue); + return null; + } + return session.lpush(stringKey, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long rPush(byte[] key, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(value); + + try { + + if (isPipelined()) { + pipeline.rpush(stringKey, stringValue); + return null; + } + return session.rpush(stringKey, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public List bLPop(int timeout, byte[]... keys) { + String[] stringKeys = RjcUtils.decodeMultiple(keys); + + try { + if (isPipelined()) { + pipeline.blpop(stringKeys); + return null; + } + return RjcUtils.convertToList(session.blpop(timeout, stringKeys)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public List bRPop(int timeout, byte[]... keys) { + String[] stringKeys = RjcUtils.decodeMultiple(keys); + + try { + if (isPipelined()) { + pipeline.brpop(stringKeys); + return null; + } + return RjcUtils.convertToList(session.brpop(timeout, stringKeys)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public byte[] lIndex(byte[] key, long index) { + String stringKey = RjcUtils.decode(key); + + + try { + if (isPipelined()) { + pipeline.lindex(stringKey, (int) index); + return null; + } + return RjcUtils.encode(session.lindex(stringKey, (int) index)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long lInsert(byte[] key, Position where, byte[] pivot, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(value); + String stringPivot = RjcUtils.decode(pivot); + Client.LIST_POSITION position = RjcUtils.convertPosition(where); + + try { + if (isPipelined()) { + pipeline.linsert(stringKey, position, stringPivot, stringValue); + return null; + } + return session.linsert(stringKey, position, stringPivot, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long lLen(byte[] key) { + String stringKey = RjcUtils.decode(key); + + try { + + if (isPipelined()) { + pipeline.llen(stringKey); + return null; + } + return session.llen(stringKey); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public byte[] lPop(byte[] key) { + String stringKey = RjcUtils.decode(key); + + try { + + if (isPipelined()) { + pipeline.lpop(stringKey); + return null; + } + return RjcUtils.encode(session.lpop(stringKey)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public List lRange(byte[] key, long start, long end) { + String stringKey = RjcUtils.decode(key); + + try { + + if (isPipelined()) { + pipeline.lrange(stringKey, (int) start, (int) end); + return null; + } + return RjcUtils.convertToList(session.lrange(stringKey, (int) start, (int) end)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long lRem(byte[] key, long count, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(value); + + try { + + if (isPipelined()) { + pipeline.lrem(stringKey, (int) count, stringValue); + return null; + } + return session.lrem(stringKey, (int) count, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public void lSet(byte[] key, long index, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(value); + try { + + if (isPipelined()) { + pipeline.lset(stringKey, (int) index, stringValue); + return; + } + session.lset(stringKey, (int) index, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public void lTrim(byte[] key, long start, long end) { + String stringKey = RjcUtils.decode(key); + + try { + + if (isPipelined()) { + pipeline.ltrim(stringKey, (int) start, (int) end); + return; + } + session.ltrim(stringKey, (int) start, (int) end); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public byte[] rPop(byte[] key) { + String stringKey = RjcUtils.decode(key); + + try { + + if (isPipelined()) { + pipeline.rpop(stringKey); + return null; + } + return RjcUtils.encode(session.rpop(stringKey)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public byte[] rPopLPush(byte[] srcKey, byte[] dstKey) { + String stringKey = RjcUtils.decode(srcKey); + String stringDest = RjcUtils.decode(dstKey); + + try { + + if (isPipelined()) { + pipeline.rpoplpush(stringKey, stringDest); + return null; + } + return RjcUtils.encode(session.rpoplpush(stringKey, stringDest)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public byte[] bRPopLPush(int timeout, byte[] srcKey, byte[] dstKey) { + String stringKey = RjcUtils.decode(srcKey); + String stringDest = RjcUtils.decode(dstKey); + + try { + if (isPipelined()) { + pipeline.brpoplpush(stringKey, stringDest, timeout); + return null; + } + return RjcUtils.encode(session.brpoplpush(stringKey, stringDest, timeout)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long lPushX(byte[] key, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(value); + try { + if (isPipelined()) { + pipeline.lpushx(stringKey, stringValue); + return null; + } + return session.lpushx(stringKey, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long rPushX(byte[] key, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(value); + try { + if (isPipelined()) { + pipeline.rpushx(stringKey, stringValue); + return null; + } + return session.rpushx(stringKey, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + + // + // Set commands + // + + @Override + public Boolean sAdd(byte[] key, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(value); + + try { + + if (isPipelined()) { + pipeline.sadd(stringKey, stringValue); + return null; + } + return session.sadd(stringKey, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long sCard(byte[] key) { + String stringKey = RjcUtils.decode(key); + + try { + + if (isPipelined()) { + pipeline.scard(stringKey); + return null; + } + return session.scard(stringKey); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Set sDiff(byte[]... keys) { + String[] stringKeys = RjcUtils.decodeMultiple(keys); + + try { + + if (isPipelined()) { + pipeline.sdiff(stringKeys); + return null; + } + return RjcUtils.convertToSet(session.sdiff(stringKeys)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public void sDiffStore(byte[] destKey, byte[]... keys) { + String stringKey = RjcUtils.decode(destKey); + String[] stringKeys = RjcUtils.decodeMultiple(keys); + + try { + + if (isPipelined()) { + pipeline.sdiffstore(stringKey, stringKeys); + return; + } + session.sdiffstore(stringKey, stringKeys); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Set sInter(byte[]... keys) { + String[] stringKeys = RjcUtils.decodeMultiple(keys); + try { + + if (isPipelined()) { + pipeline.sinter(stringKeys); + return null; + } + return RjcUtils.convertToSet(session.sinter(stringKeys)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public void sInterStore(byte[] destKey, byte[]... keys) { + String stringKey = RjcUtils.decode(destKey); + String[] stringKeys = RjcUtils.decodeMultiple(keys); + try { + + if (isPipelined()) { + pipeline.sinterstore(stringKey, stringKeys); + return; + } + session.sinterstore(stringKey, stringKeys); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Boolean sIsMember(byte[] key, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(value); + + try { + + if (isPipelined()) { + pipeline.sismember(stringKey, stringValue); + return null; + } + return session.sismember(stringKey, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Set sMembers(byte[] key) { + String stringKey = RjcUtils.decode(key); + try { + + if (isPipelined()) { + pipeline.smembers(stringKey); + return null; + } + return RjcUtils.convertToSet(session.smembers(stringKey)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Boolean sMove(byte[] srcKey, byte[] destKey, byte[] value) { + String stringSrc = RjcUtils.decode(srcKey); + String stringDest = RjcUtils.decode(destKey); + String stringValue = RjcUtils.decode(value); + + try { + + if (isPipelined()) { + pipeline.smove(stringSrc, stringDest, stringValue); + return null; + } + return session.smove(stringSrc, stringDest, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public byte[] sPop(byte[] key) { + String stringKey = RjcUtils.decode(key); + try { + + if (isPipelined()) { + pipeline.spop(stringKey); + return null; + } + return RjcUtils.encode(session.spop(stringKey)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public byte[] sRandMember(byte[] key) { + String stringKey = RjcUtils.decode(key); + try { + + if (isPipelined()) { + pipeline.srandmember(stringKey); + return null; + } + return RjcUtils.encode(session.srandmember(stringKey)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Boolean sRem(byte[] key, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(value); + + try { + + if (isPipelined()) { + pipeline.srem(stringKey, stringValue); + return null; + } + return session.srem(stringKey, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Set sUnion(byte[]... keys) { + String[] stringKeys = RjcUtils.decodeMultiple(keys); + + try { + + if (isPipelined()) { + pipeline.sunion(stringKeys); + return null; + } + return RjcUtils.convertToSet(session.sunion(stringKeys)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public void sUnionStore(byte[] destKey, byte[]... keys) { + String stringKey = RjcUtils.decode(destKey); + String[] stringKeys = RjcUtils.decodeMultiple(keys); + + try { + + if (isPipelined()) { + pipeline.sunionstore(stringKey, stringKeys); + return; + } + session.sunionstore(stringKey, stringKeys); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + // + // ZSet commands + // + + @Override + public Boolean zAdd(byte[] key, double score, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(value); + + try { + if (isPipelined()) { + pipeline.zadd(stringKey, score, stringValue); + return null; + } + return session.zadd(stringKey, score, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long zCard(byte[] key) { + String stringKey = RjcUtils.decode(key); + + try { + if (isPipelined()) { + pipeline.zcard(stringKey); + return null; + } + return session.zcard(stringKey); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long zCount(byte[] key, double min, double max) { + String stringKey = RjcUtils.decode(key); + try { + if (isPipelined()) { + pipeline.zcount(stringKey, min, max); + return null; + } + + return session.zcount(stringKey, min, max); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Double zIncrBy(byte[] key, double increment, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(value); + + try { + if (isPipelined()) { + pipeline.zincrby(stringKey, increment, stringValue); + return null; + } + return Double.valueOf(session.zincrby(stringKey, increment, stringValue)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long zInterStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets) { + String stringKey = RjcUtils.decode(destKey); + String[] stringKeys = RjcUtils.decodeMultiple(sets); + + ZParams zparams = RjcUtils.toZParams(aggregate, weights); + + try { + if (isPipelined()) { + pipeline.zinterstore(stringKey, zparams, stringKeys); + return null; + } + return session.zinterstore(stringKey, zparams, stringKeys); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long zInterStore(byte[] destKey, byte[]... sets) { + String stringKey = RjcUtils.decode(destKey); + String[] stringKeys = RjcUtils.decodeMultiple(sets); + try { + if (isPipelined()) { + pipeline.zinterstore(stringKey, stringKeys); + return null; + } + + return session.zinterstore(stringKey, stringKeys); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Set zRange(byte[] key, long start, long end) { + String stringKey = RjcUtils.decode(key); + try { + + if (isPipelined()) { + pipeline.zrange(stringKey, (int) start, (int) end); + return null; + } + return RjcUtils.convertToSet(session.zrange(stringKey, (int) start, (int) end)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Set zRangeWithScore(byte[] key, long start, long end) { + String stringKey = RjcUtils.decode(key); + try { + + if (isPipelined()) { + pipeline.zrangeWithScores(stringKey, (int) start, (int) end); + return null; + } + return RjcUtils.convertElementScore(session.zrangeWithScores(stringKey, (int) start, (int) end)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Set zRangeByScore(byte[] key, double min, double max) { + String stringKey = RjcUtils.decode(key); + String minString = Double.toString(min); + String maxString = Double.toString(max); + + try { + if (isPipelined()) { + pipeline.zrangeByScore(stringKey, minString, maxString); + return null; + } + return RjcUtils.convertToSet(session.zrangeByScore(stringKey, minString, maxString)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Set zRangeByScoreWithScore(byte[] key, double min, double max) { + String stringKey = RjcUtils.decode(key); + String minString = Double.toString(min); + String maxString = Double.toString(max); + + try { + if (isPipelined()) { + pipeline.zrangeByScoreWithScores(stringKey, minString, maxString); + return null; + } + return RjcUtils.convertElementScore(session.zrangeByScoreWithScores(stringKey, minString, maxString)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Set zRevRangeWithScore(byte[] key, long start, long end) { + String stringKey = RjcUtils.decode(key); + String minString = Long.toString(start); + String maxString = Long.toString(end); + + try { + + if (isPipelined()) { + pipeline.zrangeByScoreWithScores(stringKey, minString, maxString); + return null; + } + return RjcUtils.convertElementScore(session.zrangeByScoreWithScores(stringKey, minString, maxString)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Set zRangeByScore(byte[] key, double min, double max, long offset, long count) { + String stringKey = RjcUtils.decode(key); + String minString = Double.toString(min); + String maxString = Double.toString(max); + + try { + if (isPipelined()) { + pipeline.zrangeByScore(stringKey, minString, maxString, (int) offset, (int) count); + return null; + } + return RjcUtils.convertToSet(session.zrangeByScore(stringKey, minString, maxString, (int) offset, + (int) count)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Set zRangeByScoreWithScore(byte[] key, double min, double max, long offset, long count) { + String stringKey = RjcUtils.decode(key); + String minString = Double.toString(min); + String maxString = Double.toString(max); + + try { + if (isPipelined()) { + pipeline.zrangeByScoreWithScores(stringKey, minString, maxString, (int) offset, (int) count); + return null; + } + return RjcUtils.convertElementScore(session.zrangeByScoreWithScores(stringKey, minString, maxString, + (int) offset, (int) count)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long zRank(byte[] key, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(value); + + try { + if (isPipelined()) { + pipeline.zrank(stringKey, stringValue); + return null; + } + return session.zrank(stringKey, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Boolean zRem(byte[] key, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(value); + + try { + if (isPipelined()) { + pipeline.zrem(stringKey, stringValue); + return null; + } + return session.zrem(stringKey, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long zRemRange(byte[] key, long start, long end) { + String stringKey = RjcUtils.decode(key); + try { + if (isPipelined()) { + pipeline.zremrangeByRank(stringKey, (int) start, (int) end); + return null; + } + return session.zremrangeByRank(stringKey, (int) start, (int) end); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long zRemRangeByScore(byte[] key, double min, double max) { + String stringKey = RjcUtils.decode(key); + String minString = Double.toString(min); + String maxString = Double.toString(max); + + try { + if (isPipelined()) { + pipeline.zremrangeByScore(stringKey, minString, maxString); + return null; + } + return session.zremrangeByScore(stringKey, minString, maxString); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Set zRevRange(byte[] key, long start, long end) { + String stringKey = RjcUtils.decode(key); + try { + + if (isPipelined()) { + pipeline.zrevrange(stringKey, (int) start, (int) end); + return null; + } + return RjcUtils.convertToSet(session.zrevrange(stringKey, (int) start, (int) end)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long zRevRank(byte[] key, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(value); + + try { + if (isPipelined()) { + pipeline.zrevrank(stringKey, stringValue); + return null; + } + return session.zrevrank(stringKey, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Double zScore(byte[] key, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringValue = RjcUtils.decode(value); + + try { + if (isPipelined()) { + pipeline.zscore(stringKey, stringValue); + return null; + } + return Double.valueOf(session.zscore(stringKey, stringValue)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long zUnionStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets) { + String stringKey = RjcUtils.decode(destKey); + String[] stringKeys = RjcUtils.decodeMultiple(destKey); + + ZParams zparams = RjcUtils.toZParams(aggregate, weights); + + try { + if (isPipelined()) { + pipeline.zunionstore(stringKey, zparams, stringKeys); + return null; + } + return session.zunionstore(stringKey, zparams, stringKeys); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long zUnionStore(byte[] destKey, byte[]... sets) { + String stringKey = RjcUtils.decode(destKey); + String[] stringKeys = RjcUtils.decodeMultiple(sets); + + try { + if (isPipelined()) { + pipeline.zunionstore(stringKey, stringKeys); + return null; + } + return session.zunionstore(stringKey, stringKeys); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + // + // Hash commands + // + + @Override + public Boolean hSet(byte[] key, byte[] field, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringField = RjcUtils.decode(field); + String stringValue = RjcUtils.decode(value); + + try { + if (isPipelined()) { + pipeline.hset(stringKey, stringField, stringValue); + return null; + } + return session.hset(stringKey, stringField, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Boolean hSetNX(byte[] key, byte[] field, byte[] value) { + String stringKey = RjcUtils.decode(key); + String stringField = RjcUtils.decode(field); + String stringValue = RjcUtils.decode(value); + + try { + if (isPipelined()) { + pipeline.hsetnx(stringKey, stringField, stringValue); + return null; + } + return session.hsetnx(stringKey, stringField, stringValue); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Boolean hDel(byte[] key, byte[] field) { + String stringKey = RjcUtils.decode(key); + String stringField = RjcUtils.decode(field); + + try { + if (isPipelined()) { + pipeline.hdel(stringKey, stringField); + return null; + } + return session.hdel(stringKey, stringField); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Boolean hExists(byte[] key, byte[] field) { + String stringKey = RjcUtils.decode(key); + String stringField = RjcUtils.decode(field); + + try { + if (isPipelined()) { + pipeline.hexists(stringKey, stringField); + return null; + } + return session.hexists(stringKey, stringField); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public byte[] hGet(byte[] key, byte[] field) { + String stringKey = RjcUtils.decode(key); + String stringField = RjcUtils.decode(field); + + try { + if (isPipelined()) { + pipeline.hget(stringKey, stringField); + return null; + } + return RjcUtils.encode(session.hget(stringKey, stringField)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Map hGetAll(byte[] key) { + String stringKey = RjcUtils.decode(key); + + try { + if (isPipelined()) { + pipeline.hgetAll(stringKey); + return null; + } + return RjcUtils.encodeMap(session.hgetAll(stringKey)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long hIncrBy(byte[] key, byte[] field, long delta) { + String stringKey = RjcUtils.decode(key); + String stringField = RjcUtils.decode(field); + + try { + if (isPipelined()) { + pipeline.hincrBy(stringKey, stringField, (int) delta); + return null; + } + return session.hincrBy(stringKey, stringField, (int) delta); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Set hKeys(byte[] key) { + String stringKey = RjcUtils.decode(key); + try { + if (isPipelined()) { + pipeline.hkeys(stringKey); + return null; + } + return RjcUtils.convertToSet(session.hkeys(stringKey)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public Long hLen(byte[] key) { + String stringKey = RjcUtils.decode(key); + try { + if (isPipelined()) { + pipeline.hlen(stringKey); + return null; + } + return session.hlen(stringKey); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public List hMGet(byte[] key, byte[]... fields) { + String stringKey = RjcUtils.decode(key); + String[] stringKeys = RjcUtils.decodeMultiple(fields); + + try { + if (isPipelined()) { + pipeline.hmget(stringKey, stringKeys); + return null; + } + return RjcUtils.convertToList(session.hmget(stringKey, stringKeys)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public void hMSet(byte[] key, Map tuple) { + String stringKey = RjcUtils.decode(key); + Map stringTuple = RjcUtils.decodeMap(tuple); + + try { + if (isPipelined()) { + pipeline.hmset(stringKey, stringTuple); + return; + } + session.hmset(stringKey, stringTuple); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + @Override + public List hVals(byte[] key) { + String stringKey = RjcUtils.decode(key); + try { + + if (isPipelined()) { + pipeline.hvals(stringKey); + return null; + } + return RjcUtils.convertToList(session.hvals(stringKey)); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } + } + + + // + // Pub/Sub functionality + // + @Override + public Long publish(byte[] channel, byte[] message) { + try { + if (isQueueing()) { + throw new UnsupportedOperationException(); + } + if (isPipelined()) { + throw new UnsupportedOperationException(); + } + return session.publish(channel, message); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } } @Override public Subscription getSubscription() { - throw new UnsupportedOperationException(); + return subscription; } @Override public boolean isSubscribed() { - throw new UnsupportedOperationException(); + return (subscription != null && subscription.isAlive()); } @Override public void pSubscribe(MessageListener listener, byte[]... patterns) { - throw new UnsupportedOperationException(); - } + String[] stringKeys = RjcUtils.decodeMultiple(patterns); - @Override - public Long publish(byte[] channel, byte[] message) { - throw new UnsupportedOperationException(); + if (isSubscribed()) { + throw new SubscribedRedisConnectionException( + "Connection already subscribed; use the connection Subscription to cancel or add new channels"); + } + + try { + if (isQueueing()) { + throw new UnsupportedOperationException(); + } + if (isPipelined()) { + throw new UnsupportedOperationException(); + } + + BinarysessionPubSub sessionPubSub = RjcUtils.adaptPubSub(listener); + + subscription = new sessionSubscription(listener, sessionPubSub, null, patterns); + session.psubscribe(sessionPubSub, patterns); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } } @Override public void subscribe(MessageListener listener, byte[]... channels) { - throw new UnsupportedOperationException(); + String[] stringKeys = RjcUtils.decodeMultiple(channels); + + if (isSubscribed()) { + throw new SubscribedRedisConnectionException( + "Connection already subscribed; use the connection Subscription to cancel or add new channels"); + } + + try { + if (isQueueing()) { + throw new UnsupportedOperationException(); + } + if (isPipelined()) { + throw new UnsupportedOperationException(); + } + + BinarysessionPubSub sessionPubSub = RjcUtils.adaptPubSub(listener); + + subscription = new sessionSubscription(listener, sessionPubSub, channels, null); + session.subscribe(sessionPubSub, channels); + } catch (Exception ex) { + throw convertRjcAccessException(ex); + } } -} + private void checkSubscription() { + if (isSubscribed()) { + throw new SubscribedRedisConnectionException("Cannot execute command - connection is subscribed"); + } + } +} \ No newline at end of file diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnectionFactory.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnectionFactory.java index 97f1c65cd..5c149f107 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnectionFactory.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnectionFactory.java @@ -87,7 +87,7 @@ public class RjcConnectionFactory implements InitializingBean, DisposableBean, R @Override public RedisConnection getConnection() { - return postProcessConnection(new RjcConnection(dataSource.getConnection(), usePool, dbIndex)); + return postProcessConnection(new RjcConnection(dataSource.getConnection(), dbIndex)); } /** diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcUtils.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcUtils.java index 9c0370bb0..50c92316f 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcUtils.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcUtils.java @@ -15,10 +15,33 @@ */ package org.springframework.data.keyvalue.redis.connection.rjc; +import java.io.StringReader; +import java.util.Collection; +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.idevlab.rjc.ElementScore; import org.idevlab.rjc.RedisException; +import org.idevlab.rjc.SortingParams; +import org.idevlab.rjc.ZParams; +import org.idevlab.rjc.Client.LIST_POSITION; import org.springframework.dao.DataAccessException; import org.springframework.dao.InvalidDataAccessApiUsageException; import org.springframework.data.keyvalue.redis.UncategorizedRedisException; +import org.springframework.data.keyvalue.redis.connection.DataType; +import org.springframework.data.keyvalue.redis.connection.DefaultTuple; +import org.springframework.data.keyvalue.redis.connection.SortParameters; +import org.springframework.data.keyvalue.redis.connection.RedisListCommands.Position; +import org.springframework.data.keyvalue.redis.connection.RedisZSetCommands.Aggregate; +import org.springframework.data.keyvalue.redis.connection.RedisZSetCommands.Tuple; +import org.springframework.data.keyvalue.redis.connection.SortParameters.Order; +import org.springframework.data.keyvalue.redis.connection.SortParameters.Range; +import org.springframework.data.keyvalue.redis.connection.util.DecodeUtils; + /** * Helper class featuring methods for RJC connection handling, providing support for exception translation. @@ -27,6 +50,10 @@ import org.springframework.data.keyvalue.redis.UncategorizedRedisException; */ public abstract class RjcUtils { + private static final String ONE = "1"; + private static final String ZERO = "0"; + + public static DataAccessException convertRjcAccessException(RuntimeException ex) { if (ex instanceof RedisException) { return convertRjcAccessException((RedisException) ex); @@ -38,4 +65,155 @@ public abstract class RjcUtils { public static DataAccessException convertRjcAccessException(RedisException ex) { return new InvalidDataAccessApiUsageException(ex.getMessage(), ex); } -} + + static DataType convertDataType(String type) { + if ("string".equals(type)) { + return DataType.STRING; + } + else if ("list".equals(type)) { + return DataType.LIST; + } + else if ("set".equals(type)) { + return DataType.SET; + } + else if ("zset".equals(type)) { + return DataType.ZSET; + } + else if ("hash".equals(type)) { + return DataType.HASH; + } + else if ("none".equals(type)) { + return DataType.NONE; + } + + return null; + } + + static String decode(byte[] bytes) { + return DecodeUtils.decode(bytes); + } + + static byte[] encode(String string) { + return DecodeUtils.encode(string); + } + + static String[] decodeMultiple(byte[]... bytes) { + return DecodeUtils.decodeMultiple(bytes); + } + + static String[] flatten(Map tuple) { + String[] result = new String[tuple.size() * 2]; + int index = 0; + for (Map.Entry entry : tuple.entrySet()) { + result[index++] = decode(entry.getKey()); + result[index++] = decode(entry.getValue()); + } + return result; + + } + + static Set convertToSet(Collection keys) { + if (keys == null) { + return null; + } + + return DecodeUtils.convertToSet(keys); + } + + static List convertToList(Collection keys) { + if (keys == null) { + return null; + } + return DecodeUtils.convertToList(keys); + } + + static SortingParams convertSortParams(SortParameters params) { + SortingParams rjcSort = null; + + if (params != null) { + rjcSort = new SortingParams(); + + byte[] byPattern = params.getByPattern(); + if (byPattern != null) { + rjcSort.by(DecodeUtils.decode(byPattern)); + } + byte[][] getPattern = params.getGetPattern(); + + if (getPattern != null && getPattern.length > 0) { + for (byte[] bs : getPattern) { + rjcSort.get(DecodeUtils.decode(bs)); + } + } + Range limit = params.getLimit(); + if (limit != null) { + rjcSort.limit((int) limit.getStart(), (int) limit.getCount()); + } + Order order = params.getOrder(); + if (order != null && order.equals(Order.DESC)) { + rjcSort.desc(); + } + Boolean isAlpha = params.isAlphabetic(); + if (isAlpha != null && isAlpha) { + rjcSort.alpha(); + } + } + return rjcSort; + } + + static Properties info(String string) { + Properties info = new Properties(); + StringReader stringReader = new StringReader(string); + try { + info.load(stringReader); + } catch (Exception ex) { + throw new UncategorizedRedisException("Cannot read Redis info", ex); + } finally { + stringReader.close(); + } + return info; + } + + static String asBit(boolean value) { + return (value ? ONE : ZERO); + } + + static LIST_POSITION convertPosition(Position where) { + switch (where) { + case BEFORE: + return LIST_POSITION.BEFORE; + + case AFTER: + return LIST_POSITION.AFTER; + } + return null; + } + + static ZParams toZParams(Aggregate aggregate, int[] weights) { + return new ZParams().weights(weights).aggregate(ZParams.Aggregate.valueOf(aggregate.name())); + } + + static Set convertElementScore(List tuples) { + Set value = new LinkedHashSet(tuples.size()); + for (ElementScore tuple : tuples) { + value.add(new DefaultTuple(encode(tuple.getElement()), Double.valueOf(tuple.getScore()))); + } + + return value; + } + + static Map encodeMap(Map map) { + Map result = new LinkedHashMap(map.size()); + for (Map.Entry entry : map.entrySet()) { + result.put(encode(entry.getKey()), encode(entry.getValue())); + } + return result; + } + + static Map decodeMap(Map map) { + Map result = new LinkedHashMap(map.size()); + for (Map.Entry entry : map.entrySet()) { + result.put(decode(entry.getKey()), decode(entry.getValue())); + } + return result; + } +} \ No newline at end of file diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/Base64.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/util/Base64.java similarity index 99% rename from spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/Base64.java rename to spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/util/Base64.java index 6feb3a4d6..3e99472d6 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/Base64.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/util/Base64.java @@ -1,4 +1,4 @@ -package org.springframework.data.keyvalue.redis.connection.jredis; +package org.springframework.data.keyvalue.redis.connection.util; import java.util.Arrays; diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/util/DecodeUtils.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/util/DecodeUtils.java new file mode 100644 index 000000000..d3588856c --- /dev/null +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/util/DecodeUtils.java @@ -0,0 +1,82 @@ +/* + * 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.keyvalue.redis.connection.util; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * Simple class containing various decoding utilities. + * + * @author Costin Leau + */ +public abstract class DecodeUtils { + + public static String decode(byte[] bytes) { + return Base64.encodeToString(bytes, false); + } + + public static String[] decodeMultiple(byte[]... bytes) { + String[] result = new String[bytes.length]; + for (int i = 0; i < bytes.length; i++) { + result[i] = decode(bytes[i]); + } + return result; + } + + public static byte[] encode(String string) { + return Base64.decode(string); + } + + public static Map encodeMap(Map map) { + Map result = new LinkedHashMap(map.size()); + for (Map.Entry entry : map.entrySet()) { + result.put(encode(entry.getKey()), entry.getValue()); + } + return result; + } + + public static Map decodeMap(Map tuple) { + Map result = new LinkedHashMap(tuple.size()); + for (Map.Entry entry : tuple.entrySet()) { + result.put(decode(entry.getKey()), entry.getValue()); + } + return result; + } + + public static Set convertToSet(Collection keys) { + Set set = new LinkedHashSet(keys.size()); + + for (String string : keys) { + set.add(Base64.decode(string)); + } + return set; + } + + public static List convertToList(Collection keys) { + List set = new ArrayList(keys.size()); + + for (String string : keys) { + set.add(Base64.decode(string)); + } + return set; + } +} \ No newline at end of file From 79631fb6ecaabe7d9177fe67ee2a81dc54a0d0c5 Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Wed, 16 Mar 2011 14:48:49 +0200 Subject: [PATCH 3/4] DATAKV-46 + wrap up RJC connector with pub sub support --- .../redis/connection/rjc/RjcConnection.java | 25 +-- .../connection/rjc/RjcMessageListener.java | 45 ++++++ .../redis/connection/rjc/RjcSubscription.java | 149 ++++++++++++++++++ 3 files changed, 207 insertions(+), 12 deletions(-) create mode 100644 spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcMessageListener.java create mode 100644 spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcSubscription.java diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java index a95e5a2d3..93e452427 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java @@ -27,6 +27,7 @@ import org.idevlab.rjc.Session; import org.idevlab.rjc.SessionFactoryImpl; import org.idevlab.rjc.SortingParams; import org.idevlab.rjc.ZParams; +import org.idevlab.rjc.message.RedisNodeSubscriber; import org.springframework.dao.DataAccessException; import org.springframework.data.keyvalue.UncategorizedKeyvalueStoreException; import org.springframework.data.keyvalue.redis.SubscribedRedisConnectionException; @@ -50,9 +51,14 @@ public class RjcConnection implements RedisConnection { private final Session session; private volatile Client pipeline; + private volatile RjcSubscription subscription; + private volatile RedisNodeSubscriber subscriber; + public RjcConnection(org.idevlab.rjc.ds.RedisConnection connection, int dbIndex) { - session = new SessionFactoryImpl(new SingleDataSource(connection)).create(); + SingleDataSource connectionDataSource = new SingleDataSource(connection); + session = new SessionFactoryImpl().create(); client = new Client(connection); + subscriber = new RedisNodeSubscriber(connectionDataSource); this.dbIndex = dbIndex; @@ -73,6 +79,7 @@ public class RjcConnection implements RedisConnection { public void close() throws DataAccessException { isClosed = true; try { + subscriber.close(); session.close(); } catch (Exception ex) { throw convertRjcAccessException(ex); @@ -1969,7 +1976,7 @@ public class RjcConnection implements RedisConnection { if (isPipelined()) { throw new UnsupportedOperationException(); } - return session.publish(channel, message); + return session.publish(RjcUtils.decode(channel), RjcUtils.decode(message)); } catch (Exception ex) { throw convertRjcAccessException(ex); } @@ -1987,8 +1994,6 @@ public class RjcConnection implements RedisConnection { @Override public void pSubscribe(MessageListener listener, byte[]... patterns) { - String[] stringKeys = RjcUtils.decodeMultiple(patterns); - if (isSubscribed()) { throw new SubscribedRedisConnectionException( "Connection already subscribed; use the connection Subscription to cancel or add new channels"); @@ -2002,10 +2007,9 @@ public class RjcConnection implements RedisConnection { throw new UnsupportedOperationException(); } - BinarysessionPubSub sessionPubSub = RjcUtils.adaptPubSub(listener); + subscription = new RjcSubscription(listener, subscriber); + subscription.pSubscribe(patterns); - subscription = new sessionSubscription(listener, sessionPubSub, null, patterns); - session.psubscribe(sessionPubSub, patterns); } catch (Exception ex) { throw convertRjcAccessException(ex); } @@ -2013,8 +2017,6 @@ public class RjcConnection implements RedisConnection { @Override public void subscribe(MessageListener listener, byte[]... channels) { - String[] stringKeys = RjcUtils.decodeMultiple(channels); - if (isSubscribed()) { throw new SubscribedRedisConnectionException( "Connection already subscribed; use the connection Subscription to cancel or add new channels"); @@ -2028,10 +2030,9 @@ public class RjcConnection implements RedisConnection { throw new UnsupportedOperationException(); } - BinarysessionPubSub sessionPubSub = RjcUtils.adaptPubSub(listener); + subscription = new RjcSubscription(listener, subscriber); + subscription.pSubscribe(channels); - subscription = new sessionSubscription(listener, sessionPubSub, channels, null); - session.subscribe(sessionPubSub, channels); } catch (Exception ex) { throw convertRjcAccessException(ex); } diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcMessageListener.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcMessageListener.java new file mode 100644 index 000000000..c16a2040f --- /dev/null +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcMessageListener.java @@ -0,0 +1,45 @@ +/* + * 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.keyvalue.redis.connection.rjc; + +import org.idevlab.rjc.message.MessageListener; +import org.idevlab.rjc.message.PMessageListener; +import org.springframework.data.keyvalue.redis.connection.DefaultMessage; + +/** + * Message listener adapter for RJC library. + * + * @author Costin Leau + */ +class RjcMessageListener implements MessageListener, PMessageListener { + + private final org.springframework.data.keyvalue.redis.connection.MessageListener listener; + + RjcMessageListener(org.springframework.data.keyvalue.redis.connection.MessageListener messageListener) { + this.listener = messageListener; + } + + @Override + public void onMessage(String channel, String message) { + listener.onMessage(new DefaultMessage(RjcUtils.encode(channel), RjcUtils.encode(message)), null); + } + + @Override + public void onMessage(String pattern, String channel, String message) { + listener.onMessage(new DefaultMessage(RjcUtils.encode(channel), RjcUtils.encode(message)), + RjcUtils.encode(pattern)); + } +} \ No newline at end of file diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcSubscription.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcSubscription.java new file mode 100644 index 000000000..a1075a120 --- /dev/null +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcSubscription.java @@ -0,0 +1,149 @@ +/* + * 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.keyvalue.redis.connection.rjc; + +import java.util.ArrayList; +import java.util.Collection; + +import org.idevlab.rjc.message.RedisSubscriber; +import org.springframework.data.keyvalue.redis.connection.MessageListener; +import org.springframework.data.keyvalue.redis.connection.Subscription; +import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; + +/** + * Message subscription on top of RJC. + * + * @author Costin Leau + */ +class RjcSubscription implements Subscription { + + private final MessageListener listener; + private final RedisSubscriber subscriber; + private final RjcMessageListener listenerAdapter; + + private final Collection channels = new ArrayList(2); + private final Collection patterns = new ArrayList(2); + + RjcSubscription(MessageListener listener, RedisSubscriber subscriber) { + Assert.notNull(listener); + this.listener = listener; + this.subscriber = subscriber; + this.listenerAdapter = new RjcMessageListener(listener); + } + + @Override + public Collection getChannels() { + synchronized (channels) { + return new ArrayList(channels); + } + } + + @Override + public MessageListener getListener() { + return listener; + } + + @Override + public Collection getPatterns() { + synchronized (patterns) { + return new ArrayList(patterns); + } + } + + @Override + public void pSubscribe(byte[]... patterns) { + Assert.notEmpty(patterns, "at least one pattern required"); + + synchronized (this.patterns) { + for (byte[] bs : patterns) { + this.patterns.add(bs); + } + } + + for (String pattern : RjcUtils.decodeMultiple(patterns)) { + subscriber.psubscribe(pattern, listenerAdapter); + } + } + + @Override + public void pUnsubscribe() { + pUnsubscribe(null); + + synchronized (patterns) { + patterns.clear(); + } + } + + @Override + public void pUnsubscribe(byte[]... patterns) { + if (ObjectUtils.isEmpty(patterns)) { + patterns = this.patterns.toArray(new byte[this.patterns.size()][]); + } + + synchronized (this.patterns) { + for (byte[] bs : patterns) { + this.patterns.remove(bs); + } + } + + subscriber.punsubscribe(RjcUtils.decodeMultiple(patterns)); + } + + @Override + public void subscribe(byte[]... channels) { + Assert.notEmpty(channels, "at least one channel required"); + + synchronized (this.channels) { + for (byte[] bs : channels) { + this.channels.add(bs); + } + } + + for (String channel : RjcUtils.decodeMultiple(channels)) { + subscriber.subscribe(channel, listenerAdapter); + } + } + + @Override + public void unsubscribe() { + unsubscribe(null); + + synchronized (patterns) { + patterns.clear(); + } + } + + @Override + public void unsubscribe(byte[]... channels) { + if (ObjectUtils.isEmpty(channels)) { + channels = this.channels.toArray(new byte[this.channels.size()][]); + } + + synchronized (this.channels) { + for (byte[] bs : channels) { + this.channels.remove(bs); + } + } + + subscriber.punsubscribe(RjcUtils.decodeMultiple(channels)); + } + + @Override + public boolean isAlive() { + return (!channels.isEmpty() || !patterns.isEmpty()); + } +} \ No newline at end of file From a1a562c7863153ae29d634d34d6600af6b034ab2 Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Wed, 16 Mar 2011 14:49:09 +0200 Subject: [PATCH 4/4] update value get/set operations --- .../DefaultStringRedisConnection.java | 6 ++--- .../connection/StringRedisConnection.java | 4 ++-- .../redis/core/BoundValueOperations.java | 4 ++-- .../core/DefaultBoundValueOperations.java | 6 ++--- .../redis/core/DefaultValueOperations.java | 7 +++--- .../keyvalue/redis/core/RedisTemplate.java | 23 +++++++++++++------ .../keyvalue/redis/core/ValueOperations.java | 4 ++-- 7 files changed, 32 insertions(+), 22 deletions(-) diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/DefaultStringRedisConnection.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/DefaultStringRedisConnection.java index 9db1d2af4..85799bca9 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/DefaultStringRedisConnection.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/DefaultStringRedisConnection.java @@ -683,7 +683,7 @@ public class DefaultStringRedisConnection implements StringRedisConnection { } @Override - public String getRange(String key, int start, int end) { + public String getRange(String key, long start, long end) { return deserialize(delegate.getRange(serialize(key), start, end)); } @@ -919,8 +919,8 @@ public class DefaultStringRedisConnection implements StringRedisConnection { } @Override - public void setRange(String key, int start, int end) { - delegate.setRange(serialize(key), start, end); + public void setRange(String key, long offset, String value) { + delegate.setRange(serialize(key), offset, serialize(value)); } @Override diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/StringRedisConnection.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/StringRedisConnection.java index 7622c3b56..53517c8ea 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/StringRedisConnection.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/StringRedisConnection.java @@ -95,9 +95,9 @@ public interface StringRedisConnection extends RedisConnection { Long append(String key, String value); - String getRange(String key, int start, int end); + String getRange(String key, long start, long end); - void setRange(String key, int start, int end); + void setRange(String key, long offset, String value); Boolean getBit(String key, long offset); diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/BoundValueOperations.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/BoundValueOperations.java index 6e0450465..18f0fd3a9 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/BoundValueOperations.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/BoundValueOperations.java @@ -40,9 +40,9 @@ public interface BoundValueOperations extends BoundKeyOperations { Integer append(String value); - String get(int start, int end); + String get(long start, long end); - void set(int start, int end); + void set(long offset, V value); Long size(); } diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/DefaultBoundValueOperations.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/DefaultBoundValueOperations.java index c808847d5..f8691ffec 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/DefaultBoundValueOperations.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/DefaultBoundValueOperations.java @@ -58,7 +58,7 @@ class DefaultBoundValueOperations extends DefaultBoundKeyOperations imp } @Override - public String get(int start, int end) { + public String get(long start, long end) { return ops.get(getKey(), start, end); } @@ -78,8 +78,8 @@ class DefaultBoundValueOperations extends DefaultBoundKeyOperations imp } @Override - public void set(int start, int end) { - ops.set(getKey(), start, end); + public void set(long offset, V value) { + ops.set(getKey(), offset, null); } @Override diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/DefaultValueOperations.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/DefaultValueOperations.java index 37172140a..3d1b6c5f4 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/DefaultValueOperations.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/DefaultValueOperations.java @@ -96,7 +96,7 @@ class DefaultValueOperations extends AbstractOperations implements V } @Override - public String get(K key, final int start, final int end) { + public String get(K key, final long start, final long end) { final byte[] rawKey = rawKey(key); byte[] rawReturn = execute(new RedisCallback() { @@ -217,13 +217,14 @@ class DefaultValueOperations extends AbstractOperations implements V @Override - public void set(K key, final int start, final int end) { + public void set(K key, final long offset, V value) { final byte[] rawKey = rawKey(key); + final byte[] rawValue = rawValue(value); execute(new RedisCallback() { @Override public Object doInRedis(RedisConnection connection) { - connection.setRange(rawKey, start, end); + connection.setRange(rawKey, offset, rawValue); return null; } }, true); diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisTemplate.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisTemplate.java index 6358593c5..e91bfc499 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisTemplate.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/RedisTemplate.java @@ -223,13 +223,22 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation return execute(new RedisCallback>() { public List doInRedis(RedisConnection connection) throws DataAccessException { connection.openPipeline(); - Object result = action.doInRedis(connection); - if (result != null) { - throw new InvalidDataAccessApiUsageException( - "Callback cannot returned a non-null value as it gets overwritten by the pipeline"); + boolean pipelinedClosed = false; + try { + Object result = action.doInRedis(connection); + if (result != null) { + throw new InvalidDataAccessApiUsageException( + "Callback cannot returned a non-null value as it gets overwritten by the pipeline"); + } + List pipeline = connection.closePipeline(); + pipelinedClosed = true; + return SerializationUtils.deserialize(pipeline, resultSerializer); + + } finally { + if (!pipelinedClosed) { + connection.closePipeline(); + } } - List pipeline = connection.closePipeline(); - return SerializationUtils.deserialize(pipeline, resultSerializer); } }); } @@ -377,7 +386,7 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation * Sets the string value serializer to be used by this template (when the arguments or return types * are always strings). Defaults to {@link StringRedisSerializer}. * - * @see ValueOperations#get(Object, int, int) + * @see ValueOperations#get(Object, long, long) * @param stringSerializer The stringValueSerializer to set. */ public void setStringSerializer(RedisSerializer stringSerializer) { diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/ValueOperations.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/ValueOperations.java index 3fd581ad0..479bebd5b 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/ValueOperations.java +++ b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/core/ValueOperations.java @@ -47,9 +47,9 @@ public interface ValueOperations { Integer append(K key, String value); - String get(K key, int start, int end); + String get(K key, long start, long end); - void set(K key, int start, int end); + void set(K key, long offset, V value); Long size(K key);