Merge pull request #243 from scubasau/master

DNS monitoring for SingleServerConnection

src/main/java/org/redisson/SingleServerConfig.java

@@ -38,6 +38,21 @@ public class SingleServerConfig extends BaseConfig<SingleServerConfig> {
      */
     private int connectionPoolSize = 100;
 
+        
+    /**
+     * Should the server address be monitored for changes in DNS? Useful for 
+     * AWS ElastiCache where the client is pointed at the endpoint for a replication group
+     * which is a DNS alias to the current master node.<br>
+     * <em>NB: applications must ensure the JVM DNS cache TTL is low enough to support this.</em> 
+     * e.g., http://docs.aws.amazon.com/AWSSdkDocsJava/latest/DeveloperGuide/java-dg-jvm-ttl.html
+     */
+    private boolean dnsMonitoring = false;
+
+    /**
+     * Interval in milliseconds to check DNS
+     */
+    private long dnsMonitoringInterval = 5000;
+    
     SingleServerConfig() {
     }
 
@@ -46,6 +61,8 @@ public class SingleServerConfig extends BaseConfig<SingleServerConfig> {
         setAddress(config.getAddress());
         setConnectionPoolSize(config.getConnectionPoolSize());
         setSubscriptionConnectionPoolSize(config.getSubscriptionConnectionPoolSize());
+        setDnsMonitoring(config.isDnsMonitoring());
+        setDnsMonitoringInterval(config.getDnsMonitoringInterval());
     }
 
     /**
@@ -93,4 +110,33 @@ public class SingleServerConfig extends BaseConfig<SingleServerConfig> {
         this.address = address;
     }
 
+    /**
+     * Monitoring of the endpoint address for DNS changes.
+     * Default is false.
+     * 
+     * @param dnsMonitoring
+     * @return
+     */
+    public SingleServerConfig setDnsMonitoring(boolean dnsMonitoring) {
+        this.dnsMonitoring = dnsMonitoring;
+        return this;
+    }
+    public boolean isDnsMonitoring() {
+        return dnsMonitoring;
+    }
+
+    /**
+     * Interval in milliseconds to check the endpoint DNS if {@link #isDnsMonitoring()} is true.
+     * Default is 5000.
+     * 
+     * @param dnsMonitoringInterval
+     * @return
+     */
+    public SingleServerConfig setDnsMonitoringInterval(long dnsMonitoringInterval) {
+        this.dnsMonitoringInterval = dnsMonitoringInterval;
+        return this;
+    }
+    public long getDnsMonitoringInterval() {
+        return dnsMonitoringInterval;
+    }
 }

src/main/java/org/redisson/connection/SingleConnectionManager.java

@@ -15,12 +15,29 @@
  */
 package org.redisson.connection;
 
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
 import org.redisson.Config;
 import org.redisson.MasterSlaveServersConfig;
 import org.redisson.SingleServerConfig;
+import org.redisson.client.RedisConnectionException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.util.concurrent.GlobalEventExecutor;
+import io.netty.util.concurrent.ScheduledFuture;
 
 public class SingleConnectionManager extends MasterSlaveConnectionManager {
 
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    private final AtomicReference<InetAddress> currentMaster = new AtomicReference<InetAddress>();
+
+    private ScheduledFuture<?> monitorFuture;
+    
     public SingleConnectionManager(SingleServerConfig cfg, Config config) {
         MasterSlaveServersConfig newconfig = new MasterSlaveServersConfig();
         String addr = cfg.getAddress().getHost() + ":" + cfg.getAddress().getPort();
@@ -37,6 +54,16 @@ public class SingleConnectionManager extends MasterSlaveConnectionManager {
         newconfig.setSlaveSubscriptionConnectionPoolSize(cfg.getSubscriptionConnectionPoolSize());
 
         init(newconfig, config);
+        
+        if (cfg.isDnsMonitoring()) {
+            try {
+                this.currentMaster.set(InetAddress.getByName(cfg.getAddress().getHost()));
+            } catch (UnknownHostException e) {
+                throw new RedisConnectionException("Unknown host", e);
+            }
+            log.debug("DNS monitoring enabled; Current master set to {}", currentMaster.get());
+            monitorDnsChange(cfg);
+        }
     }
 
     @Override
@@ -45,4 +72,35 @@ public class SingleConnectionManager extends MasterSlaveConnectionManager {
         entries.put(MAX_SLOT, entry);
     }
         
+    private void monitorDnsChange(final SingleServerConfig cfg) {
+        monitorFuture = GlobalEventExecutor.INSTANCE.scheduleWithFixedDelay(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    InetAddress master = currentMaster.get();
+                    InetAddress now = InetAddress.getByName(cfg.getAddress().getHost());
+                    if (!now.getHostAddress().equals(master.getHostAddress())) {
+                        log.info("Detected DNS change. {} has changed from {} to {}", cfg.getAddress().getHost(), master.getHostAddress(), now.getHostAddress());
+                        if (currentMaster.compareAndSet(master, now)) {
+                            changeMaster(MAX_SLOT,cfg.getAddress().getHost(), cfg.getAddress().getPort());
+                            log.info("Master has been changed");
+                        }
+                    }
+
+                } catch (Exception e) {
+                    log.error(e.getMessage(), e);
+                }
+
+            }
+
+        }, cfg.getDnsMonitoringInterval(), cfg.getDnsMonitoringInterval(), TimeUnit.MILLISECONDS);
+    }
+
+    @Override
+    public void shutdown() {
+        if (monitorFuture != null) {
+            monitorFuture.cancel(true);
+        }
+        super.shutdown();
+    }
 }