diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/common/Constants.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/common/Constants.java index 70989c2452..4fc989c8df 100644 --- a/symmetric-core/src/main/java/org/jumpmind/symmetric/common/Constants.java +++ b/symmetric-core/src/main/java/org/jumpmind/symmetric/common/Constants.java @@ -32,7 +32,9 @@ private Constants() { public static final long VIRTUAL_BATCH_FOR_REGISTRATION = -9999; - public static final String NONE_TOKEN = "$(none)"; + public static final String NONE_TOKEN = "$(none)"; + + public static final String DATA_CONTEXT_ENGINE = "engine"; public static final String DATA_CONTEXT_TARGET_NODE = "targetNode"; diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/IDataService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/IDataService.java index 378f0224ae..70f552ae53 100644 --- a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/IDataService.java +++ b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/IDataService.java @@ -100,7 +100,7 @@ public long insertDataAndDataEventAndOutgoingBatch(ISqlTransaction transaction, public void insertSqlEvent(Node targetNode, String sql, boolean isLoad, long loadId, String createBy); - public void insertCreateEvent(Node targetNode, TriggerRouter triggerRouter, TriggerHistory triggerHistory, String xml, boolean isLoad, long loadId, String createBy); + public void insertCreateEvent(Node targetNode, TriggerHistory triggerHistory, String xml, boolean isLoad, long loadId, String createBy); /** * Count the number of data ids in a range diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/ITriggerRouterService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/ITriggerRouterService.java index 8c96cebc97..4b3be19bbe 100644 --- a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/ITriggerRouterService.java +++ b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/ITriggerRouterService.java @@ -180,6 +180,8 @@ public TriggerHistory getNewestTriggerHistoryForTrigger(String triggerId, String public Map> fillTriggerRoutersByHistIdAndSortHist( String sourceNodeGroupId, String targetNodeGroupId, List triggerHistories); + public TriggerHistory findTriggerHistoryForGenericSync(); + public void clearCache(); } \ No newline at end of file diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataLoaderService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataLoaderService.java index 6c0b06b607..6467166a68 100644 --- a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataLoaderService.java +++ b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataLoaderService.java @@ -374,6 +374,7 @@ protected List loadDataFromTransport(final ProcessInfo processInf Throwable error = null; try { Node targetNode = nodeService.findIdentity(); + ctx.put(Constants.DATA_CONTEXT_ENGINE, engine); if (targetNode != null) { ctx.put(Constants.DATA_CONTEXT_TARGET_NODE, targetNode); ctx.put(Constants.DATA_CONTEXT_TARGET_NODE_ID, targetNode.getNodeId()); diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataService.java index af921173fe..6077cbc034 100644 --- a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataService.java +++ b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataService.java @@ -427,8 +427,8 @@ private void insertCreateBatchesForReload(Node targetNode, long loadId, String c targetNode)) { String xml = symmetricDialect.getCreateTableXML(triggerHistory, triggerRouter); - insertCreateEvent(transaction, targetNode, triggerRouter, triggerHistory, - xml, true, loadId, createBy); + insertCreateEvent(transaction, targetNode, triggerHistory, xml, + true, loadId, createBy); if (!transactional) { transaction.commit(); } @@ -585,19 +585,8 @@ protected void insertPurgeEvent(ISqlTransaction transaction, Node targetNode, } } - private TriggerHistory findTriggerHistoryForGenericSync() { - String triggerTableName = TableConstants.getTableName(tablePrefix, - TableConstants.SYM_TRIGGER); - TriggerHistory history = engine.getTriggerRouterService().findTriggerHistory(null, null, triggerTableName - .toUpperCase()); - if (history == null) { - history = engine.getTriggerRouterService().findTriggerHistory(null, null, triggerTableName); - } - return history; - } - public void insertSqlEvent(Node targetNode, String sql, boolean isLoad, long loadId, String createBy) { - TriggerHistory history = findTriggerHistoryForGenericSync(); + TriggerHistory history = engine.getTriggerRouterService().findTriggerHistoryForGenericSync(); boolean useReloadChannel = parameterService.is(ParameterConstants.INITIAL_LOAD_USE_RELOAD_CHANNEL); Data data = new Data(history.getSourceTableName(), DataEventType.SQL, CsvUtils.escapeCsvData(sql), null, history, useReloadChannel && isLoad ? Constants.CHANNEL_RELOAD : Constants.CHANNEL_CONFIG, null, null); @@ -612,7 +601,7 @@ public void insertSqlEvent(Node targetNode, String sql, boolean isLoad, long loa public void insertSqlEvent(ISqlTransaction transaction, Node targetNode, String sql, boolean isLoad, long loadId, String createBy) { - TriggerHistory history = findTriggerHistoryForGenericSync(); + TriggerHistory history = engine.getTriggerRouterService().findTriggerHistoryForGenericSync(); insertSqlEvent(transaction, history, Constants.CHANNEL_CONFIG, targetNode, sql, isLoad, loadId, createBy); } @@ -647,12 +636,12 @@ public void checkForAndUpdateMissingChannelIds(long firstDataId, long lastDataId } } - public void insertCreateEvent(final Node targetNode, final TriggerRouter triggerRouter, - TriggerHistory triggerHistory, String xml, boolean isLoad, long loadId, String createBy) { + public void insertCreateEvent(final Node targetNode, TriggerHistory triggerHistory, + String xml, boolean isLoad, long loadId, String createBy) { ISqlTransaction transaction = null; try { transaction = sqlTemplate.startSqlTransaction(); - insertCreateEvent(transaction, targetNode, triggerRouter, triggerHistory, xml, isLoad, loadId, createBy); + insertCreateEvent(transaction, targetNode, triggerHistory, xml, isLoad, loadId, createBy); transaction.commit(); } catch (Error ex) { if (transaction != null) { @@ -670,8 +659,7 @@ public void insertCreateEvent(final Node targetNode, final TriggerRouter trigger } public void insertCreateEvent(ISqlTransaction transaction, Node targetNode, - TriggerRouter triggerRouter, TriggerHistory triggerHistory, String xml, boolean isLoad, long loadId, String createBy) { - Trigger trigger = triggerRouter.getTrigger(); + TriggerHistory triggerHistory, String xml, boolean isLoad, long loadId, String createBy) { Data data = new Data( triggerHistory.getSourceTableName(), DataEventType.CREATE, @@ -679,11 +667,11 @@ public void insertCreateEvent(ISqlTransaction transaction, Node targetNode, null, triggerHistory, parameterService.is(ParameterConstants.INITIAL_LOAD_USE_RELOAD_CHANNEL) && isLoad ? Constants.CHANNEL_RELOAD - : trigger.getChannelId(), null, null); + : Constants.CHANNEL_CONFIG, null, null); try { if (isLoad) { insertDataAndDataEventAndOutgoingBatch(transaction, data, targetNode.getNodeId(), - triggerRouter.getRouter().getRouterId(), isLoad, loadId, createBy, Status.NE); + Constants.UNKNOWN_ROUTER_ID, isLoad, loadId, createBy, Status.NE); } else { data.setNodeList(targetNode.getNodeId()); insertData(transaction, data); @@ -893,7 +881,7 @@ private void insertNodeSecurityUpdate(ISqlTransaction transaction, String nodeId public void sendScript(String nodeId, String script, boolean isLoad) { Node targetNode = engine.getNodeService().findNode(nodeId); - TriggerHistory history = findTriggerHistoryForGenericSync(); + TriggerHistory history = engine.getTriggerRouterService().findTriggerHistoryForGenericSync(); Data data = new Data(history.getSourceTableName(), DataEventType.BSH, CsvUtils.escapeCsvData(script), null, history, Constants.CHANNEL_CONFIG, null, null); data.setNodeList(nodeId); @@ -927,7 +915,7 @@ public boolean sendSchema(String nodeId, String catalogName, String schemaName, for (TriggerRouter triggerRouter : triggerRouters) { eventCount++; String xml = symmetricDialect.getCreateTableXML(triggerHistory, triggerRouter); - insertCreateEvent(targetNode, triggerRouter, triggerHistory, xml, false, -1, null); + insertCreateEvent(targetNode, triggerHistory, xml, false, -1, null); } } diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/TriggerRouterService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/TriggerRouterService.java index f3ba2bc392..0b6c4978d4 100644 --- a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/TriggerRouterService.java +++ b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/TriggerRouterService.java @@ -1796,6 +1796,18 @@ public void addExtraConfigTable(String table) { public Map getFailedTriggers() { return this.failureListener.getFailures(); } + + public TriggerHistory findTriggerHistoryForGenericSync() { + String triggerTableName = TableConstants.getTableName(tablePrefix, + TableConstants.SYM_TRIGGER); + TriggerHistory history = findTriggerHistory(null, null, triggerTableName + .toUpperCase()); + if (history == null) { + history = findTriggerHistory(null, null, triggerTableName); + } + return history; + } + public Map> fillTriggerRoutersByHistIdAndSortHist( String sourceNodeGroupId, String targetNodeGroupId, List triggerHistories) { diff --git a/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/DataEventType.java b/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/DataEventType.java index 54303d444d..8b4e51a008 100644 --- a/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/DataEventType.java +++ b/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/DataEventType.java @@ -62,8 +62,7 @@ public enum DataEventType { * An event the indicates that the data payload is going to be a Java bean shell script that is to be run at the * client. */ - BSH("B"), - + BSH("B"), UNKNOWN("U"); diff --git a/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/writer/DatabaseWriter.java b/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/writer/DatabaseWriter.java index 412ef1fa44..b6bbd64e61 100644 --- a/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/writer/DatabaseWriter.java +++ b/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/writer/DatabaseWriter.java @@ -871,6 +871,7 @@ protected boolean script(CsvData data) { Map variables = new HashMap(); variables.put("SOURCE_NODE_ID", batch.getSourceNodeId()); variables.put("TARGET_NODE_ID", batch.getTargetNodeId()); + variables.putAll(context.getContext()); ISqlTemplate template = platform.getSqlTemplate(); Class templateClass = template.getClass(); if (templateClass.getSimpleName().equals("JdbcSqlTemplate")) { diff --git a/symmetric-util/src/main/java/org/jumpmind/util/Context.java b/symmetric-util/src/main/java/org/jumpmind/util/Context.java index d5cf37070a..a9154b3b91 100644 --- a/symmetric-util/src/main/java/org/jumpmind/util/Context.java +++ b/symmetric-util/src/main/java/org/jumpmind/util/Context.java @@ -44,4 +44,8 @@ public Set keySet() { return context.keySet(); } + public Map getContext() { + return context; + } + }