Refactor - tidy the near cache sending messages

Author: rbygrave

pom.xml

@@ -28,7 +28,7 @@
     <dependency>
       <groupId>io.ebean</groupId>
       <artifactId>ebean</artifactId>
-      <version>11.31.1-SNAPSHOT</version>
+      <version>11.31.1</version>
       <scope>provided</scope>
     </dependency>
 

src/main/java/io/ebean/redis/RedisCacheFactory.java

@@ -273,74 +273,79 @@ private class DNearCacheNotify implements NearCacheNotify {
 
     @Override
     public void invalidateKeys(String cacheKey, Set<Object> keySet) {
-
-      long nanos = System.nanoTime();
-      try (Jedis resource = jedisPool.getResource()) {
-        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();
-
-        resource.publish(CHANNEL_NEAR_BYTES, ba.toByteArray());
-
+      try {
+        sendMessage(messageInvalidateKeys(cacheKey, keySet));
       } catch (IOException e) {
         logger.error("failed to transmit invalidateKeys() message", e);
-      } finally {
-        metricMsgOut.add((System.nanoTime() - nanos) / 1000);
       }
     }
 
     @Override
     public void invalidateKey(String cacheKey, Object id) {
-
-      long nanos = System.nanoTime();
-      try (Jedis resource = jedisPool.getResource()) {
-        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();
-
-        resource.publish(CHANNEL_NEAR_BYTES, ba.toByteArray());
-
+      try {
+        sendMessage(messageInvalidateKey(cacheKey, id));
       } catch (IOException e) {
         logger.error("failed to transmit invalidateKeys() message", e);
-      } finally {
-        metricMsgOut.addSinceNanos(nanos);
       }
     }
 
     @Override
     public void invalidateClear(String cacheKey) {
-
-      long nanos = System.nanoTime();
-      try (Jedis resource = jedisPool.getResource()) {
-        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();
-
-        resource.publish(CHANNEL_NEAR_BYTES, ba.toByteArray());
-
+      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 {
+        try (Jedis resource = jedisPool.getResource()) {
+          resource.publish(CHANNEL_NEAR_BYTES, message);
+        }
       } finally {
         metricMsgOut.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();
+    }
   }
 
   /**

src/test/java/io/ebean/redis/ClusterTest.java

@@ -1,6 +1,7 @@
 package io.ebean.redis;
 
 import domain.EFoo;
+import io.ebean.Ebean;
 import io.ebean.EbeanServer;
 import io.ebean.EbeanServerFactory;
 import io.ebean.config.ServerConfig;
@@ -26,6 +27,9 @@ private EbeanServer createOther() {
   @Test
   public void test() throws InterruptedException {
 
+    // ensure the default server exists first
+    Ebean.getDefaultServer();
+
     EbeanServer other = createOther();
 
     for (int i = 0; i < 10; i++) {