0003234: Ability to store additional statistics by channel

JumpMind · Aug 30, 2017 · eb02543 · eb02543
1 parent 7c4d6bf
commit eb02543
Show file tree

Hide file tree

Showing 4 changed files with 34 additions and 0 deletions.
diff --git a/symmetric-client/src/main/java/org/jumpmind/symmetric/ClientSymmetricEngine.java b/symmetric-client/src/main/java/org/jumpmind/symmetric/ClientSymmetricEngine.java
@@ -24,6 +24,7 @@
 
 import java.io.File;
 import java.io.StringReader;
+import java.lang.reflect.Constructor;
 import java.sql.Connection;
 import java.sql.SQLException;
 import java.util.ArrayList;
@@ -57,10 +58,17 @@
 import org.jumpmind.symmetric.io.stage.IStagingManager;
 import org.jumpmind.symmetric.job.IJobManager;
 import org.jumpmind.symmetric.job.JobManager;
+import org.jumpmind.symmetric.service.IClusterService;
+import org.jumpmind.symmetric.service.IConfigurationService;
 import org.jumpmind.symmetric.service.IExtensionService;
 import org.jumpmind.symmetric.service.IMonitorService;
+import org.jumpmind.symmetric.service.INodeService;
+import org.jumpmind.symmetric.service.IParameterService;
+import org.jumpmind.symmetric.service.IStatisticService;
 import org.jumpmind.symmetric.service.impl.ClientExtensionService;
 import org.jumpmind.symmetric.service.impl.MonitorService;
+import org.jumpmind.symmetric.statistic.IStatisticManager;
+import org.jumpmind.symmetric.statistic.StatisticManager;
 import org.jumpmind.symmetric.util.LogSummaryAppenderUtils;
 import org.jumpmind.symmetric.util.SnapshotUtil;
 import org.jumpmind.symmetric.util.TypedPropertiesFactory;
@@ -351,6 +359,24 @@ protected IStagingManager createStagingManager() {
         String directory = parameterService.getTempDirectory();
         return new BatchStagingManager(this, directory);
     }
+
+    @Override
+    protected IStatisticManager createStatisticManager() {
+        String statisticManagerClassName = parameterService.getString(ParameterConstants.STATISTIC_MANAGER_CLASS);
+        if (statisticManagerClassName != null) {
+            try {
+                Constructor<?> cons = Class.forName(statisticManagerClassName).getConstructor(IParameterService.class, 
+                        INodeService.class, IConfigurationService.class, IStatisticService.class, IClusterService.class);
+                return (IStatisticManager) cons.newInstance(parameterService, nodeService,
+                        configurationService, statisticService, clusterService);
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
+        }
+
+        return new StatisticManager(parameterService, nodeService,
+                configurationService, statisticService, clusterService);
+    }
 
     protected static void waitForAvailableDatabase(DataSource dataSource) {
         boolean success = false;

diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/AbstractSymmetricEngine.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/AbstractSymmetricEngine.java
@@ -328,6 +328,7 @@ protected void init() {
         this.statisticService = new StatisticService(parameterService, symmetricDialect);
         this.statisticManager = new StatisticManager(parameterService, nodeService,
                 configurationService, statisticService, clusterService);
+        this.statisticManager = createStatisticManager();
         this.concurrentConnectionManager = new ConcurrentConnectionManager(parameterService,
                 statisticManager);
         this.purgeService = new PurgeService(parameterService, symmetricDialect, clusterService,
@@ -398,6 +399,8 @@ protected INodeCommunicationService buildNodeCommunicationService(IClusterServic
 
     abstract protected IStagingManager createStagingManager();
 
+    abstract protected IStatisticManager createStatisticManager();
+
     abstract protected ISymmetricDialect createSymmetricDialect();
 
     abstract protected IExtensionService createExtensionService();

diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/common/ParameterConstants.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/common/ParameterConstants.java
@@ -393,6 +393,8 @@ private ParameterConstants() {
 
     public final static String LOG_CONFLICT_RESOLUTION = "log.conflict.resolution";
 
+    public final static String STATISTIC_MANAGER_CLASS = "statistic.manager.class";
+
     public static Map<String, ParameterMetaData> getParameterMetaData() {
         return parameterMetaData;
     }

diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorService.java
@@ -1073,6 +1073,9 @@ protected OutgoingBatch sendOutgoingBatch(ProcessInfo processInfo, Node targetNo
                                 SymmetricUtils.copyFile(extractedBatch.getFile(), targetResource.getFile());
                                 targetResource.setState(State.DONE);
                                 isRetry = true;
+
+                                	statisticManager.incrementDataSent(currentBatch.getChannelId(), currentBatch.getDataEventCount());
+                            		statisticManager.incrementDataBytesSent(currentBatch.getChannelId(), extractedBatch.getFile().length());
                             } catch (Exception e) {   
                                 FileUtils.deleteQuietly(targetResource.getFile());
                                 throw new RuntimeException(e);