class AWSObjectStoreFactory implements ObjectStoreFactory {

    @Override
    public ObjectStore create(PrivateKeyStore keyStore) {
        
        rdsUser = System.getProperty(ZMSConsts.ZMS_PROP_AWS_RDS_USER);
        rdsIamRole = System.getProperty(ZMSConsts.ZMS_PROP_AWS_RDS_IAM_ROLE);
        rdsMaster = System.getProperty(ZMSConsts.ZMS_PROP_AWS_RDS_MASTER_INSTANCE);
        rdsReplica = System.getProperty(ZMSConsts.ZMS_PROP_AWS_RDS_REPLICA_INSTANCE);
        rdsPort = Integer.parseInt(System.getProperty(ZMSConsts.ZMS_PROP_AWS_RDS_MASTER_PORT, "3306"));
        
        final String rdsEngine = System.getProperty(ZMSConsts.ZMS_PROP_AWS_RDS_ENGINE, "mysql");
        final String rdsDatabase = System.getProperty(ZMSConsts.ZMS_PROP_AWS_RDS_DATABASE, "zms_store");
        final String jdbcMasterStore = String.format("jdbc:%s://%s:%d/%s", rdsEngine,
                rdsMaster, rdsPort, rdsDatabase);
        final String rdsMasterToken = getAuthToken(rdsMaster, rdsPort, rdsUser, rdsIamRole);
        
        if (LOG.isDebugEnabled()) {
            LOG.debug("Connecting to master {} with auth token {}", jdbcMasterStore, rdsMasterToken);
        }
        
        mysqlMasterConnectionProperties.setProperty(ZMSConsts.DB_PROP_VERIFY_SERVER_CERT,
                System.getProperty(ZMSConsts.ZMS_PROP_JDBC_VERIFY_SERVER_CERT, "true"));
        mysqlMasterConnectionProperties.setProperty(ZMSConsts.DB_PROP_USE_SSL,
                System.getProperty(ZMSConsts.ZMS_PROP_JDBC_USE_SSL, "true"));
        mysqlMasterConnectionProperties.setProperty(ZMSConsts.DB_PROP_USER, rdsUser);
        mysqlMasterConnectionProperties.setProperty(ZMSConsts.DB_PROP_PASSWORD, rdsMasterToken);
        
        PoolableDataSource dataMasterSource = DataSourceFactory.create(jdbcMasterStore, mysqlMasterConnectionProperties);
        
        // now check to see if we also have a read-only replica jdbc store configured

        PoolableDataSource dataReplicaSource = null;
        if (rdsReplica != null) {
            
            final String jdbcReplicaStore = String.format("jdbc:%s://%s:%d/%s", rdsEngine,
                    rdsReplica, rdsPort, rdsDatabase);
            final String rdsReplicaToken = getAuthToken(rdsReplica, rdsPort, rdsUser, rdsIamRole);
            
            if (LOG.isDebugEnabled()) {
                LOG.debug("Connecting to replica {} with auth token {}", jdbcReplicaStore, rdsReplicaToken);
            }
            
            mysqlReplicaConnectionProperties.setProperty(ZMSConsts.DB_PROP_VERIFY_SERVER_CERT,
                    System.getProperty(ZMSConsts.ZMS_PROP_JDBC_VERIFY_SERVER_CERT, "true"));
            mysqlReplicaConnectionProperties.setProperty(ZMSConsts.DB_PROP_USE_SSL,
                    System.getProperty(ZMSConsts.ZMS_PROP_JDBC_USE_SSL, "true"));
            mysqlReplicaConnectionProperties.setProperty(ZMSConsts.DB_PROP_USER, rdsUser);
            mysqlReplicaConnectionProperties.setProperty(ZMSConsts.DB_PROP_PASSWORD, rdsReplicaToken);
            
            dataReplicaSource = DataSourceFactory.create(jdbcReplicaStore, mysqlReplicaConnectionProperties);
        }
        
        // start our credentials refresh task
        
        long credsRefreshTime = Integer.parseInt(System.getProperty(ZMSConsts.ZMS_PROP_AWS_RDS_CREDS_REFRESH_TIME, "300"));

        scheduledThreadPool = Executors.newScheduledThreadPool(1);
        scheduledThreadPool.scheduleAtFixedRate(new CredentialsUpdater(), credsRefreshTime,
                credsRefreshTime, TimeUnit.SECONDS);
        
        return new JDBCObjectStore(dataMasterSource, dataReplicaSource);
    }

}