diff --git a/.classpath b/.classpath new file mode 100644 index 000000000..06c221d7f --- /dev/null +++ b/.classpath @@ -0,0 +1,43 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/.project b/.project new file mode 100644 index 000000000..862bb3d6d --- /dev/null +++ b/.project @@ -0,0 +1,16 @@ + + + spring-data-redis + Spring Data Redis + + + org.eclipse.jdt.core.javanature + + + + org.eclipse.jdt.core.javabuilder + + + + + diff --git a/.settings/org.eclipse.jdt.core.prefs b/.settings/org.eclipse.jdt.core.prefs new file mode 100644 index 000000000..8c9e1f31e --- /dev/null +++ b/.settings/org.eclipse.jdt.core.prefs @@ -0,0 +1,13 @@ +# +#Wed Apr 04 09:18:47 EEST 2012 +org.eclipse.jdt.core.compiler.debug.localVariable=generate +org.eclipse.jdt.core.compiler.compliance=1.5 +org.eclipse.jdt.core.compiler.codegen.unusedLocal=preserve +org.eclipse.jdt.core.compiler.debug.sourceFile=generate +org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.5 +org.eclipse.jdt.core.compiler.problem.enumIdentifier=error +org.eclipse.jdt.core.compiler.debug.lineNumber=generate +eclipse.preferences.version=1 +org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled +org.eclipse.jdt.core.compiler.source=1.5 +org.eclipse.jdt.core.compiler.problem.assertIdentifier=error diff --git a/build.gradle b/build.gradle index b561df415..7ce199f4a 100644 --- a/build.gradle +++ b/build.gradle @@ -87,6 +87,7 @@ dependencies { // Redis Drivers compile "redis.clients:jedis:$jedisVersion" + compile "com.github.spullara.redis:client:$srpVersion" compile("org.jredis:jredis-anthonylauzon:$jredisVersion") { optional = true } compile("org.idevlab:rjc:$rjcVersion") { optional = true } diff --git a/gradle.properties b/gradle.properties index 66da45e04..7ca1a04e2 100644 --- a/gradle.properties +++ b/gradle.properties @@ -15,7 +15,8 @@ mockitoVersion = 1.8.5 # Drivers jedisVersion = 2.1.0 jredisVersion = 03122010 -rjcVersion= 0.6.4 +rjcVersion = 0.6.4 +srpVersion = 0.2 # Manifest properties @@ -24,6 +25,7 @@ spring.range = "[3.1.0, 4.0.0)" jedis.range = "[2.1.0, 2.1.0]" jackson.range = "[1.6, 2.0.0)" rjc.range = "[0.6.4, 0.6.4]" +srp.range = "[0.2, 1.0)" # -------------------- # Project wide version diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java new file mode 100644 index 000000000..ae56e207b --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java @@ -0,0 +1,1824 @@ +/* + * Copyright 2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.redis.connection.srp; + +import java.io.IOException; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Set; +import java.util.concurrent.BlockingQueue; + +import org.springframework.dao.DataAccessException; +import org.springframework.data.redis.RedisConnectionFailureException; +import org.springframework.data.redis.RedisSystemException; +import org.springframework.data.redis.connection.DataType; +import org.springframework.data.redis.connection.MessageListener; +import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.connection.RedisSubscribedConnectionException; +import org.springframework.data.redis.connection.SortParameters; +import org.springframework.data.redis.connection.Subscription; +import org.springframework.util.Assert; + +import redis.Command; +import redis.client.RedisClient; +import redis.client.RedisClient.Pipeline; +import redis.client.RedisException; +import redis.reply.Reply; + +import com.google.common.base.Charsets; + +/** + * {@code RedisConnection} implementation on top of spullara Redis Protocol library. + * + * @author Costin Leau + */ +public class SrpConnection implements RedisConnection { + + private final RedisClient client; + private final BlockingQueue queue; + + private boolean isClosed = false; + private boolean isMulti = false; + private Pipeline pipeline; + private volatile SrpSubscription subscription; + + public SrpConnection(String host, int port, BlockingQueue queue) { + try { + this.client = new RedisClient(host, port); + this.queue = queue; + } catch (IOException e) { + throw new RedisConnectionFailureException("Could not connect", e); + } + } + + protected DataAccessException convertSRAccessException(Exception ex) { + if (ex instanceof RedisException) { + return SrpUtils.convertSRedisAccessException((RedisException) ex); + } + if (ex instanceof IOException) { + return new RedisConnectionFailureException("Redis connection failed", (IOException) ex); + } + + return new RedisSystemException("Unknown SRedis exception", ex); + } + + public Object execute(String command, byte[]... args) { + Assert.hasText(command, "a valid command needs to be specified"); + String name = command.trim().toUpperCase(); + Command cmd = new Command(name.getBytes(Charsets.UTF_8), args); + if (isPipelined()) { + client.pipeline(name, cmd); + return null; + } + else { + return client.execute(name, cmd); + } + } + + public void close() throws DataAccessException { + isClosed = true; + queue.remove(this); + + try { + client.close(); + } catch (IOException ex) { + throw convertSRAccessException(ex); + } + } + + public boolean isClosed() { + return isClosed; + } + + public RedisClient getNativeConnection() { + return client; + } + + + public boolean isQueueing() { + return isMulti; + } + + public boolean isPipelined() { + return (pipeline != null); + } + + + public void openPipeline() { + if (pipeline == null) { + pipeline = client.pipeline(); + } + } + + public List closePipeline() { + // if (pipeline != null) { + // //ListenableFuture reply = pipeline.exec(); + // pipeline = null; + // if (reply != null) { + // try { + // return SrpUtils.toList(reply.get().data()); + // } catch (Exception ex) { + // throw convertSRAccessException(ex); + // } + // } + // } + throw new UnsupportedOperationException(); + //return Collections.emptyList(); + } + + + public List sort(byte[] key, SortParameters params) { + + byte[] sort = SrpUtils.sort(params); + + try { + if (isPipelined()) { + pipeline.sort(key, sort, null, (Object[]) null); + return null; + } + return SrpUtils.toBytesList((Reply[]) client.sort(key, sort, null, (Object[]) null).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + public Long sort(byte[] key, SortParameters params, byte[] sortKey) { + + byte[] sort = SrpUtils.sort(params, sortKey); + + try { + if (isPipelined()) { + pipeline.sort(key, sort, null, (Object[]) null); + return null; + } + return ((Long) client.sort(key, sort, null, (Object[]) null).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + public Long dbSize() { + try { + if (isPipelined()) { + pipeline.dbsize(); + return null; + } + return client.dbsize().data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + + public void flushDb() { + try { + if (isPipelined()) { + pipeline.flushdb(); + return; + } + client.flushdb(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void flushAll() { + try { + if (isPipelined()) { + pipeline.flushall(); + return; + } + client.flushall(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void bgSave() { + try { + if (isPipelined()) { + pipeline.bgsave(); + return; + } + client.bgsave(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void bgWriteAof() { + try { + if (isPipelined()) { + pipeline.bgrewriteaof(); + return; + } + client.bgrewriteaof(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void save() { + try { + if (isPipelined()) { + pipeline.save(); + return; + } + client.save(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public List getConfig(String param) { + try { + if (isPipelined()) { + pipeline.config_get(param); + return null; + } + return Collections.singletonList(client.config_get(param).toString()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Properties info() { + try { + if (isPipelined()) { + pipeline.info(); + return null; + } + return SrpUtils.info(client.info()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long lastSave() { + try { + if (isPipelined()) { + pipeline.lastsave(); + return null; + } + return client.lastsave().data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void setConfig(String param, String value) { + try { + if (isPipelined()) { + pipeline.config_set(param, value); + return; + } + client.config_set(param, value); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + + public void resetConfigStats() { + try { + if (isPipelined()) { + pipeline.config_resetstat(); + return; + } + client.config_resetstat(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void shutdown() { + byte[] save = "SAVE".getBytes(Charsets.UTF_8); + try { + if (isPipelined()) { + pipeline.shutdown(save, null); + return; + } + client.shutdown(save, null); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] echo(byte[] message) { + try { + if (isPipelined()) { + pipeline.echo(message); + return null; + } + return client.echo(message).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public String ping() { + try { + if (isPipelined()) { + pipeline.ping(); + } + return client.ping().data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long del(byte[]... keys) { + try { + if (isPipelined()) { + pipeline.del((Object[]) keys); + return null; + } + return client.del((Object[]) keys).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void discard() { + isMulti = false; + try { + if (isPipelined()) { + client.discard(); + } + + client.discard(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public List exec() { + isMulti = false; + try { + // if (isPipelined()) { + return Collections.singletonList((Object) client.exec()); + // } + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean exists(byte[] key) { + try { + if (isPipelined()) { + pipeline.exists(key); + return null; + } + return client.exists(key).data() == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean expire(byte[] key, long seconds) { + try { + if (isPipelined()) { + pipeline.expire(key, seconds); + return null; + } + return client.expire(key, seconds).data() == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean expireAt(byte[] key, long unixTime) { + try { + if (isPipelined()) { + pipeline.expireat(key, unixTime); + return null; + } + return client.expireat(key, unixTime).data() == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set keys(byte[] pattern) { + try { + if (isPipelined()) { + pipeline.keys(pattern); + return null; + } + return SrpUtils.toSet(client.keys(pattern).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void multi() { + if (isQueueing()) { + return; + } + isMulti = true; + openPipeline(); + try { + if (isPipelined()) { + client.multi(); + return; + } + client.multi(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean persist(byte[] key) { + try { + if (isPipelined()) { + pipeline.persist(key); + return null; + } + return client.persist(key).data() == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean move(byte[] key, int dbIndex) { + try { + if (isPipelined()) { + pipeline.move(key, dbIndex); + return null; + } + return client.move(key, dbIndex).data() == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] randomKey() { + try { + if (isPipelined()) { + pipeline.randomkey(); + return null; + } + return client.randomkey().data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void rename(byte[] oldName, byte[] newName) { + try { + if (isPipelined()) { + pipeline.rename(oldName, newName); + return; + } + client.rename(oldName, newName); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean renameNX(byte[] oldName, byte[] newName) { + try { + if (isPipelined()) { + pipeline.renamenx(oldName, newName); + return null; + } + return (client.renamenx(oldName, newName).data() == 1); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void select(int dbIndex) { + try { + if (isPipelined()) { + throw new UnsupportedOperationException(); + } + client.select(dbIndex); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long ttl(byte[] key) { + try { + if (isPipelined()) { + pipeline.ttl(key); + return null; + } + return client.ttl(key).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public DataType type(byte[] key) { + try { + if (isPipelined()) { + pipeline.type(key); + return null; + } + return DataType.fromCode(client.type(key).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void unwatch() { + try { + client.unwatch(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void watch(byte[]... keys) { + if (isQueueing()) { + throw new UnsupportedOperationException(); + } + try { + if (isPipelined()) { + pipeline.watch((Object[]) keys); + } + else { + client.watch((Object[]) keys); + } + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + // + // String commands + // + + + public byte[] get(byte[] key) { + try { + if (isPipelined()) { + pipeline.get(key); + return null; + } + + return client.get(key).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void set(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.set(key, value); + return; + } + client.set(key, value); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + + public byte[] getSet(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.getset(key, value); + return null; + } + return client.getset(key, value).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long append(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.append(key, value); + return null; + } + return client.append(key, value).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public List mGet(byte[]... keys) { + try { + if (isPipelined()) { + pipeline.mget((Object[]) keys); + return null; + } + return SrpUtils.toBytesList(client.mget((Object[]) keys).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void mSet(Map tuples) { + try { + if (isPipelined()) { + pipeline.mset((Object[]) SrpUtils.convert(tuples)); + return; + } + client.mset((Object[]) SrpUtils.convert(tuples)); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void mSetNX(Map tuples) { + try { + if (isPipelined()) { + pipeline.msetnx((Object[]) SrpUtils.convert(tuples)); + return; + } + client.msetnx((Object[]) SrpUtils.convert(tuples)); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void setEx(byte[] key, long time, byte[] value) { + try { + if (isPipelined()) { + pipeline.setex(key, time, value); + return; + } + client.setex(key, time, value); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean setNX(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.setnx(key, value); + return null; + } + return client.setnx(key, value).data() == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] getRange(byte[] key, long start, long end) { + try { + if (isPipelined()) { + pipeline.getrange(key, start, end); + return null; + } + return client.getrange(key, start, end).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long decr(byte[] key) { + try { + if (isPipelined()) { + pipeline.decr(key); + return null; + } + return client.decr(key).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long decrBy(byte[] key, long value) { + try { + if (isPipelined()) { + pipeline.decrby(key, value); + return null; + } + return client.decrby(key, value).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long incr(byte[] key) { + try { + if (isPipelined()) { + pipeline.incr(key); + return null; + } + return client.incr(key).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long incrBy(byte[] key, long value) { + try { + if (isPipelined()) { + pipeline.incrby(key, value); + return null; + } + return client.incrby(key, value).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean getBit(byte[] key, long offset) { + try { + if (isQueueing()) { + throw new UnsupportedOperationException(); + } + if (isPipelined()) { + throw new UnsupportedOperationException(); + } + return (client.getbit(key, offset).data() == 1); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void setBit(byte[] key, long offset, boolean value) { + try { + if (isQueueing()) { + throw new UnsupportedOperationException(); + } + if (isPipelined()) { + throw new UnsupportedOperationException(); + } + client.setbit(key, offset, SrpUtils.asBit(value)); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void setRange(byte[] key, byte[] value, long start) { + try { + if (isQueueing()) { + throw new UnsupportedOperationException(); + } + if (isPipelined()) { + throw new UnsupportedOperationException(); + } + client.setrange(key, start, value); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long strLen(byte[] key) { + try { + if (isPipelined()) { + pipeline.strlen(key); + return null; + } + return client.strlen(key).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + // + // List commands + // + + + public Long lPush(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.lpush(key, new Object[] { value }); + return null; + } + return client.lpush(key, new Object[] { value }).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long rPush(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.rpush(key, new Object[] { value }); + return null; + } + return client.rpush(key, new Object[] { value }).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public List bLPop(int timeout, byte[]... keys) { + try { + if (isPipelined()) { + // pipeline.blpop(timeout, keys); + return null; + } + // return SrpUtils.toBytesList(client.blpop(timeout, keys).data()); + throw new UnsupportedOperationException(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public List bRPop(int timeout, byte[]... keys) { + try { + if (isPipelined()) { + // pipeline.brpop(timeout, keys); + return null; + } + // return SrpUtils.toBytesList(client.brpop(timeout, keys).data()); + throw new UnsupportedOperationException(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] lIndex(byte[] key, long index) { + try { + if (isPipelined()) { + pipeline.lindex(key, index); + return null; + } + return client.lindex(key, index).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long lInsert(byte[] key, Position where, byte[] pivot, byte[] value) { + try { + if (isPipelined()) { + pipeline.linsert(key, SrpUtils.convertPosition(where), pivot, value); + return null; + } + return client.linsert(key, SrpUtils.convertPosition(where), pivot, value).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long lLen(byte[] key) { + try { + if (isPipelined()) { + pipeline.llen(key); + return null; + } + return client.llen(key).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] lPop(byte[] key) { + try { + if (isPipelined()) { + pipeline.lpop(key); + return null; + } + return client.lpop(key).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public List lRange(byte[] key, long start, long end) { + try { + if (isPipelined()) { + pipeline.lrange(key, start, end); + return null; + } + return SrpUtils.toBytesList(client.lrange(key, start, end).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long lRem(byte[] key, long count, byte[] value) { + try { + if (isPipelined()) { + pipeline.lrem(key, count, value); + return null; + } + return client.lrem(key, count, value).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void lSet(byte[] key, long index, byte[] value) { + try { + if (isPipelined()) { + pipeline.lset(key, index, value); + return; + } + client.lset(key, index, value); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void lTrim(byte[] key, long start, long end) { + try { + if (isPipelined()) { + pipeline.ltrim(key, start, end); + return; + } + client.ltrim(key, start, end); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] rPop(byte[] key) { + try { + if (isPipelined()) { + pipeline.rpop(key); + return null; + } + return client.rpop(key).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] rPopLPush(byte[] srcKey, byte[] dstKey) { + try { + if (isPipelined()) { + pipeline.rpoplpush(srcKey, dstKey); + return null; + } + return client.rpoplpush(srcKey, dstKey).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] bRPopLPush(int timeout, byte[] srcKey, byte[] dstKey) { + try { + if (isPipelined()) { + pipeline.brpoplpush(srcKey, dstKey, timeout); + return null; + } + return client.brpoplpush(srcKey, dstKey, timeout).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long lPushX(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.lpushx(key, value); + return null; + } + return client.lpushx(key, value).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long rPushX(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.rpushx(key, value); + return null; + } + return client.rpushx(key, value).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + // + // Set commands + // + + + public Boolean sAdd(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.sadd(key, new Object[] { value }); + return null; + } + return (client.sadd(key, new Object[] { value }).data() == 1); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long sCard(byte[] key) { + try { + if (isPipelined()) { + pipeline.scard(key); + return null; + } + return client.scard(key).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set sDiff(byte[]... keys) { + try { + if (isPipelined()) { + pipeline.sdiff((Object[]) keys); + return null; + } + return SrpUtils.toSet(client.sdiff((Object[]) keys).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long sDiffStore(byte[] destKey, byte[]... keys) { + try { + if (isPipelined()) { + pipeline.sdiffstore(destKey, (Object[]) keys); + return null; + } + return client.sdiffstore(destKey, (Object[]) keys).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set sInter(byte[]... keys) { + try { + if (isPipelined()) { + pipeline.sinter((Object[]) keys); + return null; + } + return SrpUtils.toSet(client.sinter((Object[]) keys).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long sInterStore(byte[] destKey, byte[]... keys) { + try { + if (isPipelined()) { + pipeline.sinterstore(destKey, (Object[]) keys); + return null; + } + return client.sinterstore(destKey, (Object[]) keys).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean sIsMember(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.sismember(key, value); + return null; + } + return client.sismember(key, value).data() == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set sMembers(byte[] key) { + try { + if (isPipelined()) { + pipeline.smembers(key); + return null; + } + return SrpUtils.toSet(client.smembers(key).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean sMove(byte[] srcKey, byte[] destKey, byte[] value) { + try { + if (isPipelined()) { + pipeline.smove(srcKey, destKey, value); + return null; + } + return client.smove(srcKey, destKey, value).data() == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] sPop(byte[] key) { + try { + if (isPipelined()) { + pipeline.spop(key); + return null; + } + return client.spop(key).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] sRandMember(byte[] key) { + try { + if (isPipelined()) { + pipeline.srandmember(key); + return null; + } + return client.srandmember(key).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean sRem(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.srem(key, new Object[] { value }); + return null; + } + return client.srem(key, new Object[] { value }).data() == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set sUnion(byte[]... keys) { + try { + if (isPipelined()) { + pipeline.sunion((Object[]) keys); + return null; + } + return SrpUtils.toSet(client.sunion((Object[]) keys).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long sUnionStore(byte[] destKey, byte[]... keys) { + try { + if (isPipelined()) { + pipeline.sunionstore(destKey, (Object[]) keys); + return null; + } + return client.sunionstore(destKey, (Object[]) keys).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + // + // ZSet commands + // + + + public Boolean zAdd(byte[] key, double score, byte[] value) { + try { + if (isPipelined()) { + pipeline.zadd(new Object[] { key, score, value }); + return null; + } + return client.zadd(new Object[] { key, score, value }).data() == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long zCard(byte[] key) { + try { + if (isPipelined()) { + pipeline.zcard(key); + return null; + } + return client.zcard(key).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long zCount(byte[] key, double min, double max) { + try { + if (isQueueing()) { + pipeline.zcount(key, min, max); + return null; + } + return client.zcount(key, min, max).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Double zIncrBy(byte[] key, double increment, byte[] value) { + try { + if (isPipelined()) { + pipeline.zincrby(key, increment, value); + return null; + } + return SrpUtils.toDouble(client.zincrby(key, increment, value).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long zInterStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets) { + throw new UnsupportedOperationException(); + } + + + public Long zInterStore(byte[] destKey, byte[]... sets) { + + Object[] args = new Object[2 + sets.length]; + + args[0] = destKey; + args[1] = sets.length; + int i = 2; + for (byte[] set : sets) { + args[i++] = set; + } + + try { + if (isQueueing()) { + pipeline.zinterstore(args); + return null; + } + return client.zinterstore(args).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + public Set zRange(byte[] key, long start, long end) { + try { + if (isPipelined()) { + pipeline.zrange(key, start, end, null); + return null; + } + return SrpUtils.toSet(client.zrange(key, start, end, null).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRangeWithScores(byte[] key, long start, long end) { + try { + if (isPipelined()) { + pipeline.zrange(key, start, end, SrpUtils.WITHSCORES); + return null; + } + return SrpUtils.convertTuple(client.zrange(key, start, end, SrpUtils.WITHSCORES)); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRangeByScore(byte[] key, double min, double max) { + try { + if (isPipelined()) { + pipeline.zrangebyscore(key, min, max, null, null); + return null; + } + return SrpUtils.toSet(client.zrangebyscore(key, min, max, null, null).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRangeByScoreWithScores(byte[] key, double min, double max) { + try { + if (isPipelined()) { + pipeline.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, null); + return null; + } + return SrpUtils.convertTuple(client.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, null)); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRevRangeWithScores(byte[] key, long start, long end) { + try { + if (isPipelined()) { + pipeline.zrevrange(key, start, end, SrpUtils.WITHSCORES); + return null; + } + return SrpUtils.convertTuple(client.zrevrange(key, start, end, SrpUtils.WITHSCORES)); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRangeByScore(byte[] key, double min, double max, long offset, long count) { + try { + byte[] limit = SrpUtils.limit(offset, count); + if (isPipelined()) { + pipeline.zrangebyscore(key, min, max, null, limit); + return null; + } + return SrpUtils.toSet(client.zrangebyscore(key, min, max, null, limit).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRangeByScoreWithScores(byte[] key, double min, double max, long offset, long count) { + try { + byte[] limit = SrpUtils.limit(offset, count); + if (isPipelined()) { + pipeline.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, limit); + return null; + } + return SrpUtils.convertTuple(client.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, limit)); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRevRangeByScore(byte[] key, double min, double max, long offset, long count) { + try { + byte[] limit = SrpUtils.limit(offset, count); + if (isPipelined()) { + client.zrevrangebyscore(key, min, max, null, limit); + } + return SrpUtils.toSet(client.zrevrangebyscore(key, min, max, null, limit).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRevRangeByScore(byte[] key, double min, double max) { + try { + if (isPipelined()) { + client.zrevrangebyscore(key, min, max, null, null); + } + return SrpUtils.toSet(client.zrevrangebyscore(key, min, max, null, null).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRevRangeByScoreWithScores(byte[] key, double min, double max, long offset, long count) { + try { + byte[] limit = SrpUtils.limit(offset, count); + if (isPipelined()) { + client.zrevrangebyscore(key, min, max, SrpUtils.WITHSCORES, limit); + } + return SrpUtils.convertTuple(client.zrevrangebyscore(key, min, max, SrpUtils.WITHSCORES, limit)); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRevRangeByScoreWithScores(byte[] key, double min, double max) { + try { + if (isPipelined()) { + client.zrevrangebyscore(key, min, max, SrpUtils.WITHSCORES, null); + } + return SrpUtils.convertTuple(client.zrevrangebyscore(key, min, max, SrpUtils.WITHSCORES, null)); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long zRank(byte[] key, byte[] value) { + try { + if (isPipelined()) { + // (Long) pipeline .zrank(key, value).data; + // return null; + throw new UnsupportedOperationException(); + } + return (Long) client.zrank(key, value).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean zRem(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.zrem(key, new Object[] { value }); + return null; + } + return client.zrem(key, new Object[] { value }).data() == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long zRemRange(byte[] key, long start, long end) { + try { + if (isPipelined()) { + pipeline.zremrangebyrank(key, start, end); + return null; + } + return client.zremrangebyrank(key, start, end).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long zRemRangeByScore(byte[] key, double min, double max) { + try { + if (isPipelined()) { + pipeline.zremrangebyscore(key, min, max); + return null; + } + return client.zremrangebyscore(key, min, max).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set zRevRange(byte[] key, long start, long end) { + try { + if (isPipelined()) { + pipeline.zrevrange(key, start, end, null); + return null; + } + return SrpUtils.toSet(client.zrevrange(key, start, end, null).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long zRevRank(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.zrevrank(key, value); + return null; + } + return (Long) client.zrevrank(key, value).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Double zScore(byte[] key, byte[] value) { + try { + if (isPipelined()) { + pipeline.zscore(key, value); + return null; + } + return SrpUtils.toDouble(client.zscore(key, value).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long zUnionStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets) { + throw new UnsupportedOperationException(); + } + + + public Long zUnionStore(byte[] destKey, byte[]... sets) { + try { + if (isPipelined()) { + pipeline.zunionstore(destKey, sets.length, (Object[]) sets); + return null; + } + return client.zunionstore(destKey, sets.length, (Object[]) sets).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + // + // Hash commands + // + + + public Boolean hSet(byte[] key, byte[] field, byte[] value) { + try { + if (isPipelined()) { + pipeline.hset(key, field, value); + return null; + } + return client.hset(key, field, value).data() == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean hSetNX(byte[] key, byte[] field, byte[] value) { + try { + if (isPipelined()) { + pipeline.hsetnx(key, field, value); + return null; + } + return client.hsetnx(key, field, value).data() == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean hDel(byte[] key, byte[] field) { + try { + if (isPipelined()) { + pipeline.hdel(key, new Object[] { field }); + return null; + } + return client.hdel(key, new Object[] { field }).data() == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Boolean hExists(byte[] key, byte[] field) { + try { + if (isPipelined()) { + pipeline.hexists(key, field); + return null; + } + return client.hexists(key, field).data() == 1; + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public byte[] hGet(byte[] key, byte[] field) { + try { + if (isPipelined()) { + pipeline.hget(key, field); + return null; + } + return client.hget(key, field).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Map hGetAll(byte[] key) { + try { + if (isPipelined()) { + pipeline.hgetall(key); + return null; + } + return SrpUtils.toMap(client.hgetall(key).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long hIncrBy(byte[] key, byte[] field, long delta) { + try { + if (isPipelined()) { + pipeline.hincrby(key, field, delta); + return null; + } + return client.hincrby(key, field, delta).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Set hKeys(byte[] key) { + try { + if (isPipelined()) { + pipeline.hkeys(key); + return null; + } + return SrpUtils.toSet(client.hkeys(key).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Long hLen(byte[] key) { + try { + if (isPipelined()) { + pipeline.hlen(key); + return null; + } + return client.hlen(key).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public List hMGet(byte[] key, byte[]... fields) { + try { + if (isPipelined()) { + pipeline.hmget(key, (Object[]) fields); + return null; + } + return SrpUtils.toBytesList(client.hmget(key, (Object[]) fields).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void hMSet(byte[] key, Map tuple) { + try { + if (isPipelined()) { + pipeline.hmset(key, SrpUtils.convert(tuple)); + return; + } + client.hmset(key, SrpUtils.convert(tuple)); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public List hVals(byte[] key) { + try { + if (isPipelined()) { + pipeline.hvals(key); + return null; + } + return SrpUtils.toBytesList(client.hvals(key).data()); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + // + // Pub/Sub functionality + // + + public Long publish(byte[] channel, byte[] message) { + try { + if (isQueueing()) { + throw new UnsupportedOperationException(); + } + if (isPipelined()) { + pipeline.publish(channel, message); + return null; + } + return client.publish(channel, message).data(); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public Subscription getSubscription() { + return subscription; + } + + + public boolean isSubscribed() { + return (subscription != null && subscription.isAlive()); + } + + + public void pSubscribe(MessageListener listener, byte[]... patterns) { + checkSubscription(); + + try { + if (isQueueing()) { + throw new UnsupportedOperationException(); + } + if (isPipelined()) { + throw new UnsupportedOperationException(); + } + + subscription = new SrpSubscription(listener, client); + subscription.pSubscribe(patterns); + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + + public void subscribe(MessageListener listener, byte[]... channels) { + checkSubscription(); + + try { + if (isPipelined()) { + throw new UnsupportedOperationException(); + } + + subscription = new SrpSubscription(listener, client); + subscription.subscribe(channels); + + } catch (Exception ex) { + throw convertSRAccessException(ex); + } + } + + private void checkSubscription() { + if (isSubscribed()) { + throw new RedisSubscribedConnectionException( + "Connection already subscribed; use the connection Subscription to cancel or add new channels"); + } + } +} \ No newline at end of file diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpConnectionFactory.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnectionFactory.java new file mode 100644 index 000000000..edd0d0879 --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnectionFactory.java @@ -0,0 +1,121 @@ +/* + * Copyright 2011-2012 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.data.redis.connection.srp; + +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.dao.DataAccessException; +import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; + +/** + * Connection factory creating Redis Protocol based connections. + * + * @author Costin Leau + */ +public class SrpConnectionFactory implements InitializingBean, DisposableBean, RedisConnectionFactory { + + private final static Log log = LogFactory.getLog(JedisConnectionFactory.class); + + private String hostName = "localhost"; + private int port = 6379; + private BlockingQueue trackedConnections = new ArrayBlockingQueue(50); + + + /** + * Constructs a new SRedisConnectionFactory instance + * with default settings. + */ + public SrpConnectionFactory() { + } + + /** + * Constructs a new SRedisConnectionFactory instance + * with default settings. + */ + public SrpConnectionFactory(String host, int port) { + this.hostName = host; + this.port = port; + } + + public void afterPropertiesSet() { + } + + public void destroy() { + SrpConnection con; + do { + con = trackedConnections.poll(); + if (con != null && !con.isClosed()) { + try { + con.close(); + } catch (Exception ex) { + // ignore + } + } + } while (con != null); + } + + public RedisConnection getConnection() { + return new SrpConnection(hostName, port, trackedConnections); + } + + public DataAccessException translateExceptionIfPossible(RuntimeException ex) { + return SrpUtils.convertSRedisAccessException(ex); + } + + /** + * Returns the current host. + * + * @return the host + */ + public String getHostName() { + return hostName; + } + + /** + * Sets the host. + * + * @param host the host to set + */ + public void setHostName(String host) { + this.hostName = host; + } + + /** + * Returns the current port. + * + * @return the port + */ + public int getPort() { + return port; + } + + /** + * Sets the port. + * + * @param port the port to set + */ + public void setPort(int port) { + this.port = port; + } +} \ No newline at end of file diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpMessageListener.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpMessageListener.java new file mode 100644 index 000000000..0534ceb81 --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpMessageListener.java @@ -0,0 +1,58 @@ +/* + * Copyright 2011-2012 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.data.redis.connection.srp; + +import org.springframework.data.redis.connection.DefaultMessage; +import org.springframework.data.redis.connection.MessageListener; +import org.springframework.util.Assert; + +import redis.client.ReplyListener; + +/** + * MessageListener wrapper around SRP {@link ReplyListener}. + * + * @author Costin Leau + */ +class SrpMessageListener implements ReplyListener { + + private final MessageListener listener; + + SrpMessageListener(MessageListener listener) { + Assert.notNull(listener, "message listener is required"); + this.listener = listener; + } + + public void message(byte[] channel, byte[] message) { + listener.onMessage(new DefaultMessage(channel, message), null); + } + + public void pmessage(byte[] pattern, byte[] channel, byte[] message) { + listener.onMessage(new DefaultMessage(channel, message), pattern); + } + + public void psubscribed(byte[] arg0, int arg1) { + } + + public void punsubscribed(byte[] arg0, int arg1) { + } + + public void subscribed(byte[] arg0, int arg1) { + } + + public void unsubscribed(byte[] arg0, int arg1) { + } +} diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpSubscription.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpSubscription.java new file mode 100644 index 000000000..473122936 --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpSubscription.java @@ -0,0 +1,74 @@ +/* + * Copyright 2011-2012 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.data.redis.connection.srp; + +import org.springframework.data.redis.connection.MessageListener; +import org.springframework.data.redis.connection.util.AbstractSubscription; + +import redis.client.RedisClient; +import redis.client.ReplyListener; + +/** + * Message subscription on top of SRP. + * + * @author Costin Leau + */ +class SrpSubscription extends AbstractSubscription { + + private final RedisClient client; + private final ReplyListener listener; + + SrpSubscription(MessageListener listener, RedisClient client) { + super(listener); + this.client = client; + this.listener = new SrpMessageListener(listener); + client.addListener(this.listener); + } + + protected void doClose() { + client.unsubscribe((Object[]) null); + client.punsubscribe((Object[]) null); + client.removeListener(this.listener); + } + + + protected void doPsubscribe(byte[]... patterns) { + client.psubscribe((Object[]) patterns); + } + + protected void doPUnsubscribe(boolean all, byte[]... patterns) { + if (all) { + client.punsubscribe((Object[]) null); + } + else { + client.punsubscribe((Object[]) patterns); + } + } + + protected void doSubscribe(byte[]... channels) { + client.subscribe((Object[]) channels); + } + + protected void doUnsubscribe(boolean all, byte[]... channels) { + if (all) { + client.unsubscribe((Object[]) null); + } + else { + client.unsubscribe((Object[]) channels); + } + } +} \ No newline at end of file diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpUtils.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpUtils.java new file mode 100644 index 000000000..808f33812 --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpUtils.java @@ -0,0 +1,219 @@ +/* + * Copyright 2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.data.redis.connection.srp; + +import java.io.StringReader; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Set; + +import org.springframework.dao.DataAccessException; +import org.springframework.data.redis.RedisSystemException; +import org.springframework.data.redis.connection.DefaultTuple; +import org.springframework.data.redis.connection.RedisListCommands.Position; +import org.springframework.data.redis.connection.RedisZSetCommands.Tuple; +import org.springframework.data.redis.connection.SortParameters; +import org.springframework.util.Assert; + +import redis.client.RedisException; +import redis.reply.BulkReply; +import redis.reply.MultiBulkReply; +import redis.reply.Reply; + +import com.google.common.base.Charsets; + +/** + * Helper class featuring methods for SRedis connection handling, providing support for exception translation. + * + * @author Costin Leau + */ +abstract class SrpUtils { + + private static final byte[] ONE = new byte[] { '1' }; + private static final byte[] ZERO = new byte[] { '0' }; + private static final byte[] BEFORE = "BEFORE".getBytes(Charsets.UTF_8); + private static final byte[] AFTER = "AFTER".getBytes(Charsets.UTF_8); + static final byte[] WITHSCORES = "WITHSCORES".getBytes(Charsets.UTF_8); + private static final byte[] SPACE = "".getBytes(Charsets.UTF_8); + private static final byte[] BY = "BY ".getBytes(Charsets.UTF_8); + private static final byte[] GET = "GET ".getBytes(Charsets.UTF_8); + private static final byte[] ALPHA = "ALPHA ".getBytes(Charsets.UTF_8); + private static final byte[] STORE = "STORE ".getBytes(Charsets.UTF_8); + + + static DataAccessException convertSRedisAccessException(RuntimeException ex) { + if (ex instanceof RedisException) { + return new RedisSystemException("redis exception", ex); + } + return null; + } + + static Properties info(BulkReply reply) { + Properties info = new Properties(); + // use the same charset as the library + StringReader stringReader = new StringReader(new String(reply.data(), Charsets.UTF_8)); + try { + info.load(stringReader); + } catch (Exception ex) { + throw new RedisSystemException("Cannot read Redis info", ex); + } finally { + stringReader.close(); + } + return info; + } + + static List toBytesList(Reply[] replies) { + List list = new ArrayList(replies.length); + for (Reply reply : replies) { + Object data = reply.data(); + if (data == null) { + list.add(null); + } + else if (data instanceof byte[]) + list.add((byte[]) data); + else + throw new IllegalArgumentException("array contains more then just nulls and bytes -> " + data); + } + + return list; + } + + static List toList(T[] byteArrays) { + return Arrays.asList(byteArrays); + } + + static Set toSet(Reply[] byteArrays) { + return new LinkedHashSet(toBytesList(byteArrays)); + } + + static byte[][] convert(Map hgetAll) { + byte[][] result = new byte[hgetAll.size() * 2][]; + + int index = 0; + for (Map.Entry entry : hgetAll.entrySet()) { + result[index++] = entry.getKey(); + result[index++] = entry.getValue(); + } + return result; + } + + static byte[] asBit(boolean value) { + return (value ? ONE : ZERO); + } + + static byte[] convertPosition(Position where) { + Assert.notNull("list positions are mandatory"); + return (Position.AFTER.equals(where) ? AFTER : BEFORE); + } + + static Double toDouble(byte[] bytes) { + return (bytes == null || bytes.length == 0 ? null : Double.valueOf(new String(bytes, Charsets.UTF_8))); + } + + static Long toLong(Object[] bytes) { + return (bytes == null || bytes.length == 0 ? null : Long.valueOf(new String((byte[]) bytes[0], Charsets.UTF_8))); + } + + static Set convertTuple(MultiBulkReply zrange) { + Reply[] byteArrays = zrange.data(); + Set tuples = new LinkedHashSet(byteArrays.length / 2 + 1); + + for (int i = 0; i < byteArrays.length; i++) { + byte[] value = (byte[]) byteArrays[i].data(); + i++; + Double score = toDouble((byte[]) byteArrays[i].data()); + tuples.add(new DefaultTuple(value, score)); + } + + return tuples; + } + + static Map toMap(Object[] byteArrays) { + Map map = new LinkedHashMap(byteArrays.length / 2); + for (int i = 0; i < byteArrays.length; i++) { + map.put((byte[]) byteArrays[i++], (byte[]) byteArrays[i]); + } + return map; + } + + static byte[] limit(long offset, long count) { + return ("LIMIT " + offset + " " + count).getBytes(Charsets.UTF_8); + } + + static byte[] sort(SortParameters params) { + return sort(params, null); + } + + static byte[] sort(SortParameters params, byte[] sortKey) { + List arrays = new ArrayList(); + + if (params.getByPattern() != null) { + arrays.add(BY); + arrays.add(params.getByPattern()); + arrays.add(SPACE); + } + + if (params.getLimit() != null) { + arrays.add(limit(params.getLimit().getStart(), params.getLimit().getCount())); + arrays.add(SPACE); + } + + if (params.getGetPattern() != null) { + byte[][] pattern = params.getGetPattern(); + for (byte[] bs : pattern) { + arrays.add(GET); + arrays.add(bs); + arrays.add(SPACE); + } + } + + if (params.getOrder() != null) { + arrays.add(params.getOrder().name().getBytes(Charsets.UTF_8)); + arrays.add(SPACE); + } + + if (params.isAlphabetic()) { + arrays.add(ALPHA); + } + + if (sortKey != null) { + arrays.add(STORE); + arrays.add(sortKey); + } + + // concatenate array + int size = 0; + + for (byte[] bs : arrays) { + size += bs.length; + } + byte[] result = new byte[size]; + + int index = 0; + for (byte[] bs : arrays) { + System.arraycopy(bs, 0, result, index, bs.length); + index += bs.length; + } + + return result; + } +} \ No newline at end of file diff --git a/src/main/java/org/springframework/data/redis/connection/srp/package-info.java b/src/main/java/org/springframework/data/redis/connection/srp/package-info.java new file mode 100644 index 000000000..d3b6d524f --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/srp/package-info.java @@ -0,0 +1,5 @@ +/** + * Connection package for spullara Redis Protocol library. + */ +package org.springframework.data.redis.connection.srp; + diff --git a/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionIntegrationTests.java new file mode 100644 index 000000000..b6fc07350 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionIntegrationTests.java @@ -0,0 +1,57 @@ +/* + * Copyright 2011-2012 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.data.redis.connection.srp; + +import org.junit.Test; +import org.springframework.data.redis.SettingsUtils; +import org.springframework.data.redis.connection.AbstractConnectionIntegrationTests; +import org.springframework.data.redis.connection.RedisConnectionFactory; + +import redis.client.RedisClient; + +/** + * @author Costin Leau + */ +public class SrpConnectionIntegrationTests extends AbstractConnectionIntegrationTests { + SrpConnectionFactory factory; + + public SrpConnectionIntegrationTests() { + factory = new SrpConnectionFactory(); + factory.setPort(SettingsUtils.getPort()); + factory.setHostName(SettingsUtils.getHost()); + + factory.afterPropertiesSet(); + } + + + protected RedisConnectionFactory getConnectionFactory() { + return factory; + } + + @Test + public void testRaw() throws Exception { + RedisClient rc = (RedisClient) factory.getConnection().getNativeConnection(); + + System.out.println(rc.dbsize()); + System.out.println(rc.exists("foobar")); + rc.set("foobar", "barfoo"); + System.out.println(rc.get("foobar")); + } + + public void testNullCollections() throws Exception { + } +} diff --git a/src/test/java/org/springframework/data/redis/support/collections/CollectionTestParams.java b/src/test/java/org/springframework/data/redis/support/collections/CollectionTestParams.java index 0108cda85..16839b639 100644 --- a/src/test/java/org/springframework/data/redis/support/collections/CollectionTestParams.java +++ b/src/test/java/org/springframework/data/redis/support/collections/CollectionTestParams.java @@ -23,6 +23,7 @@ import org.springframework.data.redis.SettingsUtils; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; import org.springframework.data.redis.connection.jredis.JredisConnectionFactory; import org.springframework.data.redis.connection.rjc.RjcConnectionFactory; +import org.springframework.data.redis.connection.srp.SrpConnectionFactory; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.serializer.JacksonJsonRedisSerializer; @@ -135,6 +136,32 @@ public abstract class CollectionTestParams { jsonPersonTemplateRJC.setConnectionFactory(rjcConnFactory); jsonPersonTemplateRJC.afterPropertiesSet(); + // SRP + SrpConnectionFactory srConnFactory = new SrpConnectionFactory(); + srConnFactory.setPort(SettingsUtils.getPort()); + srConnFactory.setHostName(SettingsUtils.getHost()); + srConnFactory.afterPropertiesSet(); + + RedisTemplate stringTemplateSRP = new StringRedisTemplate(srConnFactory); + RedisTemplate personTemplateSRP = new RedisTemplate(); + personTemplateSRP.setConnectionFactory(srConnFactory); + personTemplateSRP.afterPropertiesSet(); + + RedisTemplate xstreamStringTemplateSRP = new RedisTemplate(); + xstreamStringTemplateSRP.setConnectionFactory(srConnFactory); + xstreamStringTemplateSRP.setDefaultSerializer(serializer); + xstreamStringTemplateSRP.afterPropertiesSet(); + + RedisTemplate xstreamPersonTemplateSRP = new RedisTemplate(); + xstreamPersonTemplateSRP.setValueSerializer(serializer); + xstreamPersonTemplateSRP.setConnectionFactory(srConnFactory); + xstreamPersonTemplateSRP.afterPropertiesSet(); + + RedisTemplate jsonPersonTemplateSRP = new RedisTemplate(); + jsonPersonTemplateSRP.setValueSerializer(jsonSerializer); + jsonPersonTemplateSRP.setConnectionFactory(srConnFactory); + jsonPersonTemplateSRP.afterPropertiesSet(); + return Arrays.asList(new Object[][] { { stringFactory, stringTemplate }, { stringFactory, stringTemplateRJC }, { personFactory, personTemplateRJC }, //{ stringFactory, stringTemplateJR }, @@ -146,6 +173,10 @@ public abstract class CollectionTestParams { { personFactory, jsonPersonTemplate }, //{ personFactory, jsonPersonTemplateJR }, { stringFactory, xstreamStringTemplateRJC }, { personFactory, xstreamPersonTemplateRJC }, - { personFactory, jsonPersonTemplateRJC } }); + { personFactory, jsonPersonTemplateRJC }, + { stringFactory, stringTemplateSRP },{ personFactory, personTemplateSRP }, + { stringFactory, xstreamStringTemplateSRP }, { personFactory, xstreamPersonTemplateSRP }, + { personFactory, jsonPersonTemplateSRP } + }); } } diff --git a/template.mf b/template.mf index ed33cbcaf..182e1c84c 100644 --- a/template.mf +++ b/template.mf @@ -26,4 +26,7 @@ Import-Template: org.idevlab.rjc.*;resolution:="optional";version=${rjc.range}, org.apache.commons.pool.impl.*;resolution:="optional";version="[1.0.0, 3.0.0)", org.codehaus.jackson.*;resolution:="optional";version=${jackson.range}, - org.apache.commons.beanutils.*;resolution:="optional";version=1.8.5 \ No newline at end of file + org.apache.commons.beanutils.*;resolution:="optional";version=1.8.5, + redis.client.*;resolution:="optional";version=${srp.range}, + redis.reply.*;resolution:="optional";version=${srp.range}, + com.google.common.*;resolution:="optional";version="[11.0.0, 20.0.0)" \ No newline at end of file