1 package fr.ifremer.quadrige3.synchro.service.referential;
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26 import com.google.common.base.Joiner;
27 import com.google.common.base.Splitter;
28 import com.google.common.collect.*;
29 import fr.ifremer.common.synchro.SynchroTechnicalException;
30 import fr.ifremer.common.synchro.config.SynchroConfiguration;
31 import fr.ifremer.common.synchro.dao.DaoFactory;
32 import fr.ifremer.common.synchro.dao.SynchroBaseDao;
33 import fr.ifremer.common.synchro.dao.SynchroTableDao;
34 import fr.ifremer.common.synchro.meta.*;
35 import fr.ifremer.common.synchro.service.*;
36 import fr.ifremer.quadrige3.core.config.QuadrigeConfiguration;
37 import fr.ifremer.quadrige3.core.dao.ObjectTypes;
38 import fr.ifremer.quadrige3.core.dao.technical.Assert;
39 import fr.ifremer.quadrige3.core.dao.technical.Daos;
40 import fr.ifremer.quadrige3.core.dao.technical.hibernate.TemporaryDataHelper;
41 import fr.ifremer.quadrige3.synchro.meta.DatabaseColumns;
42 import fr.ifremer.quadrige3.synchro.meta.referential.ReferentialSynchroTables;
43 import fr.ifremer.quadrige3.synchro.meta.system.RuleSynchroTables;
44 import fr.ifremer.quadrige3.synchro.service.SynchroDirection;
45 import oracle.jdbc.OracleConnection;
46 import org.apache.commons.collections4.CollectionUtils;
47 import org.apache.commons.lang3.StringUtils;
48 import org.apache.commons.logging.Log;
49 import org.apache.commons.logging.LogFactory;
50 import org.nuiton.i18n.I18n;
51 import org.postgis.PGgeometry;
52 import org.postgresql.PGConnection;
53 import org.springframework.beans.factory.annotation.Autowired;
54 import org.springframework.stereotype.Service;
55
56 import javax.sql.DataSource;
57 import java.io.File;
58 import java.sql.Connection;
59 import java.sql.ResultSet;
60 import java.sql.SQLException;
61 import java.sql.Timestamp;
62 import java.util.*;
63
64
65
66
67
68
69
70 @Service("referentialSynchroService")
71 public class ReferentialSynchroServiceImpl
72 extends SynchroServiceImpl<ReferentialSynchroDatabaseConfiguration, ReferentialSynchroContext>
73 implements ReferentialSynchroService {
74
75 private static final Log LOG =
76 LogFactory.getLog(ReferentialSynchroServiceImpl.class);
77
78
79
80
81 private static final String TQP_DELETE_BY_COMPARISON_PREFIX = "DELETE#";
82 private static final int TQP_DEFAULT_PERSON_ID = -1;
83
84 private static final boolean DISABLE_INTEGRITY_CONSTRAINTS = true;
85 private static final boolean ALLOW_MISSING_OPTIONAL_COLUMN = true;
86 private static final boolean ALLOW_ADDITIONAL_MANDATORY_COLUMN_IN_SOURCE_SCHEMA = true;
87 private static final boolean KEEP_WHERE_CLAUSE_ON_QUERIES_BY_FKS = true;
88
89
90
91 private static final int DAO_CACHE_SIZE = 2;
92
93
94
95
96
97
98
99
100
101
102
103 @Autowired
104 public ReferentialSynchroServiceImpl(DataSource dataSource, SynchroConfiguration config) {
105 super(dataSource, config,
106 DISABLE_INTEGRITY_CONSTRAINTS,
107 ALLOW_MISSING_OPTIONAL_COLUMN,
108 ALLOW_ADDITIONAL_MANDATORY_COLUMN_IN_SOURCE_SCHEMA,
109 KEEP_WHERE_CLAUSE_ON_QUERIES_BY_FKS);
110 setDaoCacheSize(DAO_CACHE_SIZE);
111 }
112
113
114
115
116
117
118 public ReferentialSynchroServiceImpl() {
119 super(DISABLE_INTEGRITY_CONSTRAINTS,
120 ALLOW_MISSING_OPTIONAL_COLUMN,
121 ALLOW_ADDITIONAL_MANDATORY_COLUMN_IN_SOURCE_SCHEMA,
122 KEEP_WHERE_CLAUSE_ON_QUERIES_BY_FKS);
123 setDaoCacheSize(DAO_CACHE_SIZE);
124 }
125
126
127 @Override
128 public ReferentialSynchroContext createSynchroContext(File sourceDbDirectory,
129 Timestamp lastSynchronizationDate,
130 boolean enableDelete,
131 boolean enableInsertUpdate,
132 SynchroDirection direction,
133 int userId,
134 Set<String> statusCodeIncludes) {
135
136 ReferentialSynchroContext context = super.createSynchroContext(sourceDbDirectory, ReferentialSynchroTables.getImportTablesIncludes());
137
138 context.getTarget().putAllProperties(QuadrigeConfiguration.getInstance().getConnectionProperties());
139 context.setDirection(direction);
140 context.setUserId(userId);
141 context.setLastSynchronizationDate(lastSynchronizationDate);
142 context.setEnableDelete(enableDelete);
143 context.setEnableInsertOrUpdate(enableInsertUpdate);
144 context.setStatusCodeIncludes(statusCodeIncludes);
145 initContext(context);
146
147 return context;
148 }
149
150
151 @Override
152 public ReferentialSynchroContext createSynchroContext(Properties sourceConnectionProperties,
153 Timestamp lastSynchronizationDate,
154 boolean enableDelete,
155 boolean enableInsertUpdate,
156 SynchroDirection direction,
157 int userId,
158 Set<String> statusCodeIncludes) {
159
160 ReferentialSynchroContext context = super.createSynchroContext(sourceConnectionProperties, ReferentialSynchroTables.getImportTablesIncludes());
161
162 context.getTarget().putAllProperties(QuadrigeConfiguration.getInstance().getConnectionProperties());
163 context.setDirection(direction);
164 context.setUserId(userId);
165 context.setLastSynchronizationDate(lastSynchronizationDate);
166 context.setEnableDelete(enableDelete);
167 context.setEnableInsertOrUpdate(enableInsertUpdate);
168 context.setStatusCodeIncludes(statusCodeIncludes);
169 initContext(context);
170
171 return context;
172 }
173
174
175 @Override
176 public void prepare(ReferentialSynchroContext synchroContext) {
177 Assert.isInstanceOf(ReferentialSynchroContext.class, synchroContext,
178 String.format("The context must be a instance of %s", ReferentialSynchroContext.class.getName()));
179
180 SynchroDirection direction = synchroContext.getDirection();
181
182 ReferentialSynchroDatabaseConfiguration target = synchroContext.getTarget();
183 ReferentialSynchroDatabaseConfiguration source = synchroContext.getSource();
184
185
186 target.excludeUnusedColumns();
187 source.excludeUnusedColumns();
188
189
190 if (direction == SynchroDirection.IMPORT_FILE2LOCAL) {
191 source.setIsMirrorDatabase(true);
192 target.setIsMirrorDatabase(false);
193
194 source.setIsTemporary(false);
195 target.setIsTemporary(false);
196
197
198 target.excludeUnusedColumnsForFileImport();
199 source.excludeUnusedColumnsForFileImport();
200 }
201
202 super.prepare(synchroContext);
203 }
204
205
206 @Override
207 public void synchronize(ReferentialSynchroContext synchroContext) {
208 Assert.isInstanceOf(ReferentialSynchroContext.class, synchroContext,
209 String.format("The context must be a instance of %s", ReferentialSynchroContext.class.getName()));
210
211 SynchroDirection direction = synchroContext.getDirection();
212
213 ReferentialSynchroDatabaseConfiguration target = synchroContext.getTarget();
214 ReferentialSynchroDatabaseConfiguration source = synchroContext.getSource();
215
216
217 if (direction == SynchroDirection.IMPORT_FILE2LOCAL) {
218 source.setIsMirrorDatabase(true);
219 target.setIsMirrorDatabase(false);
220
221 source.setIsTemporary(false);
222 target.setIsTemporary(false);
223 }
224
225 super.synchronize(synchroContext);
226 }
227
228
229
230
231
232
233
234
235
236
237
238 protected void initContext(ReferentialSynchroContext context) {
239
240
241 context.getTarget().setColumnUpdateDate(DatabaseColumns.UPDATE_DT.name().toLowerCase());
242 context.getSource().setColumnUpdateDate(DatabaseColumns.UPDATE_DT.name().toLowerCase());
243
244
245 {
246 Set<String> statusCodeIncludes = context.getStatusCodeIncludes();
247
248 if (CollectionUtils.isEmpty(statusCodeIncludes)) {
249 String configValue = QuadrigeConfiguration.getInstance().getImportReferentialStatusIncludes();
250 if (StringUtils.isNotBlank(configValue)) {
251 statusCodeIncludes = Sets.newHashSet(Splitter.on(',').split(configValue));
252 context.setStatusCodeIncludes(statusCodeIncludes);
253 }
254 }
255 }
256
257
258 if (Daos.isPostgresqlDatabase(context.getSource().getJdbcUrl())) {
259 context.getSource().setTempQueryParameterGenerated(config.isTempQueryParameterGenerated());
260 }
261 if (Daos.isPostgresqlDatabase(context.getTarget().getJdbcUrl())) {
262 context.getTarget().setTempQueryParameterGenerated(config.isTempQueryParameterGenerated());
263 }
264
265 }
266
267
268 @Override
269 protected void prepareRootTable(
270 DaoFactory sourceDaoFactory,
271 DaoFactory targetDaoFactory,
272 SynchroTableMetadata table,
273 ReferentialSynchroContext context,
274 SynchroResult result) throws SQLException {
275
276
277
278 if (context.isEnableInsertOrUpdate()
279 && (context.isEnableDelete()
280 || !ReferentialSynchroTables.DELETED_ITEM_HISTORY.name().equalsIgnoreCase(table.getName()))) {
281
282 super.prepareRootTable(sourceDaoFactory,
283 targetDaoFactory,
284 table,
285 context,
286 result);
287 }
288
289
290
291 if (context.isEnableDelete()
292 && !ReferentialSynchroTables.DELETED_ITEM_HISTORY.name().equalsIgnoreCase(table.getName())) {
293 prepareRootTableDeletes(sourceDaoFactory,
294 targetDaoFactory,
295 table,
296 context,
297 result);
298 }
299
300
301
302
303
304
305
306
307
308
309 }
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327 protected void prepareRootTableDeletes(
328 DaoFactory sourceDaoFactory,
329 DaoFactory targetDaoFactory,
330 SynchroTableMetadata table,
331 ReferentialSynchroContext context,
332 SynchroResult result) throws SQLException {
333
334 String tableName = table.getName();
335 Set<String> objectTypeFks = ObjectTypes.getObjectTypeFromTableName(tableName, tableName);
336
337 if (CollectionUtils.isEmpty(objectTypeFks)) {
338 return;
339 }
340
341 SynchroTableDao dihSourceDao = sourceDaoFactory.getSourceDao(ReferentialSynchroTables.DELETED_ITEM_HISTORY.name());
342
343 List<List<Object>> columnValues = Lists.newArrayListWithCapacity(objectTypeFks.size());
344 for (String objectTypeFk : objectTypeFks) {
345 columnValues.add(ImmutableList.of(objectTypeFk));
346 }
347
348
349 Map<String, Object> bindings = createSelectBindingsForTable(context, ReferentialSynchroTables.DELETED_ITEM_HISTORY.name());
350 long count = dihSourceDao.countDataByFks(
351 ImmutableSet.of(DatabaseColumns.OBJECT_TYPE_CD.name()),
352 columnValues,
353 bindings
354 );
355 if (count > 0) {
356 result.addRows(tableName, (int) count);
357 }
358 }
359
360
361 @Override
362 protected List<SynchroTableOperation> getRootOperations(
363 DaoFactory sourceDaoFactory,
364 DaoFactory targetDaoFactory,
365 SynchroDatabaseMetadata dbMeta,
366 ReferentialSynchroContext context) throws SQLException {
367 List<SynchroTableOperation> result = Lists.newArrayList();
368
369
370 if (context.isEnableInsertOrUpdate()) {
371 Collection<SynchroTableOperation> defaultOperations = super.getRootOperations(sourceDaoFactory, targetDaoFactory, dbMeta, context);
372 result.addAll(defaultOperations);
373 }
374
375
376 if (context.isEnableDelete()) {
377 Collection<SynchroTableOperation> deletedItemOperations = getRootDeleteOperations(sourceDaoFactory, targetDaoFactory, dbMeta, context);
378 result.addAll(deletedItemOperations);
379 }
380
381
382
383
384
385
386
387
388
389 return result;
390 }
391
392 private Collection<SynchroTableOperation> getRootDeleteOperations(
393 DaoFactory sourceDaoFactory,
394 DaoFactory targetDaoFactory,
395 SynchroDatabaseMetadata dbMeta,
396 ReferentialSynchroContext context) throws SQLException {
397 Assert.isTrue(dbMeta.getConfiguration().isFullMetadataEnable());
398
399 Deque<SynchroTableOperation> result = Queues.newArrayDeque();
400 Map<String, SynchroTableOperation> deletedChildrenOperationByTable = Maps.newHashMap();
401
402
403 Multimap<String, String> pksToIgnoreByTableName = getPksToIgnoreByTableName(sourceDaoFactory, context);
404
405 SynchroTableDao dihSourceDao = sourceDaoFactory.getSourceDao(ReferentialSynchroTables.DELETED_ITEM_HISTORY.name());
406 Set<String> includedDataTables = context.getTableNames();
407
408
409 Map<String, Object> bindings = createSelectBindingsForTable(context, ReferentialSynchroTables.DELETED_ITEM_HISTORY.name());
410
411 ResultSet dihResultSet = null;
412 List<Object> dihIdsToRemove = Lists.newArrayList();
413 Set<String> tableNamesWithDelete = Sets.newHashSet();
414 boolean doDelete = !context.getTarget().isMirrorDatabase();
415
416 try {
417 LOG.debug(I18n.t("quadrige3.synchro.synchronizeReferential.deletedItems"));
418
419 dihResultSet = dihSourceDao.getData(bindings);
420
421 while (dihResultSet.next()) {
422
423 String objectType = dihResultSet.getString(DatabaseColumns.OBJECT_TYPE_CD.name());
424 String tableName = ObjectTypes.getTableNameFromObjectType(objectType);
425
426 boolean isReferentialTable = StringUtils.isNotBlank(tableName)
427 && includedDataTables.contains(tableName.toUpperCase());
428
429 if (isReferentialTable) {
430 SynchroTableDao targetDao = targetDaoFactory.getSourceDao(tableName);
431 SynchroTableMetadata table = targetDao.getTable();
432
433 String objectCode = dihResultSet.getString(DatabaseColumns.OBJECT_CD.name());
434 String objectId = dihResultSet.getString(DatabaseColumns.OBJECT_ID.name());
435
436
437 if (StringUtils.isNotBlank(objectCode) || StringUtils.isNotBlank(objectId)) {
438
439 if (doDelete) {
440 List<Object> pk;
441
442 if (StringUtils.isNotBlank(objectCode) && !table.isSimpleKey()) {
443 pk = SynchroTableMetadata.fromPkStr(objectCode);
444 }
445 else {
446 pk = StringUtils.isNotBlank(objectCode)
447 ? ImmutableList.of(objectCode)
448 : ImmutableList.of(objectId);
449 }
450
451
452 if (pksToIgnoreByTableName != null) {
453 Collection<String> pksToIgnore = pksToIgnoreByTableName.get(tableName);
454 if (pksToIgnore != null && pksToIgnore.contains(SynchroTableMetadata.toPkStr(pk))) {
455
456 continue;
457 }
458 }
459
460 boolean hasChildTables = table.hasChildJoins();
461
462
463 if (hasChildTables) {
464
465 SynchroTableOperation childrenOperation = deletedChildrenOperationByTable.get(tableName);
466 if (childrenOperation == null) {
467 childrenOperation = new SynchroTableOperation(tableName, context);
468 deletedChildrenOperationByTable.put(tableName, childrenOperation);
469 result.add(childrenOperation);
470 }
471 addChildrenToDelete(table, childrenOperation, ImmutableList.of(pk), result, context);
472 }
473
474
475
476
477
478 SynchroTableOperation operation = new SynchroTableOperation(tableName, context);
479
480 operation.setEnableProgress(true);
481 result.add(operation);
482
483 operation.addMissingDelete(pk);
484 }
485 }
486
487
488
489 else {
490 tableNamesWithDelete.add(tableName);
491 }
492 }
493 }
494 } finally {
495 Daos.closeSilently(dihResultSet);
496 }
497
498 if (CollectionUtils.isNotEmpty(dihIdsToRemove)) {
499 SynchroTableOperation operation = new SynchroTableOperation(ReferentialSynchroTables.DELETED_ITEM_HISTORY.name(), context);
500 operation.addChildrenToDeleteFromOneColumn(ReferentialSynchroTables.DELETED_ITEM_HISTORY.name(), DatabaseColumns.REMOTE_ID.name(),
501 dihIdsToRemove);
502 result.add(operation);
503 }
504
505 if (CollectionUtils.isNotEmpty(tableNamesWithDelete)) {
506
507
508 if (!doDelete) {
509 saveTablesWithDelete(tableNamesWithDelete,
510 sourceDaoFactory,
511 targetDaoFactory,
512 dbMeta,
513 context);
514 }
515
516
517 else if (context.getSource().isMirrorDatabase()) {
518 addDeletedItemsFromTables(
519 tableNamesWithDelete,
520 result,
521 sourceDaoFactory,
522 targetDaoFactory,
523 dbMeta,
524 context);
525 }
526
527
528
529 else {
530 saveTablesWithDelete(tableNamesWithDelete,
531 sourceDaoFactory,
532 targetDaoFactory,
533 dbMeta,
534 context);
535
536 addDeletedItemsFromTables(
537 tableNamesWithDelete,
538 result,
539 targetDaoFactory,
540 targetDaoFactory,
541 dbMeta,
542 context);
543 }
544 }
545
546 return result;
547 }
548
549 private Multimap<String, String> getPksToIgnoreByTableName(DaoFactory sourceDaoFactory, ReferentialSynchroContext context) throws SQLException {
550
551
552 if (context.getDirection() != SynchroDirection.IMPORT_TEMP2LOCAL) return null;
553
554 Multimap<String, String> result = HashMultimap.create();
555
556
557 addPksByTableName(sourceDaoFactory, RuleSynchroTables.RULE_LIST.name(), result);
558 addPksByTableName(sourceDaoFactory, RuleSynchroTables.RULE.name(), result);
559
560 return result;
561 }
562
563 private void addPksByTableName(DaoFactory sourceDaoFactory, String tableName, Multimap<String, String> result) throws SQLException {
564 SynchroTableDao sourceDao = sourceDaoFactory.getSourceDao(tableName);
565 result.putAll(tableName, sourceDao.getPksStr());
566 }
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584 protected final void addChildrenToDelete(
585 SynchroTableMetadata parentTable,
586 SynchroTableOperation childrenOperation,
587 List<List<Object>> parentPks,
588 Deque<SynchroTableOperation> pendingOperations,
589 ReferentialSynchroContext context) {
590
591 Set<String> pkNames = parentTable.getPkNames();
592
593
594 if (pkNames.size() > 1) {
595 throw new UnsupportedOperationException("Not sure of this implementation: please check before comment out this exception !");
596 }
597
598
599 for (SynchroJoinMetadata join : parentTable.getChildJoins()) {
600 SynchroTableMetadata childTable = join.getTargetTable();
601 SynchroColumnMetadata childTableColumn = join.getTargetColumn();
602
603
604 childrenOperation.addChildrenToDeleteFromManyColumns(childTable.getName(), ImmutableSet.of(childTableColumn.getName()), parentPks);
605 }
606 }
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626 protected void saveTablesWithDelete(
627 Set<String> tableNames,
628 DaoFactory sourceDaoFactory,
629 DaoFactory targetDaoFactory,
630 SynchroDatabaseMetadata dbMeta,
631 ReferentialSynchroContext context) throws SQLException {
632
633 SynchroBaseDao targetBaseDao = targetDaoFactory.getDao();
634
635
636 targetBaseDao.executeDeleteTempQueryParameter(TQP_DELETE_BY_COMPARISON_PREFIX + "%", true, TQP_DEFAULT_PERSON_ID);
637
638 for (String tableName : tableNames) {
639
640 SynchroTableDao sourceDao = sourceDaoFactory.getSourceDao(tableName);
641 Set<String> pkStrs = sourceDao.getPksStr();
642
643 targetBaseDao.executeInsertIntoTempQueryParameter(
644 ImmutableList.copyOf(pkStrs),
645 TQP_DELETE_BY_COMPARISON_PREFIX + tableName,
646 TQP_DEFAULT_PERSON_ID
647 );
648 }
649 }
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670 protected void addDeletedItemsFromTables(
671 Set<String> tableNames,
672 Deque<SynchroTableOperation> operations,
673 DaoFactory sourceDaoFactory,
674 DaoFactory targetDaoFactory,
675 SynchroDatabaseMetadata dbMeta,
676 ReferentialSynchroContext context) throws SQLException {
677
678
679 SynchroTableDao tqpDao = sourceDaoFactory.getSourceDao(SynchroBaseDao.TEMP_QUERY_PARAMETER_TABLE);
680 int tqpValueColumnIndex = tqpDao.getTable().getColumnIndex("ALPHANUMERICAL_VALUE");
681 Assert.isTrue(tqpValueColumnIndex != -1);
682
683
684 Map<String, Object> emptyBinding = Maps.newHashMap();
685 Set<String> fkNames = ImmutableSet.of("PARAMETER_NAME");
686
687
688 for (String tableName : tableNames) {
689 SynchroTableMetadata table = dbMeta.getTable(tableName);
690 boolean hasChildTables = table.hasChildJoins();
691 Set<String> tablePkNames = table.getPkNames();
692 int pkCount = tablePkNames.size();
693
694
695 List<Object> fkValue = ImmutableList.of(TQP_DELETE_BY_COMPARISON_PREFIX + tableName);
696 ResultSet rs = tqpDao.getDataByFks(
697 fkNames,
698 ImmutableList.of(fkValue),
699 emptyBinding);
700 List<List<Object>> pks = Lists.newArrayList();
701 while (rs.next()) {
702 String pkStr = rs.getString(tqpValueColumnIndex + 1);
703 List<Object> pk = SynchroTableMetadata.fromPkStr(pkStr);
704
705
706 if (pkCount != pk.size()) {
707 String expectedPkStrExample = Joiner.on(String.format(">%s<", SynchroTableMetadata.PK_SEPARATOR)).join(tablePkNames);
708 throw new SynchroTechnicalException(String.format(
709 "Unable to import delete on %s: invalid PK found in the source database (in %s). Should have %s column (e.g. %s).",
710 tableName,
711 SynchroBaseDao.TEMP_QUERY_PARAMETER_TABLE,
712 pkCount,
713 expectedPkStrExample));
714 }
715 pks.add(pk);
716 }
717 rs.close();
718
719
720 if (pks.size() == 0) {
721 throw new SynchroTechnicalException(String.format(
722 "Unable to import delete on %s: No PK found in the source database (in %s). Unable to compare and find PKs to delete.",
723 tableName,
724 SynchroBaseDao.TEMP_QUERY_PARAMETER_TABLE));
725 }
726
727 SynchroTableOperation operation = new SynchroTableOperation(tableName, context);
728 operation.setEnableProgress(true);
729 SynchroTableDao targetTableDao = targetDaoFactory.getTargetDao(tableName, null, operation);
730
731 List<List<Object>> pksToDelete = targetTableDao.getPksByNotFoundFks(
732 tablePkNames,
733 pks,
734 emptyBinding);
735
736
737 pksToDelete = filterExcludeTemporary(table, pksToDelete);
738
739
740 if (CollectionUtils.isNotEmpty(pksToDelete)) {
741
742 operation.addAllMissingDelete(pksToDelete);
743
744
745 if (hasChildTables) {
746 addDeleteChildrenToDeque(table, pksToDelete, operations, context);
747 }
748
749
750 operations.add(operation);
751 }
752
753 }
754
755
756 targetDaoFactory.getDao().executeDeleteTempQueryParameter(TQP_DELETE_BY_COMPARISON_PREFIX + "%", true, TQP_DEFAULT_PERSON_ID);
757 }
758
759
760
761
762
763
764
765
766
767
768
769
770 protected List<List<Object>> filterExcludeTemporary(
771 SynchroTableMetadata table,
772 List<List<Object>> pks) {
773 Set<String> tablePkNames = table.getPkNames();
774
775 if (tablePkNames.size() > 1) {
776 return pks;
777 }
778
779 SynchroColumnMetadata pkColumn = table.getColumnMetadata(tablePkNames.iterator().next());
780 Collection<List<Object>> result;
781
782
783 if (SynchroMetadataUtils.isNumericType(pkColumn)) {
784 result = Collections2.filter(pks,
785 input -> {
786 Long pk = input != null ? Daos.convertToLong(input.get(0)) : null;
787 return !TemporaryDataHelper.isTemporaryId(pk);
788 });
789 }
790
791
792 else {
793
794 result = Collections2.filter(pks,
795 input -> {
796 String pk = input != null ? input.get(0).toString() : null;
797 return !TemporaryDataHelper.isTemporaryCode(pk);
798 });
799 }
800
801 return ImmutableList.copyOf(result);
802 }
803
804
805 @Override
806 protected Map<String, Object> createDefaultSelectBindings(ReferentialSynchroContext context) {
807
808 Map<String, Object> bindings = super.createDefaultSelectBindings(context);
809
810
811 bindings.put("userId", context.getUserId());
812
813 return bindings;
814 }
815
816
817 @Override
818 protected Map<String, Object> createSelectBindingsForTable(ReferentialSynchroContext context, String tableName) {
819
820 Map<String, Object> result = super.createSelectBindingsForTable(context, tableName);
821
822
823
824 if (CollectionUtils.isNotEmpty(context.getTableNamesForced())
825 && context.getTableNamesForced().contains(tableName.toUpperCase())) {
826 LOG.debug(String.format("[%s] Forced synchronization (last synchronization date ignored)", tableName));
827 result.remove(SynchroTableMetadata.UPDATE_DATE_BINDPARAM);
828 }
829
830 return result;
831 }
832
833 @Override
834 protected Connection createConnection(SynchroDatabaseConfiguration databaseConfiguration) throws SQLException {
835 Connection connection = super.createConnection(databaseConfiguration);
836
837
838 if (Daos.isOracleDatabase(databaseConfiguration.getJdbcUrl()) && databaseConfiguration.isSynonymsEnable()) {
839 ((OracleConnection) Daos.unwrapConnection(connection)).setIncludeSynonyms(true);
840 }
841
842
843 if (Daos.isPostgresqlDatabase(databaseConfiguration.getJdbcUrl())) {
844 ((PGConnection) Daos.unwrapConnection(connection)).addDataType(String.format("\"%s\".\"geometry\"", QuadrigeConfiguration.getInstance().getPostgisSchema()), PGgeometry.class);
845 }
846
847
848 Daos.setTimezone(connection, QuadrigeConfiguration.getInstance().getDbTimezone());
849
850 return connection;
851 }
852
853 @Override
854 protected ReferentialSynchroDatabaseConfiguration newSynchroDatabaseConfiguration(
855 ReferentialSynchroContext context, Properties sourceConnectionProperties, boolean isTarget) {
856 return new ReferentialSynchroDatabaseConfiguration(context, sourceConnectionProperties, isTarget);
857 }
858
859 @Override
860 protected ReferentialSynchroContext newSynchroContext() {
861 return new ReferentialSynchroContext();
862 }
863 }