github/ebean
4
0
Fork 0
mirror of https://github.com/ebean-orm/ebean.git synced 2026-10-22 09:30:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2132 from ebean-orm/feature/redisModule

Browse Source 
Add ebean-redis as module
This commit is contained in:
Rob Bygrave
2020-12-18 14:58:07 +13:00
committed by GitHub
parent 55324653a5 1bdc6d8f7f
commit 5393161c16
35 changed files with 2234 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ebean-core/src/main/java/io/ebeaninternal/api/SpiTransaction.java
+16
View File
@@ -318,4 +318,20 @@ public interface SpiTransaction extends Transaction {
* Return true if explicitly set to skip cache (ignores skipOnWrite).
*/
boolean isSkipCacheExplicit();
/**
* Fire pre commit processing/listeners.
*/
void preCommit();
/**
* Fire post commit events and listeners.
*/
void postCommit();
/**
* Fire post rollback events and listeners.
*/
void postRollback(Throwable cause);
}
ebean-core/src/main/java/io/ebeaninternal/api/SpiTransactionProxy.java
+14
View File
@@ -418,4 +418,18 @@ public abstract class SpiTransactionProxy implements SpiTransaction {
transaction.flushBatchOnCollection();
}
@Override
public void preCommit() {
transaction.preCommit();
}
@Override
public void postCommit() {
transaction.postCommit();
}
@Override
public void postRollback(Throwable cause) {
transaction.postRollback(cause);
}
}
ebean-core/src/main/java/io/ebeaninternal/server/transaction/ImplicitReadOnlyTransaction.java
+15
View File
@@ -586,6 +586,21 @@ class ImplicitReadOnlyTransaction implements SpiTransaction, TxnProfileEventCode
}
}
@Override
public void preCommit() {
// do nothing
}
@Override
public void postCommit() {
// do nothing
}
@Override
public void postRollback(Throwable cause) {
// do nothing
}
/**
* Return true if the transaction is active.
*/
ebean-core/src/main/java/io/ebeaninternal/server/transaction/JdbcTransaction.java
+22 -5
View File
@@ -373,6 +373,9 @@ class JdbcTransaction implements SpiTransaction, TxnProfileEventCodes {
private void firePreCommit() {
withEachCallback(TransactionCallback::preCommit);
if (changeLogHolder != null) {
changeLogHolder.preCommit();
}
}
private void firePostCommit() {
@@ -981,14 +984,23 @@ class JdbcTransaction implements SpiTransaction, TxnProfileEventCodes {
* Batch flush, jdbc commit, trigger registered TransactionCallbacks, notify l2 cache etc.
*/
private void flushCommitAndNotify() throws SQLException {
internalBatchFlush();
firePreCommit();
// only performCommit can throw an exception
preCommit();
performCommit();
postCommit();
}
@Override
public void postCommit() {
firePostCommit();
notifyCommit();
}
@Override
public void preCommit() {
internalBatchFlush();
firePreCommit();
}
/**
* Perform a commit, fire callbacks and notify l2 cache etc.
* <p>
@@ -1132,11 +1144,16 @@ class JdbcTransaction implements SpiTransaction, TxnProfileEventCodes {
} finally {
// these will not throw an exception
firePostRollback();
notifyRollback(cause);
postRollback(cause);
}
}
@Override
public void postRollback(Throwable cause) {
firePostRollback();
notifyRollback(cause);
}
/**
* If the transaction is active then perform rollback.
*/
ebean-core/src/main/java/io/ebeaninternal/server/transaction/JtaTransaction.java
+7 -9
View File
@@ -13,9 +13,9 @@ public class JtaTransaction extends JdbcTransaction {
private final UserTransaction userTransaction;
private boolean commmitted;
private final boolean newTransaction;
private boolean newTransaction;
private boolean committed;
/**
* Create the JtaTransaction.
@@ -41,7 +41,6 @@ public class JtaTransaction extends JdbcTransaction {
if (connection.getAutoCommit()) {
connection.setAutoCommit(false);
}
} catch (SQLException e) {
throw new PersistenceException(e);
}
@@ -52,7 +51,7 @@ public class JtaTransaction extends JdbcTransaction {
*/
@Override
public void commit() {
if (commmitted) {
if (committed) {
throw new PersistenceException("This transaction has already been committed.");
}
try {
@@ -60,14 +59,14 @@ public class JtaTransaction extends JdbcTransaction {
if (newTransaction) {
userTransaction.commit();
}
notifyCommit();
postCommit();
} finally {
close();
}
} catch (Exception e) {
throw new PersistenceException(e);
}
commmitted = true;
committed = true;
}
@Override
@@ -80,7 +79,7 @@ public class JtaTransaction extends JdbcTransaction {
*/
@Override
public void rollback(Throwable e) {
if (!commmitted) {
if (!committed) {
try {
try {
if (userTransaction != null) {
@@ -90,7 +89,7 @@ public class JtaTransaction extends JdbcTransaction {
userTransaction.setRollbackOnly();
}
}
notifyRollback(e);
postRollback(e);
} finally {
closeConnection();
}
@@ -98,7 +97,6 @@ public class JtaTransaction extends JdbcTransaction {
throw new PersistenceException(ex);
}
}
}
/**
ebean-core/src/main/java/io/ebeaninternal/server/transaction/JtaTransactionManager.java
+3 -5
View File
@@ -190,19 +190,17 @@ public class JtaTransactionManager implements ExternalTransactionManager {
@Override
public void beforeCompletion() {
// Future note: for JPA2 locking we will
// have beforeCommit events to fire
transaction.preCommit();
}
@Override
public void afterCompletion(int status) {
switch (status) {
case Status.STATUS_COMMITTED:
if (logger.isDebugEnabled()) {
logger.debug("Jta Txn [" + transaction.getId() + "] committed");
}
transactionManager.notifyOfCommit(transaction);
transaction.postCommit();
// Remove this transaction object as it is completed
transactionManager.scope().clearExternal();
break;
@@ -211,7 +209,7 @@ public class JtaTransactionManager implements ExternalTransactionManager {
if (logger.isDebugEnabled()) {
logger.debug("Jta Txn [" + transaction.getId() + "] rollback");
}
transactionManager.notifyOfRollback(transaction, null);
transaction.postRollback(null);
// Remove this transaction object as it is completed
transactionManager.scope().clearExternal();
break;
ebean-core/src/main/java/io/ebeaninternal/server/transaction/NoTransaction.java
+15
View File
@@ -88,6 +88,21 @@ class NoTransaction implements SpiTransaction {
// do nothing
}
@Override
public void preCommit() {
// do nothing
}
@Override
public void postCommit() {
// do nothing
}
@Override
public void postRollback(Throwable cause) {
// do nothing
}
@Override
public String getLogPrefix() {
return null;
ebean-core/src/main/java/io/ebeaninternal/server/transaction/TChangeLogHolder.java
+16 -5
View File
@@ -56,20 +56,31 @@ public class TChangeLogHolder {
* Add a bean change to the change set.
*/
public void addBeanChange(BeanChange change) {
changes.addBeanChange(change);
if (++count >= batchSize) {
// we hit the batch size so send what we have knowing
// that the transaction has not completed yet and
// reset the changes and count
owner.sendChangeLog(changes);
changes = new ChangeSet(transactionId, ++batchId);
count = 0;
sendChanges();
}
}
private void sendChanges() {
owner.sendChangeLog(changes);
changes = new ChangeSet(transactionId, ++batchId);
count = 0;
}
/**
* On post commit send the changes we have collected.
* Send the changes held prior to transaction commit.
*/
public void preCommit() {
sendChanges();
}
/**
* On post commit send the changes we have collected. This should be
* only the COMMITTED state and with all changes sent prior to commit.
*/
public void postCommit() {
changes.setTxnState(TxnState.COMMITTED);
ebean-core/src/test/java/org/tests/changelog/TestChangeLog.java
+24 -14
View File
@@ -15,11 +15,15 @@ import io.ebean.event.changelog.ChangeLogPrepare;
import io.ebean.event.changelog.ChangeLogRegister;
import io.ebean.event.changelog.ChangeSet;
import io.ebean.event.changelog.ChangeType;
import io.ebean.event.changelog.TxnState;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.tests.model.basic.EBasicChangeLog;
import java.util.ArrayList;
import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
@@ -43,6 +47,13 @@ public class TestChangeLog extends BaseTestCase {
server.shutdown();
}
private BeanChange firstChange() {
ChangeSet changeSet0 = changeLogListener.changes.get(0);
BeanChange change = changeSet0.getChanges().get(0);
changeLogListener.changes.clear();
return change;
}
@Test
public void test() {
@@ -50,8 +61,13 @@ public class TestChangeLog extends BaseTestCase {
bean.setName("logBean");
bean.setShortDescription("hello");
server.save(bean);
BeanChange change = changeLogListener.changes.getChanges().get(0);
final List<ChangeSet> allChanges = changeLogListener.changes;
assertThat(allChanges).hasSize(2);
assertThat(allChanges.get(0).getTxnState()).isEqualTo(TxnState.IN_PROGRESS);
assertThat(allChanges.get(1).getTxnState()).isEqualTo(TxnState.COMMITTED);
BeanChange change = firstChange();
assertThat(change.getEvent()).isEqualTo(ChangeType.INSERT);
assertThat(change.getData())
.contains("\"name\":\"logBean\"")
@@ -61,8 +77,7 @@ public class TestChangeLog extends BaseTestCase {
bean.setName("ChangedName");
server.save(bean);
change = changeLogListener.changes.getChanges().get(0);
change = firstChange();
assertThat(change.getEvent()).isEqualTo(ChangeType.UPDATE);
assertThat(change.getOldData()).contains("\"name\":\"logBean\"");
assertThat(change.getData()).contains("\"name\":\"ChangedName\"");
@@ -70,14 +85,11 @@ public class TestChangeLog extends BaseTestCase {
server.delete(bean);
change = changeLogListener.changes.getChanges().get(0);
change = firstChange();
assertThat(change.getEvent()).isEqualTo(ChangeType.DELETE);
assertThat(change.getData()).isNull();
}
@Test
public void testWithNull() {
@@ -85,7 +97,7 @@ public class TestChangeLog extends BaseTestCase {
bean.setName(null);
bean.setShortDescription("hello");
server.save(bean);
BeanChange change = changeLogListener.changes.getChanges().get(0);
BeanChange change = firstChange();
assertThat(change.getEvent()).isEqualTo(ChangeType.INSERT);
assertThat(change.getData())
@@ -99,7 +111,7 @@ public class TestChangeLog extends BaseTestCase {
bean.setShortDescription("hello");
server.save(bean);
change = changeLogListener.changes.getChanges().get(0);
change = firstChange();
assertThat(change.getEvent()).isEqualTo(ChangeType.UPDATE);
assertThat(change.getOldData())
@@ -113,11 +125,10 @@ public class TestChangeLog extends BaseTestCase {
server.delete(bean);
change = changeLogListener.changes.getChanges().get(0);
change = firstChange();
assertThat(change.getEvent()).isEqualTo(ChangeType.DELETE);
assertThat(change.getData()).isNull();
}
private Database createServer() {
@@ -153,16 +164,15 @@ public class TestChangeLog extends BaseTestCase {
ObjectMapper objectMapper = new ObjectMapper();
ChangeSet changes;
List<ChangeSet> changes = new ArrayList<>();
public TDChangeLogListener() {
objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
}
@Override
public void log(ChangeSet changes) {
this.changes = changes;
this.changes.add(changes);
try {
String json = objectMapper.writeValueAsString(changes);
ChangeSet changes1 = objectMapper.readValue(json, ChangeSet.class);
ebean-redis/pom.xml
+94
View File
@@ -0,0 +1,94 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<artifactId>ebean-parent</artifactId>
<groupId>io.ebean</groupId>
<version>12.6.3-SNAPSHOT</version>
</parent>
<artifactId>ebean-redis</artifactId>
<name>ebean redis</name>
<description>Ebean Redis L2 Cache</description>
<dependencies>
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>3.4.0</version>
</dependency>
<dependency>
<groupId>io.ebean</groupId>
<artifactId>ebean-api</artifactId>
<version>12.6.3-SNAPSHOT</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>io.ebean</groupId>
<artifactId>ebean-core</artifactId>
<version>12.6.3-SNAPSHOT</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>io.ebean</groupId>
<artifactId>ebean-querybean</artifactId>
<version>12.6.3-SNAPSHOT</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.ebean</groupId>
<artifactId>querybean-generator</artifactId>
<version>12.6.3-SNAPSHOT</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.ebean</groupId>
<artifactId>ebean-test</artifactId>
<version>12.6.3-SNAPSHOT</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.avaje.composite</groupId>
<artifactId>junit</artifactId>
<version>1.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.avaje.composite</groupId>
<artifactId>logback</artifactId>
<version>1.1</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>io.repaint.maven</groupId>
<artifactId>tiles-maven-plugin</artifactId>
<version>2.18</version>
<extensions>true</extensions>
<configuration>
<tiles>
<tile>io.ebean.tile:enhancement:12.6.1</tile>
</tiles>
</configuration>
</plugin>
</plugins>
</build>
</project>
ebean-redis/src/main/java/io/ebean/redis/DuelCache.java
+161
View File
@@ -0,0 +1,161 @@
package io.ebean.redis;
import io.ebean.cache.ServerCache;
import io.ebean.cache.ServerCacheStatistics;
import io.ebean.meta.MetricVisitor;
import io.ebeaninternal.server.cache.DefaultServerCache;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
public class DuelCache implements ServerCache, NearCacheInvalidate {
private final DefaultServerCache near;
private final RedisCache remote;
private final NearCacheNotify cacheNotify;
private final String cacheKey;
public DuelCache(DefaultServerCache near, RedisCache remote, String cacheKey, NearCacheNotify cacheNotify) {
this.near = near;
this.remote = remote;
this.cacheKey = cacheKey;
this.cacheNotify = cacheNotify;
}
@Override
public void visit(MetricVisitor visitor) {
near.visit(visitor);
remote.visit(visitor);
}
@Override
public void invalidateKeys(Set<Object> keySet) {
near.removeAll(keySet);
}
@Override
public void invalidateKey(Object id) {
near.remove(id);
}
@Override
public void invalidateClear() {
near.clear();
}
@Override
public Map<Object, Object> getAll(Set<Object> keys) {
Map<Object, Object> resultMap = near.getAll(keys);
Set<Object> localKeys = resultMap.keySet();
Set<Object> remainingKeys = new HashSet<>();
for (Object key : keys) {
if (!localKeys.contains(key)) {
remainingKeys.add(key);
}
}
if (!remainingKeys.isEmpty()) {
// fetch missing ones from remote cache and merge results
Map<Object, Object> remoteMap = remote.getAll(remainingKeys);
if (!remoteMap.isEmpty()) {
near.putAll(remoteMap);
resultMap.putAll(remoteMap);
}
}
return resultMap;
}
@Override
public Object get(Object id) {
Object val = near.get(id);
if (val != null) {
return val;
}
Object remoteVal = remote.get(id);
if (remoteVal != null) {
near.put(id, remoteVal);
}
return remoteVal;
}
@Override
public void putAll(Map<Object, Object> keyValues) {
near.putAll(keyValues);
remote.putAll(keyValues);
cacheNotify.invalidateKeys(cacheKey, keyValues.keySet());
}
@Override
public void put(Object id, Object value) {
near.put(id, value);
remote.put(id, value);
cacheNotify.invalidateKey(cacheKey, id);
}
@Override
public void removeAll(Set<Object> keys) {
near.removeAll(keys);
remote.removeAll(keys);
cacheNotify.invalidateKeys(cacheKey, keys);
}
@Override
public void remove(Object id) {
near.remove(id);
remote.remove(id);
cacheNotify.invalidateKey(cacheKey, id);
}
@Override
public void clear() {
near.clear();
remote.clear();
cacheNotify.invalidateClear(cacheKey);
}
/**
* Return the near cache hit count.
*/
public long getNearHitCount() {
return near.getHitCount();
}
/**
* Return the near cache miss count.
*/
public long getNearMissCount() {
return near.getMissCount();
}
/**
* Return the redis cache hit count.
*/
public long getRemoteHitCount() {
return remote.getHitCount();
}
/**
* Return the redis cache miss count.
*/
public long getRemoteMissCount() {
return remote.getMissCount();
}
@Override
public int size() {
return 0;
}
@Override
public int getHitRatio() {
return 0;
}
@Override
public ServerCacheStatistics getStatistics(boolean reset) {
return null;
}
}
ebean-redis/src/main/java/io/ebean/redis/ModId.java
+49
View File
@@ -0,0 +1,49 @@
package io.ebean.redis;
import java.security.SecureRandom;
import java.util.Base64;
/**
* Provides a modified base64 encoded UUID and shorter 12 character random unique value.
* <p>
* <h3>newId()</h3>
* <p>
* It produces a 22 character string that is a base64 encoded UUID with the +
* and / characters replaced with - and _ so as to be URL safe without requiring
* encoding.
* </p>
* <h3>newShortId()</h3>
* <p>
* It produces a 12 character string that base64 encoded random number (72 bit).
* </p>
* <p>
* Note that this now internally uses java.util.Base64 to encode the values.
* </p>
*/
public class ModId {
private static final SecureRandom shortIdSecureRandom = new SecureRandom();
private static final Base64.Encoder urlEncoder = Base64.getUrlEncoder();
/**
* Return a 12 character string using a 72 bit randomly generated ID encoded
* in modified base64.
* <p>
* A UUID is 128 bits and this is 72 bits so quite a bit smaller but still
* very random with one in 4.7 * 10^21 chance of a collision.
* </p>
*/
public static String id() {
// Random 72 bits
byte[] randomBytes = new byte[9];
shortIdSecureRandom.nextBytes(randomBytes);
return encode64(randomBytes);
}
private static String encode64(byte[] bytes) {
return urlEncoder.encodeToString(bytes);
}
}
ebean-redis/src/main/java/io/ebean/redis/NearCacheInvalidate.java
+24
View File
@@ -0,0 +1,24 @@
package io.ebean.redis;
import java.util.Set;
/**
* Near cache invalidation.
*/
public interface NearCacheInvalidate {
/**
* Invalidate from near cache the given keys.
*/
void invalidateKeys(Set<Object> keySet);
/**
* Invalidate from near cache the given key.
*/
void invalidateKey(Object id);
/**
* Clear the near cache.
*/
void invalidateClear();
}
ebean-redis/src/main/java/io/ebean/redis/NearCacheNotify.java
+24
View File
@@ -0,0 +1,24 @@
package io.ebean.redis;
import java.util.Set;
/**
* Notify other cluster members to invalidate parts of their near cache.
*/
public interface NearCacheNotify {
/**
* Invalidate the given keys.
*/
void invalidateKeys(String cacheKey, Set<Object> keySet);
/**
* Invalidate a single key.
*/
void invalidateKey(String cacheKey, Object id);
/**
* Clear a near cache.
*/
void invalidateClear(String cacheKey);
}
ebean-redis/src/main/java/io/ebean/redis/RedisCache.java
+268
View File
@@ -0,0 +1,268 @@
package io.ebean.redis;
import io.ebean.cache.ServerCache;
import io.ebean.cache.ServerCacheConfig;
import io.ebean.cache.ServerCacheStatistics;
import io.ebean.meta.MetricVisitor;
import io.ebean.metric.CountMetric;
import io.ebean.metric.MetricFactory;
import io.ebean.metric.TimedMetric;
import io.ebean.redis.encode.Encode;
import io.ebean.redis.encode.EncodePrefixKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.ScanParams;
import redis.clients.jedis.ScanResult;
import redis.clients.jedis.util.SafeEncoder;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
class RedisCache implements ServerCache {
private static final Logger log = LoggerFactory.getLogger(RedisCache.class);
private static final String CURSOR_0 = "0";
private static final byte[] CURSOR_0_BYTES = SafeEncoder.encode(CURSOR_0);
private final JedisPool jedisPool;
private final String cacheKey;
private final Encode keyEncode;
private final Encode valueEncode;
private final TimedMetric metricGet;
private final TimedMetric metricGetAll;
private final TimedMetric metricPut;
private final TimedMetric metricPutAll;
private final TimedMetric metricRemove;
private final TimedMetric metricRemoveAll;
private final TimedMetric metricClear;
private final CountMetric hitCount;
private final CountMetric missCount;
RedisCache(JedisPool jedisPool, ServerCacheConfig config, Encode valueEncode) {
this.jedisPool = jedisPool;
this.cacheKey = config.getCacheKey();
this.keyEncode = new EncodePrefixKey(config.getCacheKey());
this.valueEncode = valueEncode;
String pre = "l2r.";
String shortName = config.getShortName();
MetricFactory factory = MetricFactory.get();
hitCount = factory.createCountMetric(pre + shortName + ".hit");
missCount = factory.createCountMetric(pre + shortName + ".miss");
metricGet = factory.createTimedMetric(pre + shortName + ".get");
metricGetAll = factory.createTimedMetric(pre + shortName + ".getMany");
metricPut = factory.createTimedMetric(pre + shortName + ".put");
metricPutAll = factory.createTimedMetric(pre + shortName + ".putMany");
metricRemove = factory.createTimedMetric(pre + shortName + ".remove");
metricRemoveAll = factory.createTimedMetric(pre + shortName + ".removeMany");
metricClear = factory.createTimedMetric(pre + shortName + ".clear");
}
@Override
public void visit(MetricVisitor visitor) {
hitCount.visit(visitor);
missCount.visit(visitor);
metricGet.visit(visitor);
metricGetAll.visit(visitor);
metricPut.visit(visitor);
metricPutAll.visit(visitor);
metricRemove.visit(visitor);
metricRemoveAll.visit(visitor);
metricClear.visit(visitor);
}
private byte[] key(Object id) {
return keyEncode.encode(id);
}
private byte[] value(Object data) {
if (data == null) {
return null;
}
return valueEncode.encode(data);
}
private Object valueDecode(byte[] data) {
try {
if (data == null) {
return null;
}
return valueEncode.decode(data);
} catch (Exception e) {
log.error("Error decoding data, treated as cache miss", e);
return null;
}
}
@Override
public Map<Object, Object> getAll(Set<Object> keys) {
long start = System.nanoTime();
Map<Object, Object> map = new LinkedHashMap<>();
List<Object> keyList = new ArrayList<>(keys);
try (Jedis resource = jedisPool.getResource()) {
List<byte[]> valsAsBytes = resource.mget(keysAsBytes(keyList));
for (int i = 0; i < keyList.size(); i++) {
Object val = valueDecode(valsAsBytes.get(i));
if (val != null) {
map.put(keyList.get(i), val);
}
}
int hits = map.size();
int miss = keys.size() - hits;
if (hits > 0) {
hitCount.add(hits);
}
if (miss > 0) {
missCount.add(miss);
}
metricGetAll.addSinceNanos(start);
return map;
}
}
@Override
public Object get(Object id) {
long start = System.nanoTime();
try (Jedis resource = jedisPool.getResource()) {
Object val = valueDecode(resource.get(key(id)));
if (val != null) {
hitCount.increment();
} else {
missCount.increment();
}
metricGet.addSinceNanos(start);
return val;
}
}
@Override
public void put(Object id, Object value) {
long start = System.nanoTime();
try (Jedis resource = jedisPool.getResource()) {
resource.set(key(id), value(value));
metricPut.addSinceNanos(start);
}
}
@Override
public void putAll(Map<Object, Object> keyValues) {
long start = System.nanoTime();
try (Jedis resource = jedisPool.getResource()) {
byte[][] raw = new byte[keyValues.size() * 2][];
int pos = 0;
for (Map.Entry<Object, Object> entry : keyValues.entrySet()) {
raw[pos++] = key(entry.getKey());
raw[pos++] = value(entry.getValue());
}
resource.mset(raw);
metricPutAll.addSinceNanos(start);
}
}
@Override
public void remove(Object id) {
long start = System.nanoTime();
try (Jedis resource = jedisPool.getResource()) {
resource.del(key(id));
metricRemove.addSinceNanos(start);
}
}
@Override
public void removeAll(Set<Object> keys) {
long start = System.nanoTime();
try (Jedis resource = jedisPool.getResource()) {
resource.del(keysAsBytes(keys));
metricRemoveAll.addSinceNanos(start);
}
}
private byte[][] keysAsBytes(Collection<Object> keys) {
byte[][] raw = new byte[keys.size()][];
int pos = 0;
for (Object id : keys) {
raw[pos++] = key(id);
}
return raw;
}
@Override
public void clear() {
long start = System.nanoTime();
try (Jedis resource = jedisPool.getResource()) {
ScanParams params = new ScanParams();
params.match(cacheKey + ":*");
String next;
byte[] nextCursor = CURSOR_0_BYTES;
do {
ScanResult<byte[]> scanResult = resource.scan(nextCursor, params);
List<byte[]> keys = scanResult.getResult();
nextCursor = scanResult.getCursorAsBytes();
if (!keys.isEmpty()) {
byte[][] raw = new byte[keys.size()][];
for (int i = 0; i < keys.size(); i++) {
raw[i] = keys.get(i);
}
resource.del(raw);
}
next = SafeEncoder.encode(nextCursor);
} while (!next.equals("0"));
metricClear.addSinceNanos(start);
}
}
/**
* Return the count of get hits.
*/
public long getHitCount() {
return hitCount.get(false);
}
/**
* Return the count of get misses.
*/
public long getMissCount() {
return missCount.get(false);
}
@Override
public int size() {
return 0;
}
@Override
public int getHitRatio() {
return 0;
}
@Override
public ServerCacheStatistics getStatistics(boolean reset) {
ServerCacheStatistics cacheStats = new ServerCacheStatistics();
cacheStats.setCacheName(cacheKey);
cacheStats.setHitCount(hitCount.get(reset));
cacheStats.setMissCount(missCount.get(reset));
cacheStats.setPutCount(metricPut.collect(reset).getCount());
cacheStats.setRemoveCount(metricRemove.collect(reset).getCount());
cacheStats.setClearCount(metricClear.collect(reset).getCount());
return cacheStats;
}
}
ebean-redis/src/main/java/io/ebean/redis/RedisCacheFactory.java
+525
View File
@@ -0,0 +1,525 @@
package io.ebean.redis;
import io.ebean.BackgroundExecutor;
import io.ebean.cache.ServerCache;
import io.ebean.cache.ServerCacheConfig;
import io.ebean.cache.ServerCacheFactory;
import io.ebean.cache.ServerCacheNotification;
import io.ebean.cache.ServerCacheNotify;
import io.ebean.config.DatabaseConfig;
import io.ebean.meta.MetricVisitor;
import io.ebean.metric.MetricFactory;
import io.ebean.metric.TimedMetric;
import io.ebean.redis.encode.EncodeBeanData;
import io.ebean.redis.encode.EncodeManyIdsData;
import io.ebean.redis.encode.EncodeSerializable;
import io.ebean.redis.topic.DaemonTopic;
import io.ebean.redis.topic.DaemonTopicRunner;
import io.ebeaninternal.server.cache.DefaultServerCache;
import io.ebeaninternal.server.cache.DefaultServerCacheConfig;
import io.ebeaninternal.server.cache.DefaultServerQueryCache;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.BinaryJedisPubSub;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.util.SafeEncoder;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import static java.util.Arrays.asList;
class RedisCacheFactory implements ServerCacheFactory {
private static final Logger log = LoggerFactory.getLogger(RedisCacheFactory.class);
private static final Logger queryLogger = LoggerFactory.getLogger("io.ebean.cache.QUERY");
private static final Logger logger = LoggerFactory.getLogger("io.ebean.cache.CACHE");
private static final Logger tableModLogger = LoggerFactory.getLogger("io.ebean.cache.TABLEMODS");
private static final int MSG_NEARCACHE_CLEAR = 1;
private static final int MSG_NEARCACHE_KEYS = 2;
private static final int MSG_NEARCACHE_KEY = 3;
/**
* Channel for standard L2 cache messages.
*/
private static final String CHANNEL_L2 = "ebean.l2cache";
/**
* Channel specifically for near cache invalidation messages.
*/
private static final String CHANNEL_NEAR = "ebean.l2near";
private static final byte[] CHANNEL_L2_BYTES = SafeEncoder.encode(CHANNEL_L2);
private static final byte[] CHANNEL_NEAR_BYTES = SafeEncoder.encode(CHANNEL_NEAR);
private final ConcurrentHashMap<String, RQueryCache> queryCaches = new ConcurrentHashMap<>();
private final Map<String, NearCacheInvalidate> nearCacheMap = new ConcurrentHashMap<>();
private final EncodeManyIdsData encodeManyIdsData = new EncodeManyIdsData();
private final EncodeBeanData encodeBeanData = new EncodeBeanData();
private final EncodeSerializable encodeSerializable = new EncodeSerializable();
private final BackgroundExecutor executor;
private final JedisPool jedisPool;
private final DaemonTopicRunner daemonTopicRunner;
private final NearCacheNotify nearCacheNotify;
private final TimedMetric metricOutNearCache;
private final TimedMetric metricOutTableMod;
private final TimedMetric metricOutQueryCache;
private final TimedMetric metricInNearCache;
private final TimedMetric metricInTableMod;
private final TimedMetric metricInQueryCache;
private final String serverId = ModId.id();
private ServerCacheNotify listener;
RedisCacheFactory(DatabaseConfig config, BackgroundExecutor executor) {
this.executor = executor;
this.nearCacheNotify = new DNearCacheNotify();
MetricFactory factory = MetricFactory.get();
this.metricOutTableMod = factory.createTimedMetric("l2a.outTableMod");
this.metricOutQueryCache = factory.createTimedMetric("l2a.outQueryCache");
this.metricOutNearCache = factory.createTimedMetric("l2a.outNearKeys");
this.metricInTableMod = factory.createTimedMetric("l2a.inTableMod");
this.metricInQueryCache = factory.createTimedMetric("l2a.inQueryCache");
this.metricInNearCache = factory.createTimedMetric("l2a.inNearKeys");
if (config.isDisableL2Cache()) {
this.jedisPool = null;
this.daemonTopicRunner = null;
} else {
this.jedisPool = getJedisPool(config);
this.daemonTopicRunner = new DaemonTopicRunner(jedisPool, new CacheDaemonTopic());
daemonTopicRunner.run();
}
}
/**
* Return the JedisPool to use (only 1 at this stage).
*/
private JedisPool getJedisPool(DatabaseConfig config) {
JedisPool jedisPool = config.getServiceObject(JedisPool.class);
if (jedisPool != null) {
return jedisPool;
}
RedisConfig redisConfig = config.getServiceObject(RedisConfig.class);
if (redisConfig == null) {
redisConfig = new RedisConfig();
}
redisConfig.loadProperties(config.getProperties());
log.info("using l2cache redis host {}:{}", redisConfig.getServer(), redisConfig.getPort());
return redisConfig.createPool();
}
@Override
public void visit(MetricVisitor visitor) {
metricOutQueryCache.visit(visitor);
metricOutTableMod.visit(visitor);
metricOutNearCache.visit(visitor);
metricInTableMod.visit(visitor);
metricInQueryCache.visit(visitor);
metricInNearCache.visit(visitor);
}
@Override
public ServerCache createCache(ServerCacheConfig config) {
if (config.isQueryCache()) {
return createQueryCache(config);
}
return createNormalCache(config);
}
private ServerCache createNormalCache(ServerCacheConfig config) {
RedisCache redisCache = createRedisCache(config);
boolean nearCache = config.getCacheOptions().isNearCache();
if (!nearCache) {
return redisCache;
}
String cacheKey = config.getCacheKey();
DefaultServerCache near = new DefaultServerCache(new DefaultServerCacheConfig(config));
near.periodicTrim(executor);
DuelCache duelCache = new DuelCache(near, redisCache, cacheKey, nearCacheNotify);
nearCacheMap.put(cacheKey, duelCache);
return duelCache;
}
private RedisCache createRedisCache(ServerCacheConfig config) {
switch (config.getType()) {
case NATURAL_KEY:
return new RedisCache(jedisPool, config, encodeSerializable);
case BEAN:
return new RedisCache(jedisPool, config, encodeBeanData);
case COLLECTION_IDS:
return new RedisCache(jedisPool, config, encodeManyIdsData);
default:
throw new IllegalArgumentException("Unexpected cache type? " + config.getType());
}
}
private ServerCache createQueryCache(ServerCacheConfig config) {
synchronized (this) {
RQueryCache cache = queryCaches.get(config.getCacheKey());
if (cache == null) {
logger.debug("create query cache [{}]", config.getCacheKey());
cache = new RQueryCache(new DefaultServerCacheConfig(config));
cache.periodicTrim(executor);
queryCaches.put(config.getCacheKey(), cache);
}
return cache;
}
}
@Override
public ServerCacheNotify createCacheNotify(ServerCacheNotify listener) {
this.listener = listener;
return new RServerCacheNotify();
}
private void sendQueryCacheInvalidation(String name) {
long nanos = System.nanoTime();
try (Jedis resource = jedisPool.getResource()) {
resource.publish(CHANNEL_L2, serverId + ":queryCache:" + name);
} finally {
metricOutQueryCache.addSinceNanos(nanos);
}
}
private void sendTableMod(String formattedMsg) {
long nanos = System.nanoTime();
try (Jedis resource = jedisPool.getResource()) {
resource.publish(CHANNEL_L2, serverId + ":tableMod:" + formattedMsg);
} finally {
metricOutTableMod.addSinceNanos(nanos);
}
}
/**
* Query cache implementation using Redis channel for message notifications.
*/
private class RQueryCache extends DefaultServerQueryCache {
RQueryCache(DefaultServerCacheConfig config) {
super(config);
}
@Override
public void clear() {
super.clear();
sendQueryCacheInvalidation(name);
}
/**
* Process the invalidation message coming from the cluster.
*/
private void invalidate() {
queryLogger.debug(" CLEAR {}(*) - cluster invalidate", name);
super.clear();
}
}
/**
* Publish table modifications using Redis channel (to other cluster members)
*/
private class RServerCacheNotify implements ServerCacheNotify {
@Override
public void notify(ServerCacheNotification tableModifications) {
Set<String> dependentTables = tableModifications.getDependentTables();
if (dependentTables != null && !dependentTables.isEmpty()) {
StringBuilder msg = new StringBuilder(50);
for (String table : dependentTables) {
msg.append(table).append(",");
}
String formattedMsg = msg.toString();
if (tableModLogger.isDebugEnabled()) {
tableModLogger.debug("Publish TableMods - {}", formattedMsg);
}
sendTableMod(formattedMsg);
}
}
}
/**
* Clear the query cache if we have it.
*/
private void queryCacheInvalidate(String key) {
long nanos = System.nanoTime();
try {
RQueryCache queryCache = queryCaches.get(key);
if (queryCache != null) {
queryCache.invalidate();
}
} finally {
metricInQueryCache.addSinceNanos(nanos);
}
}
/**
* Process a remote dependent table modify event.
*/
private void processTableNotify(String rawMessage) {
long nanos = System.nanoTime();
try {
if (logger.isDebugEnabled()) {
logger.debug("processTableNotify {}", rawMessage);
}
Set<String> tables = new HashSet<>(asList(rawMessage.split(",")));
listener.notify(new ServerCacheNotification(tables));
} finally {
metricInTableMod.addSinceNanos(nanos);
}
}
/**
* Near cache notification using a specific Redis channel (CHANNEL_NEAR).
*/
private class DNearCacheNotify implements NearCacheNotify {
@Override
public void invalidateKeys(String cacheKey, Set<Object> keySet) {
try {
sendMessage(messageInvalidateKeys(cacheKey, keySet));
} catch (IOException e) {
logger.error("failed to transmit invalidateKeys() message", e);
}
}
@Override
public void invalidateKey(String cacheKey, Object id) {
try {
sendMessage(messageInvalidateKey(cacheKey, id));
} catch (IOException e) {
logger.error("failed to transmit invalidateKeys() message", e);
}
}
@Override
public void invalidateClear(String cacheKey) {
try {
sendMessage(messageInvalidateClear(cacheKey));
} catch (IOException e) {
logger.error("failed to transmit invalidateKeys() message", e);
}
}
private void sendMessage(byte[] message) {
long nanos = System.nanoTime();
try (Jedis resource = jedisPool.getResource()) {
resource.publish(CHANNEL_NEAR_BYTES, message);
} finally {
metricOutNearCache.addSinceNanos(nanos);
}
}
private byte[] messageInvalidateKeys(String cacheKey, Set<Object> keySet) throws IOException {
ByteArrayOutputStream ba = new ByteArrayOutputStream(100);
ObjectOutputStream os = new ObjectOutputStream(ba);
os.writeUTF(serverId);
os.writeInt(MSG_NEARCACHE_KEYS);
os.writeUTF(cacheKey);
os.writeInt(keySet.size());
for (Object key : keySet) {
os.writeObject(key);
}
os.flush();
os.close();
return ba.toByteArray();
}
private byte[] messageInvalidateKey(String cacheKey, Object id) throws IOException {
ByteArrayOutputStream ba = new ByteArrayOutputStream(100);
ObjectOutputStream os = new ObjectOutputStream(ba);
os.writeUTF(serverId);
os.writeInt(MSG_NEARCACHE_KEY);
os.writeUTF(cacheKey);
os.writeObject(id);
os.flush();
os.close();
return ba.toByteArray();
}
private byte[] messageInvalidateClear(String cacheKey) throws IOException {
ByteArrayOutputStream ba = new ByteArrayOutputStream(100);
ObjectOutputStream os = new ObjectOutputStream(ba);
os.writeUTF(serverId);
os.writeInt(MSG_NEARCACHE_CLEAR);
os.writeUTF(cacheKey);
os.flush();
os.close();
return ba.toByteArray();
}
}
/**
* Redis channel listener that supports reconnection etc.
*/
private class CacheDaemonTopic implements DaemonTopic {
@Override
public void subscribe(Jedis jedis) {
jedis.subscribe(new ChannelSubscriber(), CHANNEL_L2_BYTES, CHANNEL_NEAR_BYTES);
}
@Override
public void notifyConnected() {
logger.info("Established connection to Redis");
}
/**
* Handles updates to the features (via redis topic notifications).
*/
private class ChannelSubscriber extends BinaryJedisPubSub {
@Override
public void onMessage(byte[] channel, byte[] message) {
String channelName = SafeEncoder.encode(channel);
if (channelName.equals(CHANNEL_L2)) {
processL2Message(SafeEncoder.encode(message));
} else {
processNearCacheMessage(message);
}
}
private void processNearCacheMessage(byte[] message) {
long nanos = System.nanoTime();
int msgType = 0;
String cacheKey = null;
try {
ObjectInputStream oi = new ObjectInputStream(new ByteArrayInputStream(message));
String sourceServerId = oi.readUTF();
if (sourceServerId.equals(serverId)) {
// ignore this message as we are the server that sent it
return;
}
msgType = oi.readInt();
cacheKey = oi.readUTF();
if (logger.isDebugEnabled()) {
logger.debug("processNearCacheMessage serverId:{} type:{} cacheKey:{}", sourceServerId, msgType, cacheKey);
}
switch (msgType) {
case MSG_NEARCACHE_CLEAR:
nearCacheInvalidateClear(cacheKey);
break;
case MSG_NEARCACHE_KEY:
Object key = oi.readObject();
nearCacheInvalidateKey(cacheKey, key);
break;
case MSG_NEARCACHE_KEYS:
int count = oi.readInt();
Set<Object> keys = new LinkedHashSet<>();
for (int i = 0; i < count; i++) {
keys.add(oi.readObject());
}
nearCacheInvalidateKeys(cacheKey, keys);
break;
default:
throw new IllegalStateException("Unexpected message type ? " + msgType);
}
} catch (IOException | ClassNotFoundException e) {
logger.error("failed to decode near cache message [" + SafeEncoder.encode(message) + "] for cache:" + cacheKey, e);
if (cacheKey != null) {
nearCacheInvalidateClear(cacheKey);
}
} finally {
if (msgType != 0) {
metricInNearCache.addSinceNanos(nanos);
}
}
}
private void processL2Message(String message) {
try {
String[] split = message.split(":");
if (serverId.equals(split[0])) {
// ignore this message as we are the server that sent it
return;
}
switch (split[1]) {
case "tableMod":
processTableNotify(split[2]);
break;
case "queryCache":
queryCacheInvalidate(split[2]);
break;
default:
logger.error("Unknown L2 message type[{}] on redis channel - message[{}] ", split[0], message);
}
} catch (Exception e) {
logger.error("Error handling L2 message[" + message + "]", e);
}
}
}
}
/**
* Invalidate key for a local near cache.
*/
private void nearCacheInvalidateKey(String cacheKey, Object key) {
NearCacheInvalidate invalidate = nearCacheMap.get(cacheKey);
if (invalidate == null) {
warnNearCacheNotFound(cacheKey);
} else {
invalidate.invalidateKey(key);
}
}
/**
* Invalidate keys for a local near cache.
*/
private void nearCacheInvalidateKeys(String cacheKey, Set<Object> keys) {
NearCacheInvalidate invalidate = nearCacheMap.get(cacheKey);
if (invalidate == null) {
warnNearCacheNotFound(cacheKey);
} else {
invalidate.invalidateKeys(keys);
}
}
/**
* Invalidate clear for a local near cache.
*/
private void nearCacheInvalidateClear(String cacheKey) {
NearCacheInvalidate invalidate = nearCacheMap.get(cacheKey);
if (invalidate == null) {
warnNearCacheNotFound(cacheKey);
} else {
invalidate.invalidateClear();
}
}
private void warnNearCacheNotFound(String cacheKey) {
logger.warn("No near cache found for cacheKey [" + cacheKey + "] yet - probably on startup");
}
}
ebean-redis/src/main/java/io/ebean/redis/RedisCachePlugin.java
+17
View File
@@ -0,0 +1,17 @@
package io.ebean.redis;
import io.ebean.BackgroundExecutor;
import io.ebean.cache.ServerCacheFactory;
import io.ebean.cache.ServerCachePlugin;
import io.ebean.config.DatabaseConfig;
public class RedisCachePlugin implements ServerCachePlugin {
/**
* Create the ServerCacheFactory implementation.
*/
@Override
public ServerCacheFactory create(DatabaseConfig config, BackgroundExecutor executor) {
return new RedisCacheFactory(config, executor);
}
}
ebean-redis/src/main/java/io/ebean/redis/RedisConfig.java
+179
View File
@@ -0,0 +1,179 @@
package io.ebean.redis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import java.util.Properties;
/**
* Deployment configuration for redis.
*/
public class RedisConfig {
private String server = "localhost";
private int port = 6379;
private int maxTotal = 200;
private int maxIdle = 200;
private int minIdle = 1;
private long maxWaitMillis = -1L;
private boolean blockWhenExhausted = true;
private int timeout = 2000;
private String username;
private String password;
private boolean ssl;
/**
* Return a new JedisPool based on the configuration.
*/
public JedisPool createPool() {
JedisPoolConfig poolConfig = new JedisPoolConfig();
poolConfig.setMaxTotal(maxTotal);
poolConfig.setMaxIdle(maxIdle);
poolConfig.setMinIdle(minIdle);
poolConfig.setMaxWaitMillis(maxWaitMillis);
poolConfig.setBlockWhenExhausted(blockWhenExhausted);
return new JedisPool(poolConfig, server, port, timeout, username, password, ssl);
}
public String getServer() {
return server;
}
public void setServer(String server) {
this.server = server;
}
public int getPort() {
return port;
}
public void setPort(int port) {
this.port = port;
}
public int getMaxTotal() {
return maxTotal;
}
public void setMaxTotal(int maxTotal) {
this.maxTotal = maxTotal;
}
public int getMaxIdle() {
return maxIdle;
}
public void setMaxIdle(int maxIdle) {
this.maxIdle = maxIdle;
}
public int getMinIdle() {
return minIdle;
}
public void setMinIdle(int minIdle) {
this.minIdle = minIdle;
}
public long getMaxWaitMillis() {
return maxWaitMillis;
}
public void setMaxWaitMillis(long maxWaitMillis) {
this.maxWaitMillis = maxWaitMillis;
}
public boolean isBlockWhenExhausted() {
return blockWhenExhausted;
}
public void setBlockWhenExhausted(boolean blockWhenExhausted) {
this.blockWhenExhausted = blockWhenExhausted;
}
public int getTimeout() {
return timeout;
}
public void setTimeout(int timeout) {
this.timeout = timeout;
}
public String getUsername() {
return username;
}
public void setUsername(String username) {
this.username = username;
}
public String getPassword() {
return password;
}
public void setPassword(String password) {
this.password = password;
}
public boolean isSsl() {
return ssl;
}
public void setSsl(boolean ssl) {
this.ssl = ssl;
}
public void loadProperties(Properties properties) {
Reader reader = new Reader(properties);
this.server = reader.get("ebean.redis.server", server);
this.port = reader.getInt("ebean.redis.port", port);
this.ssl = reader.getBool("ebean.redis.ssl", ssl);
this.minIdle = reader.getInt("ebean.redis.minIdle", minIdle);
this.maxIdle = reader.getInt("ebean.redis.maxIdle", maxIdle);
this.maxTotal = reader.getInt("ebean.redis.maxTotal", maxTotal);
this.maxWaitMillis = reader.getLong("ebean.redis.maxWaitMillis", maxWaitMillis);
this.timeout = reader.getInt("ebean.redis.timeout", timeout);
this.username = reader.get("ebean.redis.username", username);
this.password = reader.get("ebean.redis.password", password);
this.blockWhenExhausted = reader.getBool("ebean.redis.blockWhenExhausted", blockWhenExhausted);
}
private static class Reader {
private final Properties properties;
Reader(Properties properties) {
this.properties = (properties != null) ? properties : new Properties();
}
String get(String key, String defaultVal) {
return System.getProperty(key, properties.getProperty(key, defaultVal));
}
int getInt(String key, int defaultVal) {
final String val = get(key, null);
return val != null ? Integer.parseInt(val.trim()) : defaultVal;
}
long getLong(String key, long defaultVal) {
final String val = get(key, null);
return val != null ? Integer.parseInt(val.trim()) : defaultVal;
}
boolean getBool(String key, boolean defaultVal) {
final String val = get(key, null);
return val != null ? Boolean.parseBoolean(val.trim()) : defaultVal;
}
}
}
ebean-redis/src/main/java/io/ebean/redis/encode/Encode.java
+8
View File
@@ -0,0 +1,8 @@
package io.ebean.redis.encode;
public interface Encode {
byte[] encode(Object value);
Object decode(byte[] data);
}
ebean-redis/src/main/java/io/ebean/redis/encode/EncodeBeanData.java
+47
View File
@@ -0,0 +1,47 @@
package io.ebean.redis.encode;
import io.ebeaninternal.server.cache.CachedBeanData;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
public class EncodeBeanData implements Encode {
@Override
public byte[] encode(Object value) {
try {
CachedBeanData data = (CachedBeanData) value;
ByteArrayOutputStream os = new ByteArrayOutputStream();
ObjectOutputStream oos = new ObjectOutputStream(os);
data.writeExternal(oos);
oos.flush();
oos.close();
return os.toByteArray();
} catch (IOException e) {
throw new RuntimeException("Failed to decode cache data", e);
}
}
@Override
public Object decode(byte[] data) {
try {
ByteArrayInputStream is = new ByteArrayInputStream(data);
ObjectInputStream ois = new ObjectInputStream(is);
CachedBeanData read = new CachedBeanData();
read.readExternal(ois);
return read;
} catch (ClassNotFoundException | IOException e) {
throw new RuntimeException("Failed to decode cache data", e);
}
}
}
ebean-redis/src/main/java/io/ebean/redis/encode/EncodeManyIdsData.java
+47
View File
@@ -0,0 +1,47 @@
package io.ebean.redis.encode;
import io.ebeaninternal.server.cache.CachedManyIds;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
public class EncodeManyIdsData implements Encode {
@Override
public byte[] encode(Object value) {
try {
CachedManyIds data = (CachedManyIds) value;
ByteArrayOutputStream os = new ByteArrayOutputStream();
ObjectOutputStream oos = new ObjectOutputStream(os);
data.writeExternal(oos);
oos.flush();
oos.close();
return os.toByteArray();
} catch (IOException e) {
throw new RuntimeException("Failed to decode cache data", e);
}
}
@Override
public Object decode(byte[] data) {
try {
ByteArrayInputStream is = new ByteArrayInputStream(data);
ObjectInputStream ois = new ObjectInputStream(is);
CachedManyIds read = new CachedManyIds();
read.readExternal(ois);
return read;
} catch (ClassNotFoundException | IOException e) {
throw new RuntimeException("Failed to decode cache data", e);
}
}
}
ebean-redis/src/main/java/io/ebean/redis/encode/EncodePrefixKey.java
+41
View File
@@ -0,0 +1,41 @@
package io.ebean.redis.encode;
import io.ebean.cache.TenantAwareKey;
import java.nio.charset.StandardCharsets;
public class EncodePrefixKey implements Encode {
private final String prefix;
public EncodePrefixKey(String cacheKey) {
this.prefix = cacheKey + ":";
}
@Override
public byte[] encode(Object value) {
try {
if (!(value instanceof String) && !(value instanceof TenantAwareKey.CacheKey)) {
throw new IllegalStateException("Expecting String keys but got type:" + value.getClass());
}
String key = prefix + value.toString();
return key.getBytes(StandardCharsets.UTF_8);
} catch (Exception e) {
throw new RuntimeException("Failed to decode cache data", e);
}
}
@Override
public Object decode(byte[] data) {
try {
String key = new String(data, StandardCharsets.UTF_8);
int pos = key.indexOf(':');
return key.substring(pos);
} catch (Exception e) {
throw new RuntimeException("Failed to decode cache data", e);
}
}
}
ebean-redis/src/main/java/io/ebean/redis/encode/EncodeSerializable.java
+37
View File
@@ -0,0 +1,37 @@
package io.ebean.redis.encode;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
public class EncodeSerializable implements Encode {
@Override
public byte[] encode(Object value) {
try {
ObjectOutputStream oos = new ObjectOutputStream(new ByteArrayOutputStream());
oos.writeObject(value);
oos.flush();
oos.close();
return new ByteArrayOutputStream().toByteArray();
} catch (IOException e) {
throw new RuntimeException("Failed to decode cache data", e);
}
}
@Override
public Object decode(byte[] data) {
try {
try (ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(data))) {
return ois.readObject();
}
} catch (ClassNotFoundException | IOException e) {
throw new RuntimeException("Failed to decode cache data", e);
}
}
}
ebean-redis/src/main/java/io/ebean/redis/encode/PrefixKey.java
+22
View File
@@ -0,0 +1,22 @@
package io.ebean.redis.encode;
import java.io.Serializable;
public class PrefixKey implements Serializable {
private final String prefix;
private final Object key;
PrefixKey(String prefix, Object key) {
this.prefix = prefix;
this.key = key;
}
public String getPrefix() {
return prefix;
}
public Object getKey() {
return key;
}
}
ebean-redis/src/main/java/io/ebean/redis/topic/DaemonTopic.java
+21
View File
@@ -0,0 +1,21 @@
package io.ebean.redis.topic;
import redis.clients.jedis.Jedis;
/**
* Topic subscriber that has re-connect notification.
*/
public interface DaemonTopic {
/**
* Subscribe to the topic/channel (blocking).
*
* @param jedis The redis connection to subscribe (and block on).
*/
void subscribe(Jedis jedis);
/**
* Notify that the topic subscription has been connected (or reconnected).
*/
void notifyConnected();
}
ebean-redis/src/main/java/io/ebean/redis/topic/DaemonTopicRunner.java
+95
View File
@@ -0,0 +1,95 @@
package io.ebean.redis.topic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.exceptions.JedisException;
import java.util.Timer;
import java.util.TimerTask;
/**
* Subscribe to redis topic listening for changes.
* <p>
* Handles reconnection to redis when the connection is lost and
* notification of when reconnection takes place.
* </p>
*/
public class DaemonTopicRunner {
private static final Logger log = LoggerFactory.getLogger(DaemonTopicRunner.class);
private static final long reconnectWaitMillis = 1000;
private final JedisPool jedisPool;
private final DaemonTopic daemonTopic;
public DaemonTopicRunner(JedisPool jedisPool, DaemonTopic daemonTopic) {
this.jedisPool = jedisPool;
this.daemonTopic = daemonTopic;
}
public void run() {
Thread t = new Thread(() -> attemptConnections(), "redis-sub");
t.start();
}
private void attemptConnections() {
Timer reloadTimer = new Timer("redis-sub-notify");
ReloadNotifyTask notifyTask = null;
int attempts = 1;
while (true) {
if (notifyTask != null) {
// we didn't successfully re-connect to redis
notifyTask.cancel();
}
notifyTask = new ReloadNotifyTask();
reloadTimer.schedule(notifyTask, reconnectWaitMillis + 500);
attempts++;
try {
subscribe();
} catch (JedisException e) {
log.debug("... redis subscribe connection attempt:{} failed:{}", attempts, e.getMessage());
try {
// wait a little before retrying
Thread.sleep(reconnectWaitMillis);
} catch (InterruptedException e1) {
log.warn("Interrupted redis re-connection wait", e1);
}
}
}
}
/**
* Subscribe and block when successful.
*/
private void subscribe() {
Jedis jedis = jedisPool.getResource();
jedis.echo("hi");
try {
daemonTopic.subscribe(jedis);
} catch (Exception e) {
log.error("Lost connection to topic, starting re-connection loop", e);
attemptConnections();
} finally {
try {
jedis.close();
} catch (Exception e) {
log.warn("Error closing probably broken Redis connection", e);
}
}
}
private class ReloadNotifyTask extends TimerTask {
@Override
public void run() {
daemonTopic.notifyConnected();
}
}
}
ebean-redis/src/main/resources/META-INF/services/io.ebean.cache.ServerCachePlugin
+1
View File
@@ -0,0 +1 @@
io.ebean.redis.RedisCachePlugin
ebean-redis/src/test/java/io/ebean/redis/RedisConfigTest.java
+46
View File
@@ -0,0 +1,46 @@
package io.ebean.redis;
import org.junit.Test;
import java.util.Properties;
import static org.assertj.core.api.Assertions.assertThat;
public class RedisConfigTest {
@Test
public void loadProperties() {
Properties p = new Properties();
p.setProperty("ebean.redis.server", "test-server");
p.setProperty("ebean.redis.port", "99");
p.setProperty("ebean.redis.maxIdle", "5");
p.setProperty("ebean.redis.maxTotal", "6");
p.setProperty("ebean.redis.minIdle", "7");
p.setProperty("ebean.redis.maxWaitMillis", "8");
p.setProperty("ebean.redis.username", "un");
p.setProperty("ebean.redis.password", "pw");
p.setProperty("ebean.redis.ssl", "true");
RedisConfig config = new RedisConfig();
config.loadProperties(p);
assertThat(config.getServer()).isEqualTo("test-server");
assertThat(config.getPort()).isEqualTo(99);
assertThat(config.getMaxIdle()).isEqualTo(5);
assertThat(config.getMaxTotal()).isEqualTo(6);
assertThat(config.getMinIdle()).isEqualTo(7);
assertThat(config.getMaxWaitMillis()).isEqualTo(8);
assertThat(config.getUsername()).isEqualTo("un");
assertThat(config.getPassword()).isEqualTo("pw");
assertThat(config.isSsl()).isEqualTo(true);
}
@Test
public void test_defaultValues() {
RedisConfig config = new RedisConfig();
assertThat(config.getUsername()).isNull();
assertThat(config.getPassword()).isNull();
assertThat(config.isSsl()).isEqualTo(false);
}
}
ebean-redis/src/test/java/org/domain/EBase.java
+58
View File
@@ -0,0 +1,58 @@
package org.domain;
import io.ebean.Model;
import io.ebean.annotation.WhenCreated;
import io.ebean.annotation.WhenModified;
import javax.persistence.Id;
import javax.persistence.MappedSuperclass;
import javax.persistence.Version;
import java.time.Instant;
@MappedSuperclass
public class EBase extends Model {
@Id
protected long id;
@Version
protected long version;
@WhenCreated
protected Instant whenCreated;
@WhenModified
protected Instant whenModified;
public long getId() {
return id;
}
public void setId(long id) {
this.id = id;
}
public long getVersion() {
return version;
}
public void setVersion(long version) {
this.version = version;
}
public Instant getWhenCreated() {
return whenCreated;
}
public void setWhenCreated(Instant whenCreated) {
this.whenCreated = whenCreated;
}
public Instant getWhenModified() {
return whenModified;
}
public void setWhenModified(Instant whenModified) {
this.whenModified = whenModified;
}
}
ebean-redis/src/test/java/org/domain/Person.java
+69
View File
@@ -0,0 +1,69 @@
package org.domain;
import io.ebean.annotation.Cache;
import io.ebean.annotation.Index;
import javax.persistence.Entity;
import java.time.LocalDate;
@Cache(enableQueryCache = true, nearCache = true, naturalKey = "name")
@Entity
public class Person extends EBase {
public enum Status {
NEW,
ACTIVE,
INACTIVE
}
@Index(unique = true)
String name;
Status status;
LocalDate localDate;
String notes;
public Person(String name) {
this.name = name;
this.status = Status.NEW;
}
public String toString() {
return "[id:" + id + " name:" + name + "date:" + localDate + ']';
}
public Status getStatus() {
return status;
}
public void setStatus(Status status) {
this.status = status;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getNotes() {
return notes;
}
public void setNotes(String notes) {
this.notes = notes;
}
public LocalDate getLocalDate() {
return localDate;
}
public void setLocalDate(LocalDate localDate) {
this.localDate = localDate;
}
}
ebean-redis/src/test/java/org/integration/ClusterTest.java
+123
View File
@@ -0,0 +1,123 @@
package org.integration;
import io.ebean.redis.DuelCache;
import org.domain.Person;
import io.ebean.DB;
import io.ebean.Database;
import io.ebean.DatabaseFactory;
import io.ebean.config.DatabaseConfig;
import org.domain.query.QPerson;
import org.junit.Test;
import javax.sql.DataSource;
import static org.assertj.core.api.Assertions.assertThat;
public class ClusterTest {
private Database createOther(DataSource dataSource) {
DatabaseConfig config = new DatabaseConfig();
config.setDataSource(dataSource);
config.loadFromProperties();
config.setDefaultServer(false);
config.setName("other");
config.setDdlGenerate(false);
config.setDdlRun(false);
return DatabaseFactory.create(config);
}
@Test
public void testBothNear() {
// ensure the default server exists first
final Database db = DB.getDefault();
Database other = createOther(db.getPluginApi().getDataSource());
new QPerson()
.name.eq("Someone")
.delete();
Person foo = new Person("Someone");
foo.save();
DB.getServerCacheManager().clearAll();
other.getMetaInfoManager().resetAllMetrics();
Person fooA = DB.find(Person.class, foo.getId());
Person fooB = other.find(Person.class, foo.getId());
fooA = DB.find(Person.class, foo.getId());
fooB = other.find(Person.class, foo.getId());
fooA = DB.find(Person.class, foo.getId());
fooB = other.find(Person.class, foo.getId());
DuelCache dualCache = (DuelCache) other.getServerCacheManager().getBeanCache(Person.class);
assertCounts(dualCache, 2, 1, 1, 0);
}
@Test
public void test() throws InterruptedException {
// ensure the default server exists first
final Database db = DB.getDefault();
Database other = createOther(db.getPluginApi().getDataSource());
for (int i = 0; i < 10; i++) {
Person foo = new Person("name " + i);
foo.save();
}
other.getServerCacheManager().clearAll();
other.getMetaInfoManager().resetAllMetrics();
DuelCache dualCache = (DuelCache) other.getServerCacheManager().getBeanCache(Person.class);
Person foo0 = other.find(Person.class, 1);
assertCounts(dualCache, 0, 1, 0, 1);
other.find(Person.class, 1);
assertCounts(dualCache, 1, 1, 0, 1);
other.find(Person.class, 1);
assertCounts(dualCache, 2, 1, 0, 1);
other.find(Person.class, 1);
assertCounts(dualCache, 3, 1, 0, 1);
other.find(Person.class, 2);
assertCounts(dualCache, 3, 2, 0, 2);
foo0.setName("name2");
foo0.save();
allowAsyncMessaging();
Person foo3 = other.find(Person.class, 1);
assertThat(foo3.getName()).isEqualTo("name2");
assertCounts(dualCache, 3, 3, 1, 2);
foo0.setName("name3");
foo0.save();
allowAsyncMessaging();
foo3 = other.find(Person.class, 1);
assertThat(foo3.getName()).isEqualTo("name3");
assertCounts(dualCache, 3, 4, 2, 2);
}
private void assertCounts(DuelCache dualCache, int nearHits, int nearMiss, int remoteHit, int remoteMiss) {
assertThat(dualCache.getNearHitCount()).isEqualTo(nearHits);
assertThat(dualCache.getNearMissCount()).isEqualTo(nearMiss);
assertThat(dualCache.getRemoteHitCount()).isEqualTo(remoteHit);
assertThat(dualCache.getRemoteMissCount()).isEqualTo(remoteMiss);
}
private void allowAsyncMessaging() throws InterruptedException {
Thread.sleep(10);
}
}
ebean-redis/src/test/java/org/integration/IntegrationTest.java
+107
View File
@@ -0,0 +1,107 @@
package org.integration;
import io.ebean.DB;
import org.domain.Person;
import org.domain.query.QPerson;
import org.junit.Test;
import java.time.LocalDate;
import java.util.ArrayList;
import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
public class IntegrationTest {
@Test
public void test() throws InterruptedException {
insertSomePeople();
Person fiona = findByName("Fiona");
fiona.setName("Fortuna");
fiona.setLocalDate(LocalDate.now());
fiona.update();
Thread.sleep(100);
Person one = findById(1);
assertThat(one).isNotNull();
for (int i = 1; i < 4; i++) {
System.out.println("hit " + findById(i));
}
List<Person> one2 = nameStartsWith("fo");
assertThat(one2).hasSize(1);
one2 = nameStartsWith("j");
assertThat(one2).hasSize(2);
one2 = nameStartsWith("j");
assertThat(one2).hasSize(2);
List<Person> byNames = findByNames("Jack", "Rob");
assertThat(byNames).hasSize(2);
byNames = findByNames("Jack", "Rob", "Moby");
assertThat(byNames).hasSize(3);
fiona.setName("fo2");
fiona.setLocalDate(LocalDate.now());
fiona.update();
byNames = findByNames("Jack", "Rob", "Moby");
assertThat(byNames).hasSize(3);
Thread.sleep(200);
one2 = nameStartsWith("fo%");
System.out.println("one2 " + one2);
one2 = nameStartsWith("f0%");
System.out.println("one2 " + one2);
DB.getServerCacheManager().clear(Person.class);
System.out.println("done");
}
private List<Person> insertSomePeople() {
List<Person> people = new ArrayList<>();
for (String name : new String[]{"Jack", "John", "Rob", "Moby", "Fiona"}) {
people.add(new Person(name));
}
DB.saveAll(people);
return people;
}
private Person findByName(String name) {
return new QPerson()
.name.eq(name)
.findOne();
}
private List<Person> findByNames(String... names) {
return new QPerson()
.name.in(names)
.setUseCache(true)
.findList();
}
private Person findById(int id) {
return new QPerson()
.id.eq(id)
.findOne();
}
private List<Person> nameStartsWith(String pattern) {
return new QPerson()
.name.istartsWith(pattern)
.setUseQueryCache(true)
.findList();
}
}
ebean-redis/src/test/resources/application-test.yaml
+9
View File
@@ -0,0 +1,9 @@
ebean:
dumpMetricsOnShutdown: true
dumpMetricsOptions: sql,hash,loc
test:
redis: latest
# shutdown: stop # stop | remove
platform: h2 # h2, postgres, mysql, oracle, sqlserver, sqlite
ddlMode: dropCreate # none | dropCreate | create | migration | createOnly | migrationDropCreate
dbName: myapp
ebean-redis/src/test/resources/logback-test.xml
+29
View File
@@ -0,0 +1,29 @@
<configuration scan="true" scanPeriod="10 seconds">
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>TRACE</level>
</filter>
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
</encoder>
</appender>
<root level="WARN">
<appender-ref ref="STDOUT"/>
</root>
<!--<logger name="com.avaje.ebeaninternal.server.cluster" level="TRACE"/>-->
<!-- <logger name="io.ebean" level="DEBUG"/>-->
<logger name="io.ebean.SQL" level="TRACE"/>
<logger name="io.ebean.TXN" level="TRACE"/>
<logger name="io.ebean.SUM" level="TRACE"/>
<logger name="io.ebean.cache" level="TRACE"/>
<logger name="io.ebean.cache.QUERY" level="TRACE"/>
<logger name="io.ebean.cache.BEAN" level="TRACE"/>
<logger name="io.ebean.cache.COLL" level="TRACE"/>
<logger name="io.ebean.cache.NATKEY" level="TRACE"/>
</configuration>
pom.xml
+1
View File
@@ -82,6 +82,7 @@
<module>kotlin-querybean-generator</module>
<module>ebean-querybean</module>
<module>ebean-postgis</module>
<module>ebean-redis</module>
</modules>
</project>