JDBCRowLoader

package example.gemstone.sqlfabric.callbacks;

import java.sql.*;
import java.util.*;
import java.util.concurrent.*;

import com.gemstone.gemfire.LogWriter;
import com.gemstone.gemfire.cache.CacheFactory;
import com.gemstone.sqlfabric.callbacks.RowLoader;

public class JDBCRowLoader implements RowLoader {

    public static final int VENDOR_CODE_ARCHIVE_ERROR = 0;
    public static final int VENDOR_CODE_TIMEOUT = 1;

    private final Queue<PreparedStatement[]> waitingQueries = new LinkedList<PreparedStatement[]>();
    private final Queue<PreparedStatement> availableStatements = new LinkedList<PreparedStatement>();
    
    private String url = "";
    private String user = "";
    private String password = "";
    private String queryString = "";
	
    private final Properties props = new Properties();
    private int minConnections = 0;
    private int maxConnections = 1;
    private Integer connectionCount = 0;
    private long connectionTimeout = 3000;
    private final ExecutorService backgroundExecutor = Executors.newCachedThreadPool();
	
    private final LogWriter log = CacheFactory.getAnyInstance().getLogger();
	
    public static JDBCRowLoader create() {
        return new JDBCRowLoader();
    }

    public void init(Object[] params) {
		
        log.entering("example.gemstone.sqlfabric.loader.JDBCRowLoader", "init()");

        for (int i=0; i < params.length; i++) {
            String parameter = (String) params[i];
            int equalsIndex = parameter.indexOf('=');
            if ((equalsIndex > 0) & (parameter.length() > equalsIndex + 1)){
                String key = parameter.substring(0, equalsIndex).trim();
                String value = parameter.substring(equalsIndex + 1).trim();
                props.put(key, value);
            }
        }
	
        this.url = props.getProperty("url", "");
        this.user = props.getProperty("user" , "");
        this.password = props.getProperty("password" , "");
        this.queryString = props.getProperty("query-string" , "");
        this.minConnections = Integer.parseInt(props.getProperty("min-connections" , "1"));
        this.maxConnections = Integer.parseInt(props.getProperty("max-connections" , "1"));
        this.connectionTimeout = Long.parseLong(props.getProperty("connection-timeout" , "3000"));

        char[] pwdXChars = new char[password.length()];
        for (int i=0; i< password.length(); i++) {
            pwdXChars[i] = 'x';
        }

        if (log.infoEnabled()) {
            log.info("JDBCRowLoader initialized.");
            log.info("   connection url: " + url);
            log.info("   user          : " + user);
            log.info("   password      : " + String.copyValueOf(pwdXChars));
            log.info("   query string  : " + queryString);
        }

        for(int i = 0; i < minConnections; i++) {
            synchronized (connectionCount) {
                connectionCount = connectionCount +1;
            }
            StatementCreator creator = new StatementCreator();
            this.backgroundExecutor.execute(creator);
        }
    }

    public Object getRow(String schema, String table, Object[] params) throws SQLException {
        log.entering("example.gemstone.sqlfabric.loader.JDBCRowLoader", "getRow(String schema, String table, Object[] params)");
        if (log.infoEnabled()) {
            log.info("JDBCRowLoader invoked to fetch from schema <" + schema + "> on table <" + table + ">." );
            for(int i = 0; i < params.length; i++) {
                log.info("  primary key element " + i + ": " + params[i]);
            }
        }

        PreparedStatement[] holder = new PreparedStatement[1];
        synchronized (holder) {	
            this.getPooledStatement(holder);
            if (holder[0] == null) {
                try {
                    holder.wait(connectionTimeout);
                } catch (InterruptedException e) {
                    log.warning("JDBCRowLoader interrupted while waiting for an available pooled statement.");
                    log.warning(e);
                    Thread.currentThread().interrupt();
                }
            }
        }
        PreparedStatement pstmt = holder[0];
        if (pstmt == null) {
            throw new SQLException("Timeout waiting for pooled connection to archive database", "08001",
                                    VENDOR_CODE_TIMEOUT);
        }

        try {
            for (int i = 0; i < params.length; i++) {
                pstmt.setObject(i + 1, params[i]);
            }
            ResultSet result = pstmt.executeQuery();
            // even if this result set is empty (i.e. no row found), just return the empty result set
            log.info("Query succeeded");
            recyclePooledStatement(pstmt);
            return result;
        } catch (SQLException e) {
            // throw away the pooled statement, just in case it was the problem
            releasePooledStatement(pstmt);
            log.error("Error executing prepared statement in JDBCRowLoader");
            log.error(e);
            throw new SQLException("Error executing query from archive database", e.getSQLState(),
                                    VENDOR_CODE_ARCHIVE_ERROR, e);
        }
    }
	
