diff --git a/jredisgraph.iml b/jredisgraph.iml new file mode 100644 index 0000000..79afeb9 --- /dev/null +++ b/jredisgraph.iml @@ -0,0 +1,31 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/src/main/java/com/redislabs/redisgraph/RedisGraph.java b/src/main/java/com/redislabs/redisgraph/RedisGraph.java index 6cc57ce..55c4bba 100644 --- a/src/main/java/com/redislabs/redisgraph/RedisGraph.java +++ b/src/main/java/com/redislabs/redisgraph/RedisGraph.java @@ -127,4 +127,21 @@ public interface RedisGraph extends Closeable { @Override void close(); + + + /** + * Executes a cypher query with parameters and redisgraph timeout. + * After that block the current client until all the previous cypher write queries + * are successfully transferred and acknowledged by at least 1 replica. + * If the replicationTimeout, specified in milliseconds, is reached, + * the method returns even if the specified number of replicas were not yet reached. + * + * @param graphId graph to be queried + * @param query Cypher query. + * @param params parameters map. + * @param redisGraphTimeout + * @param replicationTimeout replication timeout, specified in milliseconds + * @return a result set + */ + ResultSet replicatedQuery(String graphId, String query, Map params, long redisGraphTimeout, long replicationTimeout); } diff --git a/src/main/java/com/redislabs/redisgraph/ResultSet.java b/src/main/java/com/redislabs/redisgraph/ResultSet.java index d86ff19..2a3a4b6 100644 --- a/src/main/java/com/redislabs/redisgraph/ResultSet.java +++ b/src/main/java/com/redislabs/redisgraph/ResultSet.java @@ -9,6 +9,8 @@ public interface ResultSet extends Iterable, Iterator { int size(); + long numberReplicasReached(); + Statistics getStatistics(); Header getHeader(); diff --git a/src/main/java/com/redislabs/redisgraph/impl/api/AbstractRedisGraph.java b/src/main/java/com/redislabs/redisgraph/impl/api/AbstractRedisGraph.java index 57fce53..ea49969 100644 --- a/src/main/java/com/redislabs/redisgraph/impl/api/AbstractRedisGraph.java +++ b/src/main/java/com/redislabs/redisgraph/impl/api/AbstractRedisGraph.java @@ -53,6 +53,19 @@ public abstract class AbstractRedisGraph implements RedisGraph { */ protected abstract ResultSet sendReadOnlyQuery(String graphId, String preparedQuery, long timeout); + /** + * Executes a cypher query with parameters and redisgraph timeout. + * After that block the current client until all the previous cypher write queries + * are successfully transferred and acknowledged by at least 1 replica. + * If the replicationTimeout, specified in milliseconds, is reached, + * the method returns even if the specified number of replicas were not yet reached. + * @param graphId graph to be queried + * @param preparedQuery prepared query + * @param replicationTimeout replication timeout, specified in milliseconds + * @return a result set + */ + protected abstract ResultSet sendReplicatedQuery(String graphId, String preparedQuery, long redisGraphTimeout, long replicationTimeout); + /** * Execute a Cypher query. * @param graphId a graph to perform the query on @@ -149,6 +162,25 @@ public ResultSet query(String graphId, String query, Map params, return sendQuery(graphId, preparedQuery, timeout); } + + /** + * Executes a cypher query with parameters and redisgraph timeout. + * After that block the current client until all the previous cypher write queries + * are successfully transferred and acknowledged by at least 1 replica. + * If the replicationTimeout, specified in milliseconds, is reached, + * the method returns even if the specified number of replicas were not yet reached. + * @param graphId a graph to perform the query on. + * @param query Cypher query. + * @param params parameters map. + * @param redisGraphTimeout + * @param replicationTimeout replication timeout, specified in milliseconds + * @return a result set. + */ + public ResultSet replicatedQuery(String graphId, String query, Map params, long redisGraphTimeout, long replicationTimeout){ + String preparedQuery = Utils.prepareQuery(query, params); + return sendReplicatedQuery(graphId, preparedQuery, redisGraphTimeout,replicationTimeout); + } + /** * Executes a cypher read-only query with parameters and timeout. * @param graphId a graph to perform the query on. diff --git a/src/main/java/com/redislabs/redisgraph/impl/api/ContextedRedisGraph.java b/src/main/java/com/redislabs/redisgraph/impl/api/ContextedRedisGraph.java index 92896d0..9380db0 100644 --- a/src/main/java/com/redislabs/redisgraph/impl/api/ContextedRedisGraph.java +++ b/src/main/java/com/redislabs/redisgraph/impl/api/ContextedRedisGraph.java @@ -8,9 +8,12 @@ import com.redislabs.redisgraph.impl.resultset.ResultSetImpl; import redis.clients.jedis.Client; import redis.clients.jedis.Jedis; +import redis.clients.jedis.Pipeline; +import redis.clients.jedis.Response; import redis.clients.jedis.util.SafeEncoder; import redis.clients.jedis.exceptions.JedisDataException; import java.util.List; +import java.util.Map; /** * An implementation of RedisGraphContext. Allows sending RedisGraph and some Redis commands, @@ -126,6 +129,42 @@ protected ResultSet sendReadOnlyQuery(String graphId, String preparedQuery, long } } + /** + * Executes a cypher query with parameters and redisgraph timeout. + * After that block the current client until all the previous cypher write queries + * are successfully transferred and acknowledged by at least 1 replica. + * If the replicationTimeout, specified in milliseconds, is reached, + * the method returns even if the specified number of replicas were not yet reached. + * + * @param graphId graph to be queried + * @param preparedQuery prepared query + * @param redisGraphTimeout + * @param replicationTimeout replication timeout, specified in milliseconds + * @return a result set + */ + @Override + protected ResultSet sendReplicatedQuery(String graphId, String preparedQuery, long redisGraphTimeout, long replicationTimeout) { + Jedis conn = getConnection(); + try { + Pipeline pipe = conn.pipelined(); + pipe.setClient(conn.getClient()); + Response rawResponse = pipe.sendCommand(RedisGraphCommand.QUERY, + graphId, preparedQuery, Utils.COMPACT_STRING, Utils.TIMEOUT_STRING, Long.toString(redisGraphTimeout)); + Response waitResponse = pipe.sendCommand(RedisGraphCommand.WAIT,"1", Long.toString(replicationTimeout)); + pipe.sync(); + ResultSetImpl resultSet = new ResultSetImpl((List) rawResponse.get(), this, caches.getGraphCache(graphId)); + Long numberReplicasReached = (Long) waitResponse.get(); + resultSet.setNumberReplicasReached(numberReplicasReached); + return resultSet; + } + catch (JRedisGraphException ge) { + throw ge; + } + catch (JedisDataException de) { + throw new JRedisGraphException(de); + } + } + /** * @return Returns the instance Jedis connection. */ diff --git a/src/main/java/com/redislabs/redisgraph/impl/api/RedisGraph.java b/src/main/java/com/redislabs/redisgraph/impl/api/RedisGraph.java index 1bb0f50..5474e3e 100644 --- a/src/main/java/com/redislabs/redisgraph/impl/api/RedisGraph.java +++ b/src/main/java/com/redislabs/redisgraph/impl/api/RedisGraph.java @@ -3,12 +3,21 @@ import com.redislabs.redisgraph.RedisGraphContext; import com.redislabs.redisgraph.RedisGraphContextGenerator; import com.redislabs.redisgraph.ResultSet; +import com.redislabs.redisgraph.exceptions.JRedisGraphException; +import com.redislabs.redisgraph.impl.Utils; import com.redislabs.redisgraph.impl.graph_cache.RedisGraphCaches; +import com.redislabs.redisgraph.impl.resultset.ResultSetImpl; +import redis.clients.jedis.Client; import redis.clients.jedis.Jedis; import redis.clients.jedis.JedisPool; +import redis.clients.jedis.Pipeline; +import redis.clients.jedis.exceptions.JedisDataException; import redis.clients.jedis.util.Pool; import redis.clients.jedis.util.SafeEncoder; +import java.util.List; +import java.util.Map; + /** * */ @@ -116,6 +125,27 @@ protected ResultSet sendReadOnlyQuery(String graphId, String preparedQuery, long } } + /** + * Executes a cypher query with parameters and redisgraph timeout. + * After that block the current client until all the previous cypher write queries + * are successfully transferred and acknowledged by at least 1 replica. + * If the replicationTimeout, specified in milliseconds, is reached, + * the method returns even if the specified number of replicas were not yet reached. + * + * @param graphId graph to be queried + * @param preparedQuery prepared query + * @param redisGraphTimeout + * @param replicationTimeout replication timeout, specified in milliseconds + * @return a result set + */ + @Override + protected ResultSet sendReplicatedQuery(String graphId, String preparedQuery, long redisGraphTimeout, long replicationTimeout) { + try (ContextedRedisGraph contextedRedisGraph = new ContextedRedisGraph(getConnection())) { + contextedRedisGraph.setRedisGraphCaches(caches); + return contextedRedisGraph.sendReplicatedQuery(graphId, preparedQuery, redisGraphTimeout,replicationTimeout); + } + } + /** * Closes the Jedis pool */ @@ -124,7 +154,6 @@ public void close(){ this.client.close(); } - /** * Deletes the entire graph * @param graphId graph to delete diff --git a/src/main/java/com/redislabs/redisgraph/impl/api/RedisGraphCommand.java b/src/main/java/com/redislabs/redisgraph/impl/api/RedisGraphCommand.java index 0af0f61..1d23ef9 100644 --- a/src/main/java/com/redislabs/redisgraph/impl/api/RedisGraphCommand.java +++ b/src/main/java/com/redislabs/redisgraph/impl/api/RedisGraphCommand.java @@ -10,7 +10,8 @@ public enum RedisGraphCommand implements ProtocolCommand { QUERY("graph.QUERY"), RO_QUERY("graph.RO_QUERY"), - DELETE("graph.DELETE"); + DELETE("graph.DELETE"), + WAIT("WAIT"); private final byte[] raw; diff --git a/src/main/java/com/redislabs/redisgraph/impl/resultset/ResultSetImpl.java b/src/main/java/com/redislabs/redisgraph/impl/resultset/ResultSetImpl.java index 179db89..9fedc9b 100644 --- a/src/main/java/com/redislabs/redisgraph/impl/resultset/ResultSetImpl.java +++ b/src/main/java/com/redislabs/redisgraph/impl/resultset/ResultSetImpl.java @@ -19,6 +19,7 @@ public class ResultSetImpl implements ResultSet { private final Header header; private final Statistics statistics; private final List results; + private long numberReplicasReached = 0; private int position = 0; private final RedisGraph redisGraph; @@ -333,4 +334,12 @@ public Iterator iterator() { // TODO Auto-generated method stub return results.iterator(); } + + public void setNumberReplicasReached(Long numberReplicasReached) { + this.numberReplicasReached = numberReplicasReached; + } + + public long numberReplicasReached() { + return this.numberReplicasReached; + } } diff --git a/src/test/java/com/redislabs/redisgraph/RedisGraphAPITest.java b/src/test/java/com/redislabs/redisgraph/RedisGraphAPITest.java index 8a137bc..ab57511 100644 --- a/src/test/java/com/redislabs/redisgraph/RedisGraphAPITest.java +++ b/src/test/java/com/redislabs/redisgraph/RedisGraphAPITest.java @@ -14,6 +14,8 @@ import java.util.stream.Collectors; import java.util.stream.IntStream; import org.junit.*; +import redis.clients.jedis.JedisPool; +import redis.clients.jedis.JedisPoolConfig; public class RedisGraphAPITest { @@ -974,6 +976,19 @@ public void testCachedExecution() { r = resultSet.next(); Assert.assertEquals(params.get("val"), r.getValue(0)); Assert.assertTrue(resultSet.getStatistics().cachedExecution()); + + JedisPoolConfig jedisPoolConfig = new JedisPoolConfig(); + jedisPoolConfig.setMaxWaitMillis(10000); + JedisPool pool = new JedisPool(jedisPoolConfig, "localhost", 6379, 10000); + long connected_replicas = Long.parseLong(pool.getResource().info("replication").split("\n")[2].split(":")[1].trim()); + RedisGraph apiT = new RedisGraph(pool); + // Assert hat replicatedQuery() produces the same results + resultSet = apiT.replicatedQuery("social","MATCH (n:N {val:$val}) RETURN n.val", params,10000,5000); + Assert.assertEquals(resultSet.numberReplicasReached(),(long)connected_replicas); + r = resultSet.next(); + // Ensure that we have the same number of replicas reached, as the number of connected replicas on redis + Assert.assertEquals(params.get("val"), r.getValue(0)); + Assert.assertTrue(resultSet.getStatistics().cachedExecution()); } @Test