From 2213059b3bbe691399d0751ef817a4fc6e6a7baa Mon Sep 17 00:00:00 2001
From: chenson42 <chenson42@gmail.com>
Date: Tue, 26 Feb 2013 13:55:56 +0000
Subject: [PATCH] 0001078: Allow sym_load_filter.target_table_name to be left
 blank so that filters are applied to all tables

---
 .../symmetric/load/BshDatabaseWriterFilter.java | 16 ++++++++++++++--
 .../service/impl/LoadFilterService.java         | 17 +++++++++++------
 .../src/main/resources/symmetric-schema.xml     |  2 +-
 3 files changed, 26 insertions(+), 9 deletions(-)

diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/load/BshDatabaseWriterFilter.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/load/BshDatabaseWriterFilter.java
index c3ef1df389..5dd24cb6ce 100644
--- a/symmetric-core/src/main/java/org/jumpmind/symmetric/load/BshDatabaseWriterFilter.java
+++ b/symmetric-core/src/main/java/org/jumpmind/symmetric/load/BshDatabaseWriterFilter.java
@@ -20,6 +20,7 @@
  */
 package org.jumpmind.symmetric.load;
 
+import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -38,6 +39,7 @@
 import org.jumpmind.symmetric.io.data.writer.IDatabaseWriterFilter;
 import org.jumpmind.symmetric.model.LoadFilter;
 import org.jumpmind.util.Context;
+import org.jumpmind.util.FormatUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -208,9 +210,19 @@ protected boolean processLoadFilters(DataContext context, Table table, CsvData d
         boolean writeRow = true;
         LoadFilter currentFilter = null;
 
-        List<LoadFilter> loadFiltersForTable = loadFilters.get(table.getFullyQualifiedTableName());
+        List<LoadFilter> wildcardLoadFilters =  loadFilters.get(Table.getFullyQualifiedTableName(table.getCatalog(), table.getSchema(), FormatUtils.WILDCARD));
+        List<LoadFilter> tableSpecificLoadFilters = loadFilters.get(table.getFullyQualifiedTableName());
+        int size = (wildcardLoadFilters != null ? wildcardLoadFilters.size() : 0) + (tableSpecificLoadFilters != null ? tableSpecificLoadFilters.size() : 0);
 
-        if (loadFiltersForTable != null && loadFiltersForTable.size() > 0) {
+        if (size > 0) {
+            List<LoadFilter> loadFiltersForTable = new ArrayList<LoadFilter>(size);
+            if (wildcardLoadFilters != null) {
+                loadFiltersForTable.addAll(wildcardLoadFilters);
+            }
+            
+            if (tableSpecificLoadFilters != null) {
+                loadFiltersForTable.addAll(tableSpecificLoadFilters);
+            }
             try {
                 Interpreter interpreter = getInterpreter(context);
                 bind(interpreter, context, table, data, error);
diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/LoadFilterService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/LoadFilterService.java
index 83a2a3f586..20abaebad4 100644
--- a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/LoadFilterService.java
+++ b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/LoadFilterService.java
@@ -7,6 +7,7 @@
 import java.util.List;
 import java.util.Map;
 
+import org.apache.commons.lang.StringUtils;
 import org.jumpmind.db.model.Table;
 import org.jumpmind.db.sql.ISqlRowMapper;
 import org.jumpmind.db.sql.Row;
@@ -17,6 +18,7 @@
 import org.jumpmind.symmetric.service.IConfigurationService;
 import org.jumpmind.symmetric.service.ILoadFilterService;
 import org.jumpmind.symmetric.service.IParameterService;
+import org.jumpmind.util.FormatUtils;
 
 public class LoadFilterService extends AbstractService implements ILoadFilterService {
 
@@ -79,17 +81,20 @@ protected void refreshCache() {
                         if (loadFiltersByNodeGroup == null) {
                             loadFiltersByNodeGroup = new HashMap<String, List<LoadFilter>>();
                         }
+                        String tableName = loadFilter.getTargetTableName();
+                        if (StringUtils.isBlank(tableName)) {
+                            tableName = FormatUtils.WILDCARD;
+                        }
+                        String qualifiedName = Table.getFullyQualifiedTableName(
+                                loadFilter.getTargetCatalogName(),
+                                loadFilter.getTargetSchemaName(), tableName);
                         List<LoadFilter> loadFiltersForTable = loadFiltersByNodeGroup
-                                .get(Table.getFullyQualifiedTableName(
-                                		loadFilter.getTargetCatalogName(),
-                                        loadFilter.getTargetSchemaName(), loadFilter.getTargetTableName()));
+                                .get(qualifiedName);
                         if (loadFiltersForTable == null) {
                             loadFiltersForTable = new ArrayList<LoadFilter>();
                         }
                         loadFiltersForTable.add(loadFilter);
-                        loadFiltersByNodeGroup.put(Table.getFullyQualifiedTableName(
-                                loadFilter.getTargetCatalogName(),
-                                loadFilter.getTargetSchemaName(), loadFilter.getTargetTableName()),
+                        loadFiltersByNodeGroup.put(qualifiedName,
                                 loadFiltersForTable);
                         loadFilterCacheByNodeGroupLink.put(nodeGroupLink, loadFiltersByNodeGroup);
                     }
diff --git a/symmetric-core/src/main/resources/symmetric-schema.xml b/symmetric-core/src/main/resources/symmetric-schema.xml
index db80195c5f..1a934b5e23 100644
--- a/symmetric-core/src/main/resources/symmetric-schema.xml
+++ b/symmetric-core/src/main/resources/symmetric-schema.xml
@@ -571,7 +571,7 @@
         <column name="target_node_group_id" type="VARCHAR" size="50" required="true" description="The destination node group for the filter." />
         <column name="target_catalog_name" type="VARCHAR" size="255"  description="Optional name for the catalog the configured table is in." />
         <column name="target_schema_name" type="VARCHAR" size="255"  description="Optional name for the schema a configured table is in." />
-        <column name="target_table_name" type="VARCHAR" size="255" required="true"  description="The name of the target table that will trigger the bsh filter." />
+        <column name="target_table_name" type="VARCHAR" size="255" description="The name of the target table that will trigger the bsh filter." />
         <column name="filter_on_update" type="BOOLEANINT" size="1" required="true" default="1"  description="Whether or not the filter should apply on an update." />
         <column name="filter_on_insert" type="BOOLEANINT" size="1" required="true" default="1"  description="Whether or not the filter should apply on an insert." />
         <column name="filter_on_delete" type="BOOLEANINT" size="1" required="true" default="1"  description="Whether or not the filter should apply on a delete." />