connection monitoring and auto-reconnect

/*
 *   Copyright 2010 Joubin Houshyar
 * 
 *   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 test.net;

import java.io.BufferedReader;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.ConnectException;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;

/**
 * Demo a fault-tolerant connection.
 * Both dropped connections or server downtime are handled.
 * 
 * @author  joubin ([email protected])
 * @date    Jul 17, 2010
 * 
 */
public class FaultTolerantConnector {
	/** Run the test */
	public static void main (String[] args) {
		try {
			new FaultTolerantConnector().run();
		}
		catch (ConnectException e) { 
			System.err.format("Failed to connect to server: %s\n", e);        
			e.printStackTrace();
		}
		catch (Throwable e) { 
			System.err.format("Fault: %s\n", e);        
			e.printStackTrace();
		}
	}
	// ========================================================================
	
	String password;
	Socket s = null;
	DataOutputStream dout = null;
	BufferedReader din = null;
	volatile long lastchecktime;
	
	/**
	 * Perpetually runs a INCR sequence.  Delay between requests is chosen to be
	 * much bigger than the redis.conf:timeout settings.  (Tested with 1 sec).
	 * 
	 * @throws Exception
	 */
	public void run() throws Exception {
		
		long maxDowntime = 1000 * 60; 
		long delay = 5000; // redis.conf::timeout=1 // 1 second
		long redismaxidle = 1000; 
		
		password = "jredis";  // your server's password.  leave at null to skip AUTH
		connect();
		
		long n = 0;
		long reps = 10000;
		long start = System.currentTimeMillis();
		for(;;){
			
			// make sure connection is alive
			// tolerate up to maxDowntime msecs while Redis comes back up
			//
			if(System.currentTimeMillis() - lastchecktime > redismaxidle)
				assertConnectionEstablished(maxDowntime, false);
			
			// do something destructive 
			//
			INCR("my-cntr");
			n++;
			
			if(n > reps) {
				long delta = System.currentTimeMillis() - start;
				System.out.format("%d INCRs in %d msecs\n", reps, delta);
				
				// delay to force idle client disconnect 
				// (if delay >> redis.conf::timeout)
				takeANap (delay);
				
				start = System.currentTimeMillis();
				n = 0;
			}
		}
	}
	
	private final void assertConnectionEstablished (long maxDowntime, boolean usePings) throws Exception{
		long start = System.currentTimeMillis();
		for(;;) {
			try {
				if(usePings)
					testConnectionWithPing(); 
				else
					testConnection(); 
				
				lastchecktime = System.currentTimeMillis();
				break;
			}
			catch (Exception e){
				long downtime = System.currentTimeMillis() - start;
				if(downtime > maxDowntime){
					String errmsg = String.format("Could not re-establish connection -- server down more than %d msecs.", maxDowntime);
					throw new Exception (errmsg);
				}
			}
		}
	}
	private final void takeANap (long delay) {
		try { 
//			System.out.format("will sleep for %d msecs -- Shut down that Redis now\n", delay);
			Thread.sleep(delay); 
		}
		catch (Exception e) { }
	}
	/**
	 * Works fine if used in conjunction with a timestamp that keeps track of the
	 * last assert time.  If only invoked when the last assert time is greater than
	 * the redis client idel time setting (redis.conf::timeout) the RTT of this op's 
	 * is absorbed by all the request made during that idle time window.  A balanced
	 * setting of redis.conf timeout parameter would extend the number of these 
	 * requests to the point where the cost is effectively negligable.
	 *  
	 * @throws Exception
	 */
	public void testConnection() throws Exception{
		try {
			s.setSoTimeout(1);
			int r = din.read();
			if(r ==-1) {
				System.out.format("WARNING: Connection has been dropped\n");
				try {
					connect();
				}
				catch (Exception e){
//					System.out.format("ERROR: Failed to reconnect -- assuming net/server failure.\n");
					throw e;
				}
			}
		}
		catch (SocketTimeoutException e) {/* good news - will happen if connection is alive */ }
		catch (Exception e) {
//			System.out.format("ERROR: Failed to test connection -- assuming net/server failure.\n");
			throw e;
		}
		finally { s.setSoTimeout(0); }
	}
	/**
	 * Works but not entirely reliably.  We're paying PING/PONG round trip cost
	 * (~100 microsecs) per request.  Much better than the other method but it 
	 * seems to still miss dropped connections at times. (Not sure why).
	 * 
	 * @throws Exception
	 */
	public void testConnectionWithPing() throws Exception{
		try {
			ping();
		}
		catch (Exception e) {
			System.out.format("WARNING: Connection has been dropped\n");
			try {
				connect();
			}
			catch (Exception e2){
				System.err.format("ERROR: failed to reconnect; %s\n", e2.getMessage());
				throw new Exception ("Cannot connect to Redis server", e2);
			}
		}

	}
	public void connect() throws RuntimeException, UnknownHostException, IOException{
		s= new Socket("localhost", 6379);
		s.setSoTimeout(0);
		dout = new DataOutputStream(s.getOutputStream());
		din = new BufferedReader(new InputStreamReader(s.getInputStream()));

		if(password != null) 
			AUTH (password);
		System.out.format("INFO: Connection established.\n");
	}
	// ------ client API
	// a simple redis client impl.
	//
	public void ping () throws IOException {
		dout.write(String.format("PING\r\n").getBytes());
		System.out.format("%s\n", din.readLine());
	}
	public long INCR (String key) throws IOException {
		dout.write(String.format("INCR %s\r\n", key).getBytes());
		String resp = din.readLine();
		if(resp != null)
			return Long.parseLong(resp.substring(1));
		return -1;
	}
	public void AUTH (String password) throws IOException {
		dout.write(String.format("AUTH %s\r\n", password).getBytes());
		@SuppressWarnings("unused")
        String resp = din.readLine();
	}
}