- protected static final char DOT = '.';
- protected static final char DOT_PLUS_ONE = '.'+1;
- protected static final String FIRST_CHAR = Character.toString((char)0);
- protected static final String LAST_CHAR = Character.toString((char)Character.MAX_VALUE);
- protected static final int FIELD_COMMAS = 0;
- protected static final int QUESTION_COMMAS = 1;
- protected static final int ASSIGNMENT_COMMAS = 2;
- protected static final int WHERE_ANDS = 3;
-
- private Cluster cluster;
- /*
- * From DataStax
- * com.datastax.driver.core.Session
- A session holds connections to a Cassandra cluster, allowing it to be queried. Each session maintains multiple connections to the cluster nodes,
- provides policies to choose which node to use for each query (round-robin on all nodes of the cluster by default), and handles retries for
- failed query (when it makes sense), etc...
- Session instances are thread-safe and usually a single instance is enough per application. However, a given session can only be set to one
- keyspace at a time, so one instance per keyspace is necessary.
- */
- private Session session;
- private final String keyspace;
- // If this is null, then we own session
- private final AbsCassDAO<TRANS,?> owningDAO;
- protected Class<DATA> dataClass;
- private final String name;
-// private static Slot sessionSlot; // not used since 2015
- private static final ArrayList<AbsCassDAO<? extends TransStore,?>.PSInfo> psinfos = new ArrayList<>();
- private static final List<Object> EMPTY = new ArrayList<>(0);
- private static final Deque<ResetRequest> resetDeque = new ConcurrentLinkedDeque<ResetRequest>();
- private static boolean resetTrigger = false;
- private static long nextAvailableReset = 0;
-
- public AbsCassDAO(TRANS trans, String name, Cluster cluster, String keyspace, Class<DATA> dataClass) {
- this.name = name;
- this.cluster = cluster;
- this.keyspace = keyspace;
- owningDAO = null; // we own session
- session = null;
- this.dataClass = dataClass;
- }
-
- public AbsCassDAO(TRANS trans, String name, AbsCassDAO<TRANS,?> aDao, Class<DATA> dataClass) {
- this.name = name;
- cluster = aDao.cluster;
- keyspace = aDao.keyspace;
- session = null;
- // We do not own session
- owningDAO = aDao;
- this.dataClass = dataClass;
- }
-
-// Not used since 2015
-// public static void setSessionSlot(Slot slot) {
-// sessionSlot = slot;
-// }
-
- //Note: Lower case ON PURPOSE. These names used to create History Messages
- public enum CRUD {
- create,read,update,delete;
- }
-
- public class PSInfo {
- private PreparedStatement ps;
- private final int size;
- private final Loader<DATA> loader;
- private final CRUD crud; // Store CRUD, because it makes a difference in Object Order, see Loader
- private final String cql;
- private final ConsistencyLevel consistency;
-
-
- /**
- * Create a PSInfo and create Prepared Statement
- *
- * @param trans
- * @param theCQL
- * @param loader
- */
- public PSInfo(TRANS trans, String theCQL, Loader<DATA> loader, ConsistencyLevel consistency) {
- this.loader = loader;
- this.consistency=consistency;
- psinfos.add(this);
-
- cql = theCQL.trim().toUpperCase();
- if(cql.startsWith("INSERT")) {
- crud = CRUD.create;
- } else if(cql.startsWith("UPDATE")) {
- crud = CRUD.update;
- } else if(cql.startsWith("DELETE")) {
- crud = CRUD.delete;
- } else {
- crud = CRUD.read;
- }
-
- int idx = 0, count=0;
- while((idx=cql.indexOf('?',idx))>=0) {
- ++idx;
- ++count;
- }
- size=count;
- }
-
- public synchronized void reset() {
- ps = null;
- }
-
- private synchronized BoundStatement ps(TransStore trans) throws APIException, IOException {
- /* From Datastax
- You should prepare only once, and cache the PreparedStatement in your application (it is thread-safe).
- If you call prepare multiple times with the same query string, the driver will log a warning.
- */
- if(ps==null) {
- TimeTaken tt = trans.start("Preparing PSInfo " + crud.toString().toUpperCase() + " on " + name,Env.SUB);
- try {
- ps = getSession(trans).prepare(cql);
- ps.setConsistencyLevel(consistency);
- } catch (DriverException e) {
- reportPerhapsReset(trans,e);
- throw e;
- } finally {
- tt.done();
- }
- }
- // BoundStatements are NOT threadsafe... need a new one each time.
- return new BoundStatement(ps);
- }
-
- /**
- * Execute a Prepared Statement by extracting from DATA object
- *
- * @param trans
- * @param text
- * @param data
- * @return
- */
- public Result<ResultSetFuture> execAsync(TRANS trans, String text, DATA data) {
- TimeTaken tt = trans.start(text, Env.REMOTE);
- try {
- return Result.ok(getSession(trans).executeAsync(
- ps(trans).bind(loader.extract(data, size, crud))));
- } catch (DriverException | APIException | IOException e) {
- AbsCassDAO.this.reportPerhapsReset(trans,e);
- return Result.err(Status.ERR_Backend,"%s-%s executing %s",e.getClass().getName(),e.getMessage(), cql);
- } finally {
- tt.done();
- }
- }
-
- /**
- * Execute a Prepared Statement on Object[] key
- *
- * @param trans
- * @param text
- * @param objs
- * @return
- */
- public Result<ResultSetFuture> execAsync(TRANS trans, String text, Object ... objs) {
- TimeTaken tt = trans.start(text, Env.REMOTE);
- try {
- return Result.ok(getSession(trans).executeAsync(ps(trans).bind(objs)));
- } catch (DriverException | APIException | IOException e) {
- AbsCassDAO.this.reportPerhapsReset(trans,e);
- return Result.err(Status.ERR_Backend,"%s-%s executing %s",e.getClass().getName(),e.getMessage(), cql);
- } finally {
- tt.done();
- }
- }
-
- /*
- * Note:
- *
- */
-
- /**
- * Execute a Prepared Statement by extracting from DATA object
- *
- * @param trans
- * @param text
- * @param data
- * @return
- */
- public Result<ResultSet> exec(TRANS trans, String text, DATA data) {
- TimeTaken tt = trans.start(text, Env.REMOTE);
- try {
- /*
- * "execute" (and executeAsync)
- * Executes the provided query.
- This method blocks until at least some result has been received from the database. However,
- for SELECT queries, it does not guarantee that the result has been received in full. But it
- does guarantee that some response has been received from the database, and in particular
- guarantee that if the request is invalid, an exception will be thrown by this method.
-
- Parameters:
- statement - the CQL query to execute (that can be any Statement).
- Returns:
- the result of the query. That result will never be null but can be empty (and will
- be for any non SELECT query).
- */
- return Result.ok(getSession(trans).execute(
- ps(trans).bind(loader.extract(data, size, crud))));
- } catch (DriverException | APIException | IOException e) {
- AbsCassDAO.this.reportPerhapsReset(trans,e);
- return Result.err(Status.ERR_Backend,"%s-%s executing %s",e.getClass().getName(),e.getMessage(), cql);
- } finally {
- tt.done();
- }
- }
-
- /**
- * Execute a Prepared Statement on Object[] key
- *
- * @param trans
- * @param text
- * @param objs
- * @return
- */
- public Result<ResultSet> exec(TRANS trans, String text, Object ... objs) {
- TimeTaken tt = trans.start(text, Env.REMOTE);
- try {
- return Result.ok(getSession(trans).execute(ps(trans).bind(objs)));
- } catch (DriverException | APIException | IOException e) {
- AbsCassDAO.this.reportPerhapsReset(trans,e);
- return Result.err(Status.ERR_Backend,"%s-%s executing %s",e.getClass().getName(),e.getMessage(), cql);
- } finally {
- tt.done();
- }
- }
-
- /**
- * Read the Data from Cassandra given a Prepared Statement (defined by the
- * DAO Instance)
- *
- * This is common behavior among all DAOs.
- * @throws DAOException
- */
- public Result<List<DATA>> read(TRANS trans, String text, Object[] key) {
- TimeTaken tt = trans.start(text,Env.REMOTE);
-
- ResultSet rs;
- try {
- rs = getSession(trans).execute(key==null?ps(trans):ps(trans).bind(key));
-/// TEST CODE for Exception
-// boolean force = true;
-// if(force) {
-// Map<InetSocketAddress, Throwable> misa = new HashMap<>();
-// //misa.put(new InetSocketAddress(444),new Exception("no host was tried"));
-// misa.put(new InetSocketAddress(444),new Exception("Connection has been closed"));
-// throw new com.datastax.driver.core.exceptions.NoHostAvailableException(misa);
-//// throw new com.datastax.driver.core.exceptions.AuthenticationException(new InetSocketAddress(9999),"no host was tried");
-// }
+ protected static final char DOT = '.';
+ protected static final char DOT_PLUS_ONE = '.'+1;
+ protected static final String FIRST_CHAR = Character.toString((char)0);
+ protected static final String LAST_CHAR = Character.toString((char)Character.MAX_VALUE);
+ protected static final int FIELD_COMMAS = 0;
+ protected static final int QUESTION_COMMAS = 1;
+ protected static final int ASSIGNMENT_COMMAS = 2;
+ protected static final int WHERE_ANDS = 3;
+
+ private Cluster cluster;
+ /*
+ * From DataStax
+ * com.datastax.driver.core.Session
+ A session holds connections to a Cassandra cluster, allowing it to be queried. Each session maintains multiple connections to the cluster nodes,
+ provides policies to choose which node to use for each query (round-robin on all nodes of the cluster by default), and handles retries for
+ failed query (when it makes sense), etc...
+ Session instances are thread-safe and usually a single instance is enough per application. However, a given session can only be set to one
+ keyspace at a time, so one instance per keyspace is necessary.
+ */
+ private Session session;
+ private final String keyspace;
+ // If this is null, then we own session
+ private final AbsCassDAO<TRANS,?> owningDAO;
+ protected Class<DATA> dataClass;
+ private final String name;
+// private static Slot sessionSlot; // not used since 2015
+ private static final ArrayList<AbsCassDAO<? extends TransStore,?>.PSInfo> psinfos = new ArrayList<>();
+ private static final List<Object> EMPTY = new ArrayList<>(0);
+ private static final Deque<ResetRequest> resetDeque = new ConcurrentLinkedDeque<>();
+ private static boolean resetTrigger = false;
+ private static long nextAvailableReset = 0;
+
+ public AbsCassDAO(TRANS trans, String name, Cluster cluster, String keyspace, Class<DATA> dataClass) {
+ this.name = name;
+ this.cluster = cluster;
+ this.keyspace = keyspace;
+ owningDAO = null; // we own session
+ session = null;
+ this.dataClass = dataClass;
+ }
+
+ public AbsCassDAO(TRANS trans, String name, AbsCassDAO<TRANS,?> aDao, Class<DATA> dataClass) {
+ this.name = name;
+ cluster = aDao.cluster;
+ keyspace = aDao.keyspace;
+ session = null;
+ // We do not own session
+ owningDAO = aDao;
+ this.dataClass = dataClass;
+ }
+
+
+ //Note: Lower case ON PURPOSE. These names used to create History Messages
+ public enum CRUD {
+ create,read,update,delete;
+ }
+
+ public class PSInfo {
+ private PreparedStatement ps;
+ private final int size;
+ private final Loader<DATA> loader;
+ private final CRUD crud; // Store CRUD, because it makes a difference in Object Order, see Loader
+ private final String cql;
+ private final ConsistencyLevel consistency;
+
+
+ /**
+ * Create a PSInfo and create Prepared Statement
+ *
+ * @param trans
+ * @param theCQL
+ * @param loader
+ */
+ public PSInfo(TRANS trans, String theCQL, Loader<DATA> loader, ConsistencyLevel consistency) {
+ this.loader = loader;
+ this.consistency=consistency;
+ psinfos.add(this);
+
+ cql = theCQL.trim().toUpperCase();
+ if (cql.startsWith("INSERT")) {
+ crud = CRUD.create;
+ } else if (cql.startsWith("UPDATE")) {
+ crud = CRUD.update;
+ } else if (cql.startsWith("DELETE")) {
+ crud = CRUD.delete;
+ } else {
+ crud = CRUD.read;
+ }
+
+ int idx = 0, count=0;
+ while ((idx=cql.indexOf('?',idx))>=0) {
+ ++idx;
+ ++count;
+ }
+ size=count;
+ }
+
+ public synchronized void reset() {
+ ps = null;
+ }
+
+ private synchronized BoundStatement ps(TransStore trans) throws APIException, IOException {
+ /* From Datastax
+ You should prepare only once, and cache the PreparedStatement in your application (it is thread-safe).
+ If you call prepare multiple times with the same query string, the driver will log a warning.
+ */
+ if (ps==null) {
+ TimeTaken tt = trans.start("Preparing PSInfo " + crud.toString().toUpperCase() + " on " + name,Env.SUB);
+ try {
+ ps = getSession(trans).prepare(cql);
+ ps.setConsistencyLevel(consistency);
+ } catch (DriverException e) {
+ reportPerhapsReset(trans,e);
+ throw e;
+ } finally {
+ tt.done();
+ }
+ }
+ // BoundStatements are NOT threadsafe... need a new one each time.
+ return new BoundStatement(ps);
+ }
+
+ /**
+ * Execute a Prepared Statement by extracting from DATA object
+ *
+ * @param trans
+ * @param text
+ * @param data
+ * @return
+ */
+ public Result<ResultSetFuture> execAsync(TRANS trans, String text, DATA data) {
+ TimeTaken tt = trans.start(text, Env.REMOTE);
+ try {
+ return Result.ok(getSession(trans).executeAsync(
+ ps(trans).bind(loader.extract(data, size, crud))));
+ } catch (DriverException | APIException | IOException e) {
+ AbsCassDAO.this.reportPerhapsReset(trans,e);
+ return Result.err(Status.ERR_Backend,"%s-%s executing %s",e.getClass().getName(),e.getMessage(), cql);
+ } finally {
+ tt.done();
+ }
+ }
+
+ /**
+ * Execute a Prepared Statement on Object[] key
+ *
+ * @param trans
+ * @param text
+ * @param objs
+ * @return
+ */
+ public Result<ResultSetFuture> execAsync(TRANS trans, String text, Object ... objs) {
+ TimeTaken tt = trans.start(text, Env.REMOTE);
+ try {
+ return Result.ok(getSession(trans).executeAsync(ps(trans).bind(objs)));
+ } catch (DriverException | APIException | IOException e) {
+ AbsCassDAO.this.reportPerhapsReset(trans,e);
+ return Result.err(Status.ERR_Backend,"%s-%s executing %s",e.getClass().getName(),e.getMessage(), cql);
+ } finally {
+ tt.done();
+ }
+ }
+
+ /*
+ * Note:
+ *
+ */
+
+ /**
+ * Execute a Prepared Statement by extracting from DATA object
+ *
+ * @param trans
+ * @param text
+ * @param data
+ * @return
+ */
+ public Result<ResultSet> exec(TRANS trans, String text, DATA data) {
+ TimeTaken tt = trans.start(text, Env.REMOTE);
+ try {
+ /*
+ * "execute" (and executeAsync)
+ * Executes the provided query.
+ This method blocks until at least some result has been received from the database. However,
+ for SELECT queries, it does not guarantee that the result has been received in full. But it
+ does guarantee that some response has been received from the database, and in particular
+ guarantee that if the request is invalid, an exception will be thrown by this method.
+
+ Parameters:
+ statement - the CQL query to execute (that can be any Statement).
+ Returns:
+ the result of the query. That result will never be null but can be empty (and will
+ be for any non SELECT query).
+ */
+ return Result.ok(getSession(trans).execute(
+ ps(trans).bind(loader.extract(data, size, crud))));
+ } catch (DriverException | APIException | IOException e) {
+ AbsCassDAO.this.reportPerhapsReset(trans,e);
+ return Result.err(Status.ERR_Backend,"%s-%s executing %s",e.getClass().getName(),e.getMessage(), cql);
+ } finally {
+ tt.done();
+ }
+ }
+
+ /**
+ * Execute a Prepared Statement on Object[] key
+ *
+ * @param trans
+ * @param text
+ * @param objs
+ * @return
+ */
+ public Result<ResultSet> exec(TRANS trans, String text, Object ... objs) {
+ TimeTaken tt = trans.start(text, Env.REMOTE);
+ try {
+ return Result.ok(getSession(trans).execute(ps(trans).bind(objs)));
+ } catch (DriverException | APIException | IOException e) {
+ AbsCassDAO.this.reportPerhapsReset(trans,e);
+ return Result.err(Status.ERR_Backend,"%s-%s executing %s",e.getClass().getName(),e.getMessage(), cql);
+ } finally {
+ tt.done();
+ }
+ }
+
+ /**
+ * Read the Data from Cassandra given a Prepared Statement (defined by the
+ * DAO Instance)
+ *
+ * This is common behavior among all DAOs.
+ * @throws DAOException
+ */
+ public Result<List<DATA>> read(TRANS trans, String text, Object[] key) {
+ TimeTaken tt = trans.start(text,Env.REMOTE);
+
+ ResultSet rs;
+ try {
+ rs = getSession(trans).execute(key==null?ps(trans):ps(trans).bind(key));
+/// TEST CODE for Exception
+// boolean force = true;
+// if (force) {
+// Map<InetSocketAddress, Throwable> misa = new HashMap<>();
+// //misa.put(new InetSocketAddress(444),new Exception("no host was tried"));
+// misa.put(new InetSocketAddress(444),new Exception("Connection has been closed"));
+// throw new com.datastax.driver.core.exceptions.NoHostAvailableException(misa);
+//// throw new com.datastax.driver.core.exceptions.AuthenticationException(new InetSocketAddress(9999),"no host was tried");
+// }