+++ /dev/null
-/*******************************************************************************\r
- * ============LICENSE_START====================================================\r
- * * org.onap.aaf\r
- * * ===========================================================================\r
- * * Copyright © 2017 AT&T Intellectual Property. All rights reserved.\r
- * * ===========================================================================\r
- * * Licensed under the Apache License, Version 2.0 (the "License");\r
- * * you may not use this file except in compliance with the License.\r
- * * You may obtain a copy of the License at\r
- * * \r
- * * http://www.apache.org/licenses/LICENSE-2.0\r
- * * \r
- * * Unless required by applicable law or agreed to in writing, software\r
- * * distributed under the License is distributed on an "AS IS" BASIS,\r
- * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
- * * See the License for the specific language governing permissions and\r
- * * limitations under the License.\r
- * * ============LICENSE_END====================================================\r
- * *\r
- * * ECOMP is a trademark and service mark of AT&T Intellectual Property.\r
- * *\r
- ******************************************************************************/\r
-package org.onap.aaf.dao;\r
-\r
-import java.io.IOException;\r
-import java.util.ArrayList;\r
-import java.util.Deque;\r
-import java.util.List;\r
-import java.util.concurrent.ConcurrentLinkedDeque;\r
-\r
-import org.onap.aaf.authz.layer.Result;\r
-import org.onap.aaf.dao.aaf.cass.Status;\r
-\r
-import org.onap.aaf.inno.env.APIException;\r
-import org.onap.aaf.inno.env.Env;\r
-import org.onap.aaf.inno.env.Slot;\r
-import org.onap.aaf.inno.env.TimeTaken;\r
-import org.onap.aaf.inno.env.TransStore;\r
-import com.datastax.driver.core.BoundStatement;\r
-import com.datastax.driver.core.Cluster;\r
-import com.datastax.driver.core.ConsistencyLevel;\r
-import com.datastax.driver.core.ResultSet;\r
-import com.datastax.driver.core.ResultSetFuture;\r
-import com.datastax.driver.core.Row;\r
-import com.datastax.driver.core.Session;\r
-import com.datastax.driver.core.exceptions.DriverException;\r
-\r
-public abstract class AbsCassDAO<TRANS extends TransStore,DATA> {\r
- protected static final char DOT = '.';\r
- protected static final char DOT_PLUS_ONE = '.'+1;\r
- protected static final String FIRST_CHAR = Character.toString((char)0);\r
- protected static final String LAST_CHAR = Character.toString((char)Character.MAX_VALUE);\r
- protected static final int FIELD_COMMAS = 0;\r
- protected static final int QUESTION_COMMAS = 1;\r
- protected static final int ASSIGNMENT_COMMAS = 2;\r
- protected static final int WHERE_ANDS = 3;\r
- \r
- private Cluster cluster; \r
- private Session session;\r
- private final String keyspace;\r
- // If this is null, then we own session\r
- private final AbsCassDAO<TRANS,?> owningDAO;\r
- protected Class<DATA> dataClass;\r
- private final String name;\r
- private static Slot sessionSlot;\r
- //private static final ArrayList<AbsCassDAO<? extends TransStore,?>.PSInfo> psinfos = new ArrayList<AbsCassDAO<TransStore,?>.PSInfo>();\r
- private static final ArrayList<AbsCassDAO<? extends TransStore,?>.PSInfo> psinfos = new ArrayList<AbsCassDAO<? extends TransStore,?>.PSInfo>();\r
- private static final List<Object> EMPTY = new ArrayList<Object>(0);\r
- private static final Deque<ResetRequest> resetDeque = new ConcurrentLinkedDeque<ResetRequest>();\r
- private static boolean resetTrigger = false;\r
- private static long nextAvailableReset = 0;\r
- \r
-\r
- public AbsCassDAO(TRANS trans, String name, Cluster cluster, String keyspace, Class<DATA> dataClass) {\r
- this.name = name;\r
- this.cluster = cluster;\r
- this.keyspace = keyspace;\r
- owningDAO = null; // we own session\r
- session = null;\r
- this.dataClass = dataClass;\r
- \r
- }\r
-\r
- public AbsCassDAO(TRANS trans, String name, AbsCassDAO<TRANS,?> aDao, Class<DATA> dataClass) { \r
- this.name = name;\r
- cluster = aDao.cluster;\r
- keyspace = aDao.keyspace;\r
- session = null;\r
- owningDAO = aDao; // We do not own session\r
- this.dataClass = dataClass;\r
- }\r
- \r
- public static void setSessionSlot(Slot slot) {\r
- sessionSlot = slot;\r
- }\r
-\r
- //Note: Lower case ON PURPOSE. These names used to create History Messages\r
- public enum CRUD {\r
- create,read,update,delete\r
- ;\r
-\r
-}\r
-\r
- public class PSInfo {\r
- private BoundStatement ps;\r
- private final int size;\r
- private final Loader<DATA> loader;\r
- private final CRUD crud; // Store CRUD, because it makes a difference in Object Order, see Loader\r
- private final String cql;\r
- private final ConsistencyLevel consistency;\r
-\r
-\r
- /**\r
- * Create a PSInfo and create Prepared Statement\r
- * \r
- * @param trans\r
- * @param theCQL\r
- * @param loader\r
- */\r
- public PSInfo(TRANS trans, String theCQL, Loader<DATA> loader, ConsistencyLevel consistency) {\r
- this.loader = loader;\r
- this.consistency=consistency;\r
- psinfos.add(this);\r
-\r
- cql = theCQL.trim().toUpperCase();\r
- if(cql.startsWith("INSERT")) {\r
- crud = CRUD.create;\r
- } else if(cql.startsWith("UPDATE")) {\r
- crud = CRUD.update;\r
- } else if(cql.startsWith("DELETE")) {\r
- crud = CRUD.delete;\r
- } else {\r
- crud = CRUD.read;\r
- }\r
- \r
- int idx = 0, count=0;\r
- while((idx=cql.indexOf('?',idx))>=0) {\r
- ++idx;\r
- ++count;\r
- }\r
- size=count;\r
- }\r
- \r
- public synchronized void reset() {\r
- ps = null;\r
- }\r
- \r
- private BoundStatement ps(TransStore trans) throws APIException, IOException {\r
- if(ps==null) {\r
- synchronized(this) {\r
- if(ps==null) {\r
- TimeTaken tt = trans.start("Preparing PSInfo " + crud.toString().toUpperCase() + " on " + name,Env.SUB);\r
- try {\r
- ps = new BoundStatement(getSession(trans).prepare(cql));\r
- ps.setConsistencyLevel(consistency);\r
- } catch (DriverException e) {\r
- reportPerhapsReset(trans,e);\r
- throw e;\r
- } finally {\r
- tt.done();\r
- }\r
- }\r
- }\r
- }\r
- return ps;\r
- }\r
-\r
- /**\r
- * Execute a Prepared Statement by extracting from DATA object\r
- * \r
- * @param trans\r
- * @param text\r
- * @param data\r
- * @return\r
- */\r
- public Result<ResultSetFuture> execAsync(TRANS trans, String text, DATA data) {\r
- TimeTaken tt = trans.start(text, Env.REMOTE);\r
- try {\r
- return Result.ok(getSession(trans).executeAsync(\r
- ps(trans).bind(loader.extract(data, size, crud))));\r
- } catch (DriverException | APIException | IOException e) {\r
- AbsCassDAO.this.reportPerhapsReset(trans,e);\r
- return Result.err(Status.ERR_Backend,"%s-%s executing %s",e.getClass().getName(),e.getMessage(), cql);\r
- } finally {\r
- tt.done();\r
- }\r
- }\r
-\r
- /**\r
- * Execute a Prepared Statement on Object[] key\r
- * \r
- * @param trans\r
- * @param text\r
- * @param objs\r
- * @return\r
- */\r
- public Result<ResultSetFuture> execAsync(TRANS trans, String text, Object ... objs) {\r
- TimeTaken tt = trans.start(text, Env.REMOTE);\r
- try {\r
- return Result.ok(getSession(trans).executeAsync(ps(trans).bind(objs)));\r
- } catch (DriverException | APIException | IOException e) {\r
- AbsCassDAO.this.reportPerhapsReset(trans,e);\r
- return Result.err(Status.ERR_Backend,"%s-%s executing %s",e.getClass().getName(),e.getMessage(), cql);\r
- } finally {\r
- tt.done();\r
- }\r
- }\r
- \r
- /* \r
- * Note:\r
- * \r
- */\r
-\r
- /**\r
- * Execute a Prepared Statement by extracting from DATA object\r
- * \r
- * @param trans\r
- * @param text\r
- * @param data\r
- * @return\r
- */\r
- public Result<ResultSet> exec(TRANS trans, String text, DATA data) {\r
- TimeTaken tt = trans.start(text, Env.REMOTE);\r
- try {\r
- /*\r
- * "execute" (and executeAsync)\r
- * Executes the provided query.\r
- This method blocks until at least some result has been received from the database. However, \r
- for SELECT queries, it does not guarantee that the result has been received in full. But it \r
- does guarantee that some response has been received from the database, and in particular \r
- guarantee that if the request is invalid, an exception will be thrown by this method.\r
-\r
- Parameters:\r
- statement - the CQL query to execute (that can be any Statement).\r
- Returns:\r
- the result of the query. That result will never be null but can be empty (and will \r
- be for any non SELECT query).\r
- */\r
- return Result.ok(getSession(trans).execute(\r
- ps(trans).bind(loader.extract(data, size, crud))));\r
- } catch (DriverException | APIException | IOException e) {\r
- AbsCassDAO.this.reportPerhapsReset(trans,e);\r
- return Result.err(Status.ERR_Backend,"%s-%s executing %s",e.getClass().getName(),e.getMessage(), cql);\r
- } finally {\r
- tt.done();\r
- }\r
- }\r
-\r
- /**\r
- * Execute a Prepared Statement on Object[] key\r
- * \r
- * @param trans\r
- * @param text\r
- * @param objs\r
- * @return\r
- */\r
- public Result<ResultSet> exec(TRANS trans, String text, Object ... objs) {\r
- TimeTaken tt = trans.start(text, Env.REMOTE);\r
- try {\r
- return Result.ok(getSession(trans).execute(ps(trans).bind(objs)));\r
- } catch (DriverException | APIException | IOException e) {\r
- AbsCassDAO.this.reportPerhapsReset(trans,e);\r
- return Result.err(Status.ERR_Backend,"%s-%s executing %s",e.getClass().getName(),e.getMessage(), cql);\r
- } finally {\r
- tt.done();\r
- }\r
- }\r
-\r
- /**\r
- * Read the Data from Cassandra given a Prepared Statement (defined by the\r
- * DAO Instance)\r
- *\r
- * This is common behavior among all DAOs.\r
- * @throws DAOException\r
- */\r
- public Result<List<DATA>> read(TRANS trans, String text, Object[] key) {\r
- TimeTaken tt = trans.start(text,Env.REMOTE);\r
- \r
- ResultSet rs;\r
- try {\r
- rs = getSession(trans).execute(key==null?ps(trans):ps(trans).bind(key));\r
-/// TEST CODE for Exception \r
-// boolean force = true; \r
-// if(force) {\r
-// Map<InetSocketAddress, Throwable> misa = new HashMap<InetSocketAddress,Throwable>();\r
-// //misa.put(new InetSocketAddress(444),new Exception("no host was tried"));\r
-// misa.put(new InetSocketAddress(444),new Exception("Connection has been closed"));\r
-// throw new com.datastax.driver.core.exceptions.NoHostAvailableException(misa);\r
-//// throw new com.datastax.driver.core.exceptions.AuthenticationException(new InetSocketAddress(9999),"no host was tried");\r
-// }\r
-//// END TEST CODE\r
- } catch (DriverException | APIException | IOException e) {\r
- AbsCassDAO.this.reportPerhapsReset(trans,e);\r
- return Result.err(Status.ERR_Backend,"%s-%s executing %s",e.getClass().getName(),e.getMessage(), cql);\r
- } finally {\r
- tt.done();\r
- }\r
- \r
- return extract(loader,rs,null /*let Array be created if necessary*/,dflt);\r
- }\r
- \r
- public Result<List<DATA>> read(TRANS trans, String text, DATA data) {\r
- return read(trans,text, loader.extract(data, size, crud));\r
- }\r
- \r
- public Object[] keyFrom(DATA data) {\r
- return loader.extract(data, size, CRUD.delete); // Delete is key only\r
- }\r
-\r
- /*\r
- * Note: in case PSInfos are deleted, we want to remove them from list. This is not expected, \r
- * but we don't want a data leak if it does. Finalize doesn't have to happen quickly\r
- */\r
- @Override\r
- protected void finalize() throws Throwable {\r
- psinfos.remove(this);\r
- }\r
- }\r
-\r
- protected final Accept<DATA> dflt = new Accept<DATA>() {\r
- @Override\r
- public boolean ok(DATA data) {\r
- return true;\r
- }\r
- };\r
-\r
-\r
- @SuppressWarnings("unchecked")\r
- protected final Result<List<DATA>> extract(Loader<DATA> loader, ResultSet rs, List<DATA> indata, Accept<DATA> accept) {\r
- List<Row> rows = rs.all();\r
- if(rows.isEmpty()) {\r
- return Result.ok((List<DATA>)EMPTY); // Result sets now .emptyList(true);\r
- } else {\r
- DATA d;\r
- List<DATA> data = indata==null?new ArrayList<DATA>(rows.size()):indata;\r
- \r
- for(Row row : rows) {\r
- try {\r
- d = loader.load(dataClass.newInstance(),row);\r
- if(accept.ok(d)) {\r
- data.add(d);\r
- }\r
- } catch(Exception e) {\r
- return Result.err(e);\r
- }\r
- }\r
- return Result.ok(data);\r
- }\r
- }\r
- \r
- private static final String NEW_CASSANDRA_SESSION_CREATED = "New Cassandra Session Created";\r
- private static final String NEW_CASSANDRA_CLUSTER_OBJECT_CREATED = "New Cassandra Cluster Object Created";\r
- private static final String NEW_CASSANDRA_SESSION = "New Cassandra Session";\r
-\r
- private static class ResetRequest {\r
- //package on purpose\r
- Session session;\r
- long timestamp;\r
- \r
- public ResetRequest(Session session) {\r
- this.session = session;\r
- timestamp = System.currentTimeMillis();\r
- }\r
- }\r
-\r
- \r
- public static final void primePSIs(TransStore trans) throws APIException, IOException {\r
- for(AbsCassDAO<? extends TransStore, ?>.PSInfo psi : psinfos) {\r
- if(psi.ps==null) {\r
- psi.ps(trans);\r
- }\r
- }\r
- }\r
- \r
- public final Session getSession(TransStore trans) throws APIException, IOException {\r
- // Try to use Trans' session, if exists\r
- if(sessionSlot!=null) { // try to get from Trans\r
- Session sess = trans.get(sessionSlot, null);\r
- if(sess!=null) {\r
- return sess;\r
- }\r
- }\r
- \r
- // If there's an owning DAO, use it's session\r
- if(owningDAO!=null) {\r
- return owningDAO.getSession(trans);\r
- }\r
- \r
- // OK, nothing else works... get our own.\r
- if(session==null || resetTrigger) {\r
- Cluster tempCluster = null;\r
- Session tempSession = null;\r
- try {\r
- synchronized(NEW_CASSANDRA_SESSION_CREATED) {\r
- boolean reset = false;\r
- for(ResetRequest r : resetDeque) {\r
- if(r.session == session) {\r
- if(r.timestamp>nextAvailableReset) {\r
- reset=true;\r
- nextAvailableReset = System.currentTimeMillis() + 60000;\r
- tempCluster = cluster;\r
- tempSession = session;\r
- break;\r
- } else {\r
- trans.warn().log("Cassandra Connection Reset Ignored: Recent Reset");\r
- }\r
- }\r
- }\r
- \r
- if(reset || session == null) {\r
- TimeTaken tt = trans.start(NEW_CASSANDRA_SESSION, Env.SUB);\r
- try {\r
- // Note: Maitrayee recommended not closing the cluster, just\r
- // overwrite it. 9/30/2016 assuming same for Session\r
- // This was a bad idea. Ran out of File Handles as I suspected..\r
- if(reset) {\r
- for(AbsCassDAO<? extends TransStore, ?>.PSInfo psi : psinfos) {\r
- psi.reset();\r
- }\r
- }\r
- if(reset || cluster==null) {\r
- cluster = CassAccess.cluster(trans, keyspace);\r
- trans.warn().log(NEW_CASSANDRA_CLUSTER_OBJECT_CREATED);\r
- }\r
- if(reset || session==null) {\r
- session = cluster.connect(keyspace);\r
- trans.warn().log(NEW_CASSANDRA_SESSION_CREATED);\r
- }\r
- } finally {\r
- resetTrigger=false;\r
- tt.done();\r
- }\r
- }\r
- }\r
- } finally {\r
- TimeTaken tt = trans.start("Clear Reset Deque", Env.SUB);\r
- try {\r
- resetDeque.clear();\r
- // Not clearing Session/Cluster appears to kill off FileHandles\r
- if(tempSession!=null && !tempSession.isClosed()) {\r
- tempSession.close();\r
- }\r
- if(tempCluster!=null && !tempCluster.isClosed()) {\r
- tempCluster.close();\r
- }\r
- } finally {\r
- tt.done();\r
- }\r
- }\r
- }\r
- return session;\r
- }\r
- \r
- public final boolean reportPerhapsReset(TransStore trans, Exception e) {\r
- if(owningDAO!=null) {\r
- return owningDAO.reportPerhapsReset(trans, e);\r
- } else {\r
- boolean rv = false;\r
- if(CassAccess.isResetException(e)) {\r
- trans.warn().printf("Session Reset called for %s by %s ",session==null?"":session,e==null?"Mgmt Command":e.getClass().getName());\r
- resetDeque.addFirst(new ResetRequest(session));\r
- rv = resetTrigger = true;\r
- } \r
- trans.error().log(e);\r
- return rv;\r
- }\r
- }\r
-\r
- public void close(TransStore trans) {\r
- if(owningDAO==null) {\r
- if(session!=null) {\r
- TimeTaken tt = trans.start("Cassandra Session Close", Env.SUB);\r
- try {\r
- session.close();\r
- } finally {\r
- tt.done();\r
- }\r
- session = null;\r
- } else {\r
- trans.debug().log("close called(), Session already closed");\r
- }\r
- } else {\r
- owningDAO.close(trans);\r
- }\r
- }\r
-\r
- protected void wasModified(TRANS trans, CRUD modified, DATA data, String ... override) {\r
- }\r
- \r
- protected interface Accept<DATA> {\r
- public boolean ok(DATA data);\r
- }\r
-\r
-}\r
-\r
-\r
-\r
Code Review / aaf / authz.git / blobdiff