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