    private synchronized void getPooledStatement(PreparedStatement[] holder) {
        // Take the next available prepared statement. If there isn't one,
        // add the holder to the list of waiting queries.
        // The calling thread must call holder.wait() if it finds the holder empty.
        // holder.notify() will be called when a prepared statement is available
        // and put into the holder.
		
        holder[0] = this.availableStatements.poll();
        if (holder[0] == null) {
            this.waitingQueries.add(holder);
            synchronized (connectionCount) {
                if ((connectionCount) < maxConnections) {
                    connectionCount = connectionCount +1;
                    StatementCreator creator = new StatementCreator();
                    this.backgroundExecutor.execute(creator);
                }
            }
        }
    }
	
    private synchronized void returnPooledStatement(PreparedStatement pstmt) {
        // Check to see if there are queries waiting on a statement.
        // If not, add the statement back into the pool.
        PreparedStatement[] holder = this.waitingQueries.poll();
        if (holder == null) {
            this.availableStatements.offer(pstmt);
        } else {
            synchronized (holder) {
                holder[0] = pstmt;
                holder.notify();
            }
        }
    }

    private void recyclePooledStatement(PreparedStatement pstmt) {
        StatementRecycler recycler = new StatementRecycler(pstmt);
        this.backgroundExecutor.execute(recycler);
    }

    private void releasePooledStatement(PreparedStatement pstmt) {
        StatementReleaser releaser = new StatementReleaser(pstmt);
        this.backgroundExecutor.execute(releaser);
    }

    private class StatementCreator implements Runnable {
        public void run() {
            if (url.isEmpty()) {
                log.error("Connection url not provided for JDBCRowLoader");
                return;
            }		
            try {
                Connection con = DriverManager.getConnection(url, props);
                log.info(" Successful connection to target database: " + url);
                con.setReadOnly(true);
                PreparedStatement pstmt = con.prepareStatement(queryString);
                recyclePooledStatement(pstmt);				
            } catch (SQLException e) {
                // Connection count is incremented when the job is scheduled.
                // Since it has failed, decrement the counter
                synchronized (connectionCount) {
                    connectionCount = connectionCount - 1;
                }
                log.error("Error connecting to target database");
                log.error(e);
            }		
        }
    }

    private class StatementRecycler implements Runnable {
        private PreparedStatement pstmt;
        StatementRecycler (PreparedStatement target) {
            this.pstmt = target;
        }
        public void run() {
            try {
                pstmt.clearParameters();
                returnPooledStatement(pstmt);
            } catch (SQLException e) {
                releasePooledStatement(pstmt);
                log.warning(e);
            }		
        }
    }

    private class StatementReleaser implements Runnable {
        private PreparedStatement pstmt;
        StatementReleaser (PreparedStatement target) {
            this.pstmt = target;
        }
        public void run() {
            try {
                synchronized (connectionCount) {
                    connectionCount = connectionCount - 1;
                }
                pstmt.getConnection().close();
            } catch (SQLException e) {
                log.warning(e);
            }	
        }
    }
}