1 package org.synchronoss.cpo.cassandra;
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25 import com.datastax.oss.driver.api.core.CqlSession;
26 import com.datastax.oss.driver.api.core.cql.BatchStatement;
27 import com.datastax.oss.driver.api.core.cql.BatchType;
28 import com.datastax.oss.driver.api.core.cql.BatchableStatement;
29 import com.datastax.oss.driver.api.core.cql.BoundStatement;
30 import com.datastax.oss.driver.api.core.cql.ColumnDefinitions;
31 import com.datastax.oss.driver.api.core.cql.ResultSet;
32 import com.datastax.oss.driver.api.core.cql.Row;
33 import com.datastax.oss.driver.api.core.type.DataType;
34 import com.datastax.oss.driver.api.core.type.DataTypes;
35 import java.lang.reflect.Constructor;
36 import java.lang.reflect.InvocationTargetException;
37 import java.util.ArrayList;
38 import java.util.Collection;
39 import java.util.List;
40 import java.util.Spliterator;
41 import java.util.Spliterators;
42 import java.util.function.Consumer;
43 import java.util.stream.Stream;
44 import java.util.stream.StreamSupport;
45 import org.slf4j.Logger;
46 import org.slf4j.LoggerFactory;
47 import org.synchronoss.cpo.cassandra.meta.CassandraCpoAttribute;
48 import org.synchronoss.cpo.cassandra.meta.CassandraCpoMetaDescriptor;
49 import org.synchronoss.cpo.cassandra.meta.CassandraMethodMapper;
50 import org.synchronoss.cpo.cassandra.meta.CassandraResultSetCpoData;
51 import org.synchronoss.cpo.core.*;
52 import org.synchronoss.cpo.core.enums.Crud;
53 import org.synchronoss.cpo.core.helper.ExceptionHelper;
54 import org.synchronoss.cpo.core.meta.CpoMetaDescriptor;
55 import org.synchronoss.cpo.core.meta.DataTypeMapEntry;
56 import org.synchronoss.cpo.core.meta.domain.CpoAttribute;
57 import org.synchronoss.cpo.core.meta.domain.CpoClass;
58 import org.synchronoss.cpo.core.meta.domain.CpoFunction;
59
60
61
62
63
64
65
66 public class CassandraCpoAdapter extends CpoBaseAdapter<ClusterDataSource> {
67
68 private static final long serialVersionUID = 1L;
69
70 private static final Logger logger = LoggerFactory.getLogger(CassandraCpoAdapter.class);
71
72
73 private CassandraCpoMetaDescriptor metaDescriptor = null;
74
75
76 private final CassandraSessionStrategy sessionStrategy;
77
78 private static int unknownModifyCount = -1;
79
80
81
82
83
84
85
86
87 protected CassandraCpoAdapter(
88 CassandraCpoMetaDescriptor metaDescriptor, DataSourceInfo<ClusterDataSource> jdsiTrx)
89 throws CpoException {
90 super(jdsiTrx.getDataSourceName(), jdsiTrx.getFetchSize(), jdsiTrx.getBatchSize());
91 this.metaDescriptor = metaDescriptor;
92 setWriteDataSource(jdsiTrx.getDataSource());
93 setReadDataSource(jdsiTrx.getDataSource());
94 this.sessionStrategy = new CassandraSessionStrategy(getReadDataSource(), getWriteDataSource());
95 }
96
97
98
99
100
101
102
103
104
105
106
107 protected CassandraCpoAdapter(
108 CassandraCpoMetaDescriptor metaDescriptor,
109 DataSourceInfo<ClusterDataSource> jdsiWrite,
110 DataSourceInfo<ClusterDataSource> jdsiRead)
111 throws CpoException {
112 super(jdsiWrite.getDataSourceName(), jdsiWrite.getFetchSize(), jdsiWrite.getBatchSize());
113 this.metaDescriptor = metaDescriptor;
114 setWriteDataSource(jdsiWrite.getDataSource());
115 setReadDataSource(jdsiRead.getDataSource());
116 this.sessionStrategy = new CassandraSessionStrategy(getReadDataSource(), getWriteDataSource());
117 }
118
119
120
121
122
123
124
125
126
127
128 public static CassandraCpoAdapter getInstance(
129 CassandraCpoMetaDescriptor metaDescriptor, DataSourceInfo<ClusterDataSource> cdsiTrx)
130 throws CpoException {
131 String adapterKey = metaDescriptor + ":" + cdsiTrx.getDataSourceName();
132 CassandraCpoAdapter adapter = (CassandraCpoAdapter) findCpoAdapter(adapterKey);
133 if (adapter == null) {
134 adapter = new CassandraCpoAdapter(metaDescriptor, cdsiTrx);
135 addCpoAdapter(adapterKey, adapter);
136 }
137 return adapter;
138 }
139
140
141
142
143
144
145
146
147
148
149
150
151 public static CassandraCpoAdapter getInstance(
152 CassandraCpoMetaDescriptor metaDescriptor,
153 DataSourceInfo<ClusterDataSource> cdsiWrite,
154 DataSourceInfo<ClusterDataSource> cdsiRead)
155 throws CpoException {
156 String adapterKey =
157 metaDescriptor + ":" + cdsiWrite.getDataSourceName() + ":" + cdsiRead.getDataSourceName();
158 CassandraCpoAdapter adapter = (CassandraCpoAdapter) findCpoAdapter(adapterKey);
159 if (adapter == null) {
160 adapter = new CassandraCpoAdapter(metaDescriptor, cdsiWrite, cdsiRead);
161 addCpoAdapter(adapterKey, adapter);
162 }
163 return adapter;
164 }
165
166 @Override
167 public <T> long existsBean(CpoQuery query, T bean) throws CpoException {
168 CqlSession session = null;
169 long objCount = -1;
170
171 try {
172 session = getReadSession();
173
174 objCount = existsBean(query.groupName(), bean, session, query.wheres());
175 } catch (Exception e) {
176 throw new CpoException("existsBeans(CpoQuery, T) failed", e);
177 }
178
179 return objCount;
180 }
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196 protected <T> long existsBean(
197 String groupName, T bean, CqlSession session, Collection<CpoWhere> wheres)
198 throws CpoException {
199 long count = 0;
200 Logger localLogger = logger;
201
202 if (bean == null) {
203 throw new CpoException("NULL bean passed into existsBean");
204 }
205
206 try {
207 CpoClass cpoClass = metaDescriptor.getMetaClass(bean);
208 List<CpoFunction> cpoFunctions =
209 cpoClass.getFunctionGroup(Crud.EXIST, groupName).getFunctions();
210 localLogger = LoggerFactory.getLogger(cpoClass.getMetaClass());
211
212 for (CpoFunction cpoFunction : cpoFunctions) {
213 localLogger.info(cpoFunction.getExpression());
214 CassandraBoundStatementFactory boundStatementFactory =
215 new CassandraBoundStatementFactory(
216 session, this, cpoClass, cpoFunction, bean, wheres, null, null);
217 BoundStatement boundStatement = boundStatementFactory.getBoundStatement();
218
219 long qCount = 0;
220
221 ResultSet rs = session.execute(boundStatement);
222 boundStatementFactory.release();
223 ColumnDefinitions columnDefinitions = rs.getColumnDefinitions();
224
225
226
227 if (columnDefinitions.size() == 1 && isCountResult(columnDefinitions.get(0).getType())) {
228 Row next = rs.one();
229 if (next != null) {
230 qCount = next.getLong(0);
231 next = rs.one();
232 if (next != null) {
233
234 qCount = 2;
235 }
236 }
237 }
238
239 qCount += rs.all().size();
240
241 count += qCount;
242 }
243 } catch (Exception e) {
244 String msg = "existsBean(groupName, bean, session) failed:";
245 localLogger.error(msg, e);
246 throw new CpoException(msg, e);
247 }
248
249 return count;
250 }
251
252 private static boolean isCountResult(DataType type) {
253 return type.equals(DataTypes.BIGINT) || type.equals(DataTypes.COUNTER);
254 }
255
256
257
258
259
260
261 @Override
262 public CpoMetaDescriptor getCpoMetaDescriptor() {
263 return metaDescriptor;
264 }
265
266
267
268
269
270
271
272 protected CqlSession getReadSession() throws CpoException {
273 return sessionStrategy.getReadSession();
274 }
275
276
277
278
279
280
281
282 protected CqlSession getWriteSession() throws CpoException {
283 return sessionStrategy.getWriteSession();
284 }
285
286
287
288
289
290
291
292
293 @Override
294 public List<CpoAttribute> getCpoAttributes(String expression) throws CpoException {
295 List<CpoAttribute> attributes = new ArrayList<>();
296
297 if (expression != null && !expression.isEmpty()) {
298 CqlSession session;
299 ResultSet rs;
300 try {
301 session = getWriteSession();
302 rs = session.execute(expression);
303 ColumnDefinitions columnDefs = rs.getColumnDefinitions();
304 for (int i = 0; i < columnDefs.size(); i++) {
305 CpoAttribute attribute = new CassandraCpoAttribute();
306 String columnName = columnDefs.get(i).getName().asInternal();
307 attribute.setDataName(columnName);
308
309 DataTypeMapEntry<?> dataTypeMapEntry =
310 metaDescriptor.getDataTypeMapEntry(columnDefs.get(i).getType().getProtocolCode());
311 attribute.setDataType(dataTypeMapEntry.dataTypeName());
312 attribute.setDataTypeInt(dataTypeMapEntry.dataTypeInt());
313 attribute.setJavaType(dataTypeMapEntry.javaClass().getName());
314 attribute.setJavaName(dataTypeMapEntry.makeJavaName(columnName));
315
316 attributes.add(attribute);
317 }
318 } catch (Throwable t) {
319 logger.error(ExceptionHelper.getLocalizedMessage(t), t);
320 throw new CpoException("Error Generating Attributes", t);
321 }
322 }
323 return attributes;
324 }
325
326 private ResultSet executeBatchStatements(
327 CqlSession session, ArrayList<CassandraBoundStatementFactory> statementFactories)
328 throws Exception {
329 ResultSet resultSet;
330
331 ArrayList<BatchableStatement<?>> boundStatements = new ArrayList<>(statementFactories.size());
332
333 for (CassandraBoundStatementFactory factory : statementFactories) {
334 boundStatements.add(factory.getBoundStatement());
335 }
336
337 try {
338 BatchStatement batchStatement =
339 BatchStatement.builder(BatchType.LOGGED).addStatements(boundStatements).build();
340 resultSet = session.execute(batchStatement);
341 } finally {
342 for (CassandraBoundStatementFactory factory : statementFactories) {
343 factory.release();
344 }
345 }
346 return resultSet;
347 }
348
349 private ResultSet executeBoundStatement(
350 CqlSession session, CassandraBoundStatementFactory boundStatementFactory) throws Exception {
351 ResultSet resultSet;
352 try {
353 resultSet = session.execute(boundStatementFactory.getBoundStatement());
354 } finally {
355 boundStatementFactory.release();
356 }
357 return resultSet;
358 }
359
360 @Override
361 protected <T> long processUpdateGroup(
362 T bean,
363 Crud crud,
364 String groupName,
365 Collection<CpoWhere> wheres,
366 Collection<CpoOrderBy> orderBy,
367 Collection<CpoNativeFunction> nativeExpressions)
368 throws CpoException {
369 CqlSession sess = null;
370 long updateCount = 0;
371
372 try {
373 sess = getWriteSession();
374 updateCount =
375 processUpdateGroup(bean, crud, groupName, wheres, orderBy, nativeExpressions, sess);
376 } catch (Exception e) {
377
378 ExceptionHelper.reThrowCpoException(
379 e, "processUpdateGroup(T bean, Crud crud, String groupName) failed");
380 }
381
382 return updateCount;
383 }
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399 protected <T> long processUpdateGroup(
400 T bean,
401 Crud crud,
402 String groupName,
403 Collection<CpoWhere> wheres,
404 Collection<CpoOrderBy> orderBy,
405 Collection<CpoNativeFunction> nativeExpressions,
406 CqlSession sess)
407 throws CpoException {
408 Logger localLogger = bean == null ? logger : LoggerFactory.getLogger(bean.getClass());
409 CpoClass cpoClass;
410
411 if (bean == null) {
412 throw new CpoException(
413 "NULL bean passed into insertBean, deleteBean, updateBean, or upsertBean");
414 }
415
416 try {
417 cpoClass = metaDescriptor.getMetaClass(bean);
418 List<CpoFunction> cpoFunctions =
419 cpoClass
420 .getFunctionGroup(adjustCrud(bean, crud, groupName, sess), groupName)
421 .getFunctions();
422 localLogger.info(buildCpoClassLogLine(bean.getClass(), crud, groupName));
423
424 for (CpoFunction cpoFunction : cpoFunctions) {
425 CassandraBoundStatementFactory boundStatementFactory =
426 new CassandraBoundStatementFactory(
427 sess, this, cpoClass, cpoFunction, bean, wheres, orderBy, nativeExpressions);
428 executeBoundStatement(sess, boundStatementFactory);
429 }
430 localLogger.info(buildExecutedLogLine(bean.getClass(), crud, groupName));
431 } catch (Throwable t) {
432 String msg =
433 "ProcessUpdateGroup failed:"
434 + crud.operation
435 + ","
436 + groupName
437 + ","
438 + bean.getClass().getName();
439
440
441 localLogger.error(msg, t);
442 throw new CpoException(msg, t);
443 }
444
445 return unknownModifyCount;
446 }
447
448 @Override
449 protected <T> long processUpdateGroup(
450 List<T> beans,
451 Crud crud,
452 String groupName,
453 Collection<CpoWhere> wheres,
454 Collection<CpoOrderBy> orderBy,
455 Collection<CpoNativeFunction> nativeExpressions)
456 throws CpoException {
457 CqlSession sess;
458 long updateCount = 0;
459
460 try {
461 sess = getWriteSession();
462 updateCount =
463 processUpdateGroup(beans, crud, groupName, wheres, orderBy, nativeExpressions, sess);
464 } catch (Exception e) {
465
466 ExceptionHelper.reThrowCpoException(
467 e, "processUpdateGroup(Collection beans, Crud crud, String groupName) failed");
468 }
469
470 return updateCount;
471 }
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487 protected <T> long processUpdateGroup(
488 List<T> beans,
489 Crud crud,
490 String groupName,
491 Collection<CpoWhere> wheres,
492 Collection<CpoOrderBy> orderBy,
493 Collection<CpoNativeFunction> nativeExpressions,
494 CqlSession sess)
495 throws CpoException {
496 CpoClass cpoClass;
497 List<CpoFunction> cpoFunctions;
498 CassandraBoundStatementFactory boundStatementFactory = null;
499 Logger localLogger = logger;
500
501 if (beans.isEmpty()) return 0;
502 var beanInstance = beans.getFirst();
503
504 try {
505 cpoClass = metaDescriptor.getMetaClass(beanInstance);
506 cpoFunctions =
507 cpoClass
508 .getFunctionGroup(adjustCrud(beanInstance, crud, groupName, sess), groupName)
509 .getFunctions();
510 localLogger = LoggerFactory.getLogger(cpoClass.getMetaClass());
511
512 int numStatements = 0;
513 localLogger.info(buildCpoClassLogLine(beanInstance.getClass(), crud, groupName));
514 ArrayList<CassandraBoundStatementFactory> statemetnFactories = new ArrayList<>();
515 for (T bean : beans) {
516 for (CpoFunction function : cpoFunctions) {
517 boundStatementFactory =
518 new CassandraBoundStatementFactory(
519 sess, this, cpoClass, function, bean, wheres, orderBy, nativeExpressions);
520 statemetnFactories.add(boundStatementFactory);
521 numStatements++;
522 }
523 }
524
525 executeBatchStatements(sess, statemetnFactories);
526
527 localLogger.info(
528 buildUpdatesLogLine(numStatements, beanInstance.getClass(), crud, groupName));
529
530 } catch (Throwable t) {
531 String msg =
532 "ProcessUpdateGroup failed:"
533 + crud.operation
534 + ","
535 + groupName
536 + ","
537 + beanInstance.getClass().getName();
538
539
540 localLogger.error(msg, t);
541 throw new CpoException(msg, t);
542 }
543
544 return unknownModifyCount;
545 }
546
547 @Override
548 protected <T, C> T processExecuteGroup(String groupName, C criteria, T result)
549 throws CpoException {
550 throw new UnsupportedOperationException("Execute Functions not supported in Cassandra");
551 }
552
553 @Override
554 protected <T> T processSelectGroup(
555 T bean,
556 String groupName,
557 Collection<CpoWhere> wheres,
558 Collection<CpoOrderBy> orderBy,
559 Collection<CpoNativeFunction> nativeExpressions)
560 throws CpoException {
561 CqlSession session = null;
562 T result = null;
563
564 try {
565 session = getReadSession();
566 result = processSelectGroup(bean, groupName, wheres, orderBy, nativeExpressions, session);
567 } catch (Exception e) {
568 ExceptionHelper.reThrowCpoException(e, "processSelectGroup(T bean, String groupName) failed");
569 }
570
571 return result;
572 }
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592 protected <T> T processSelectGroup(
593 T bean,
594 String groupName,
595 Collection<CpoWhere> wheres,
596 Collection<CpoOrderBy> orderBy,
597 Collection<CpoNativeFunction> nativeExpressions,
598 CqlSession sess)
599 throws CpoException {
600 T criteriaObj = bean;
601 boolean recordsExist = false;
602 Logger localLogger = bean == null ? logger : LoggerFactory.getLogger(bean.getClass());
603
604 int recordCount = 0;
605 int attributesSet = 0;
606
607 T rObj = null;
608
609 if (bean == null) {
610 throw new CpoException("NULL bean passed into retrieveBean");
611 }
612
613 try {
614 CpoClass cpoClass = metaDescriptor.getMetaClass(criteriaObj);
615 List<CpoFunction> functions =
616 cpoClass.getFunctionGroup(Crud.RETRIEVE, groupName).getFunctions();
617
618 localLogger.info(buildCpoClassLogLine(criteriaObj.getClass(), Crud.RETRIEVE, groupName));
619
620 try {
621 rObj = (T) bean.getClass().getDeclaredConstructor().newInstance();
622 } catch (IllegalAccessException iae) {
623 localLogger.error(
624 "=================== Could not access default constructor for Class=<"
625 + bean.getClass()
626 + "> ==================");
627 throw new CpoException("Unable to access the constructor of the Return bean", iae);
628 } catch (InstantiationException iae) {
629 throw new CpoException("Unable to instantiate Return bean", iae);
630 }
631
632 for (CpoFunction cpoFunction : functions) {
633
634 CassandraBoundStatementFactory cbsf =
635 new CassandraBoundStatementFactory(
636 sess, this, cpoClass, cpoFunction, criteriaObj, wheres, orderBy, nativeExpressions);
637 BoundStatement boundStatement = cbsf.getBoundStatement();
638
639
640
641 ResultSet rs = sess.execute(boundStatement);
642 cbsf.release();
643
644 ColumnDefinitions columnDefs = rs.getColumnDefinitions();
645
646 if ((columnDefs.size() == 2)
647 && "CPO_ATTRIBUTE".equalsIgnoreCase(columnDefs.get(1).getName().asInternal())
648 && "CPO_VALUE".equalsIgnoreCase(columnDefs.get(2).getName().asInternal())) {
649 for (Row row : rs) {
650 recordsExist = true;
651 recordCount++;
652 CassandraCpoAttribute attribute =
653 (CassandraCpoAttribute) cpoClass.getAttributeData(row.getString(0));
654
655 if (attribute != null) {
656 attribute.invokeSetter(
657 rObj,
658 new CassandraResultSetCpoData(
659 CassandraMethodMapper.getMethodMapper(), row, attribute, 1));
660 attributesSet++;
661 }
662 }
663 } else {
664 Row row = rs.one();
665 if (row != null) {
666 recordsExist = true;
667 recordCount++;
668 for (int k = 0; k < columnDefs.size(); k++) {
669 CassandraCpoAttribute attribute =
670 (CassandraCpoAttribute)
671 cpoClass.getAttributeData(columnDefs.get(k).getName().asInternal());
672
673 if (attribute != null) {
674 attribute.invokeSetter(
675 rObj,
676 new CassandraResultSetCpoData(
677 CassandraMethodMapper.getMethodMapper(), row, attribute, k));
678 attributesSet++;
679 }
680 }
681
682 if (rs.one() != null) {
683 String msg = "processSelectGroup(T, String) failed: Multiple Records Returned";
684 localLogger.error(msg);
685 throw new CpoException(msg);
686 }
687 }
688 }
689 criteriaObj = rObj;
690 }
691
692 if (!recordsExist) {
693 rObj = null;
694 localLogger.info(
695 buildRecordsLogLine(0, 0, criteriaObj.getClass(), Crud.RETRIEVE, groupName));
696 } else {
697 localLogger.info(
698 buildRecordsLogLine(
699 recordCount, attributesSet, criteriaObj.getClass(), Crud.RETRIEVE, groupName));
700 }
701 } catch (Throwable t) {
702 String msg = "processSelectGroup(T) failed: " + ExceptionHelper.getLocalizedMessage(t);
703 localLogger.error(msg, t);
704 throw new CpoException(msg, t);
705 }
706
707 return rObj;
708 }
709
710 @Override
711 protected <T, C> Stream<T> processSelectGroup(
712 String groupName,
713 C criteria,
714 T result,
715 Collection<CpoWhere> wheres,
716 Collection<CpoOrderBy> orderBy,
717 Collection<CpoNativeFunction> nativeExpressions,
718 boolean useRetrieve)
719 throws CpoException {
720 CqlSession session = null;
721
722 try {
723 session = getReadSession();
724 return processSelectGroup(
725 groupName, criteria, result, wheres, orderBy, nativeExpressions, session, useRetrieve);
726 } catch (Exception e) {
727 ExceptionHelper.reThrowCpoException(
728 e,
729 "processSelectGroup(String groupName, C criteria, T result,CpoWhere where,"
730 + " Collection orderBy, boolean useRetrieve) failed");
731 }
732 return Stream.empty();
733 }
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751 protected <T, C> Stream<T> processSelectGroup(
752 String groupName,
753 C criteria,
754 T result,
755 Collection<CpoWhere> wheres,
756 Collection<CpoOrderBy> orderBy,
757 Collection<CpoNativeFunction> nativeExpressions,
758 CqlSession sess,
759 boolean useRetrieve)
760 throws CpoException {
761 Logger localLogger = criteria == null ? logger : LoggerFactory.getLogger(criteria.getClass());
762 CassandraBoundStatementFactory boundStatementFactory = null;
763 List<CpoFunction> cpoFunctions;
764 CpoClass criteriaClass;
765 CpoClass resultClass;
766
767 ColumnDefinitions columnDefs;
768 int columnCount;
769 CpoAttribute[] attributes;
770
771 if (criteria == null || result == null) {
772 throw new CpoException("NULL bean passed into retrieveBean or retrieveBeans");
773 }
774
775 try {
776 criteriaClass = metaDescriptor.getMetaClass(criteria);
777 resultClass = metaDescriptor.getMetaClass(result);
778 if (useRetrieve) {
779 localLogger.info(buildCpoClassLogLine(criteria.getClass(), Crud.RETRIEVE, groupName));
780 cpoFunctions = criteriaClass.getFunctionGroup(Crud.RETRIEVE, groupName).getFunctions();
781 } else {
782 localLogger.info(buildCpoClassLogLine(criteria.getClass(), Crud.LIST, groupName));
783 cpoFunctions = criteriaClass.getFunctionGroup(Crud.LIST, groupName).getFunctions();
784 }
785
786 CpoFunction cpoFunction = cpoFunctions.getFirst();
787 boundStatementFactory =
788 new CassandraBoundStatementFactory(
789 sess, this, criteriaClass, cpoFunction, criteria, wheres, orderBy, nativeExpressions);
790 BoundStatement boundStatement = boundStatementFactory.getBoundStatement();
791
792 localLogger.debug("Retrieving Records");
793
794 ResultSet rs = sess.execute(boundStatement);
795 boundStatementFactory.release();
796
797 localLogger.debug("Processing Records");
798
799 columnDefs = rs.getColumnDefinitions();
800
801 columnCount = columnDefs.size();
802
803 attributes = new CpoAttribute[columnCount];
804
805 for (int k = 0; k < columnCount; k++) {
806 attributes[k] = resultClass.getAttributeData(columnDefs.get(k).getName().asInternal());
807 }
808
809
810 Constructor<?> resultConstructor;
811 try {
812 resultConstructor = result.getClass().getDeclaredConstructor();
813 } catch (NoSuchMethodException e) {
814 throw new CpoException(
815 "Constructor not found for Return bean Class=<" + result.getClass() + ">", e);
816 }
817
818 CassandraBoundStatementFactory finalBoundStatementFactory = boundStatementFactory;
819 return StreamSupport.stream(
820 new Spliterators.AbstractSpliterator<T>(Long.MAX_VALUE, Spliterator.ORDERED) {
821 @Override
822 public boolean tryAdvance(Consumer<? super T> action) {
823 try {
824 Row row = rs.one();
825 if (row == null) return false;
826 T bean = null;
827 try {
828 bean = (T) resultConstructor.newInstance();
829 } catch (IllegalAccessException iae) {
830 String msg =
831 "Could not access default constructor for Class=<"
832 + result.getClass()
833 + ">";
834 throw new CpoException(msg, iae);
835 } catch (InstantiationException iae) {
836 throw new CpoException(
837 "Unable to instantiate Return bean for Class=<" + result.getClass() + ">",
838 iae);
839 } catch (InvocationTargetException e) {
840 throw new CpoException(
841 "Unable to invoke constructor for Return bean Class=<"
842 + result.getClass()
843 + ">",
844 e);
845 }
846
847 for (int k = 0; k < columnCount; k++) {
848 if (attributes[k] != null) {
849 attributes[k].invokeSetter(
850 bean,
851 new CassandraResultSetCpoData(
852 CassandraMethodMapper.getMethodMapper(), row, attributes[k], k));
853 }
854 }
855 action.accept(bean);
856 return true;
857 } catch (CpoException ex) {
858 throw new RuntimeException(ex);
859 }
860 }
861 },
862 false)
863 .onClose(
864 () -> {
865 try {
866 finalBoundStatementFactory.release();
867 } catch (CpoException e) {
868 throw new RuntimeException(e);
869 }
870 });
871 } catch (Throwable t) {
872 if (boundStatementFactory != null) boundStatementFactory.release();
873 String msg =
874 "processSelectGroup(String groupName, C criteria, T result, CpoWhere where,"
875 + " Collection orderBy, CqlSession sess) failed. Error:";
876 localLogger.error(msg, t);
877 throw new CpoException(msg, t);
878 }
879 }
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894 protected <T> Crud adjustCrud(T bean, Crud crud, String groupName, CqlSession session)
895 throws CpoException {
896 Crud retType = crud;
897 long objCount;
898
899 if (Crud.UPSERT == retType) {
900 objCount = existsBean(groupName, bean, session, null);
901
902 if (objCount == 0) {
903 retType = Crud.CREATE;
904 } else if (objCount == 1) {
905 retType = Crud.UPDATE;
906 } else {
907 throw new CpoException(
908 "UPSERT can only UPDATE one record. Your EXISTS function returned 2 or more.");
909 }
910 }
911
912 return retType;
913 }
914 }