I am cursed with this code. Fixed another routing bug. Added addition…

…al logging.
JumpMind · May 26, 2010 · 475f278 · 475f278
1 parent 563f1b5
commit 475f278
Show file tree

Hide file tree

Showing 5 changed files with 36 additions and 10 deletions.
diff --git a/symmetric/src/main/java/org/jumpmind/symmetric/route/DataToRouteReader.java b/symmetric/src/main/java/org/jumpmind/symmetric/route/DataToRouteReader.java
@@ -147,20 +147,35 @@ public Integer doInConnection(Connection c) throws SQLException, DataAccessExcep
                         ts = System.currentTimeMillis();
                     }
 
+                    ts = System.currentTimeMillis();
+
                     copyToQueue(memQueue);
 
+                    context.incrementStat(System.currentTimeMillis() - ts,
+                            RouterContext.STAT_ENQUEUE_DATA_MS);                   
+
                     return dataCount;
+
                 } finally {
+
                     JdbcUtils.closeResultSet(rs);
                     JdbcUtils.closeStatement(ps);
-                    reading = false;
+                    rs = null;
+                    ps = null;
+
+                    long ts = System.currentTimeMillis();
+
                     boolean done = false;
                     do {
                         done = dataQueue.offer(new EOD());
                         AppUtils.sleep(50);
                     } while (!done && reading);
-                    rs = null;
-                    ps = null;
+
+                    context.incrementStat(System.currentTimeMillis() - ts,
+                            RouterContext.STAT_ENQUEUE_EOD_MS);
+
+                    reading = false;
+
                 }
             }
         });

diff --git a/symmetric/src/main/java/org/jumpmind/symmetric/route/RouterContext.java b/symmetric/src/main/java/org/jumpmind/symmetric/route/RouterContext.java
@@ -44,6 +44,7 @@ public class RouterContext extends SimpleRouterContext implements IRouterContext
     public static final String STAT_DATA_ROUTER_MS = "data.router.ms";
     public static final String STAT_READ_DATA_MS = "read.data.ms";
     public static final String STAT_ENQUEUE_DATA_MS = "enqueue.data.ms";
+    public static final String STAT_ENQUEUE_EOD_MS = "enqueue.eod.data.ms";
 
     private Map<String, OutgoingBatch> batchesByNodes = new HashMap<String, OutgoingBatch>();
     private Map<TriggerRouter, Set<Node>> availableNodes = new HashMap<TriggerRouter, Set<Node>>();

diff --git a/symmetric/src/main/java/org/jumpmind/symmetric/route/SimpleRouterContext.java b/symmetric/src/main/java/org/jumpmind/symmetric/route/SimpleRouterContext.java
@@ -76,17 +76,24 @@ public long getStat(String name) {
         return val;
     }
 
-    public void logStats(ILog log, boolean infoLevel) {
+    public void logStats(ILog log, int dataCount, long totalTimeInMs) {
+        boolean infoLevel = totalTimeInMs > 30000;
         Set<String> keys = contextCache.keySet();
         for (String key : keys) {
             if (key.startsWith("Stat.")) {
+                String keyString = key.substring(key.indexOf(".") + 1);
                 if (infoLevel) {
-                    log.info("RouterStats", channel.getChannelId(), key.substring(key.indexOf(".") + 1), contextCache.get(key));                    
+                    log.info("RouterStats", channel.getChannelId(), keyString, contextCache.get(key));                    
                 } else {
-                    log.debug("RouterStats", channel.getChannelId(), key.substring(key.indexOf(".") + 1), contextCache.get(key));                    
-                }
-
+                    log.debug("RouterStats", channel.getChannelId(), keyString, contextCache.get(key));                    
+                }               
             }
         }
+
+        if (infoLevel) {
+            log.info("RouterTimeForChannel", totalTimeInMs, dataCount, channel.getChannelId());
+        } else {
+            log.debug("RouterTimeForChannel", totalTimeInMs,  dataCount, channel.getChannelId());
+        }
     }
 }
diff --git a/symmetric/src/main/java/org/jumpmind/symmetric/service/impl/RouterService.java b/symmetric/src/main/java/org/jumpmind/symmetric/service/impl/RouterService.java
@@ -159,9 +159,11 @@ protected int routeDataForChannel(final DataRef ref, final NodeChannel nodeChann
             final Node sourceNode) {
         RouterContext context = null;
         long ts = System.currentTimeMillis();
+        int dataCount = -1;
         try {
             context = new RouterContext(sourceNode.getNodeId(), nodeChannel, dataSource);
-            return selectDataAndRoute(ref, context);
+            dataCount = selectDataAndRoute(ref, context);
+            return dataCount;
         } catch (Exception ex) {
             if (context != null) {
                 context.rollback();
@@ -174,7 +176,7 @@ protected int routeDataForChannel(final DataRef ref, final NodeChannel nodeChann
             } catch (SQLException e) {
                 log.error(e);
             } finally {
-                context.logStats(log, System.currentTimeMillis() - ts > 30000);
+                context.logStats(log, dataCount, System.currentTimeMillis() - ts);
                 context.cleanup();
             }
         }

diff --git a/symmetric/src/main/resources/symmetric-messages.properties b/symmetric/src/main/resources/symmetric-messages.properties
@@ -183,6 +183,7 @@ RouterIllegalColumnMatchExpression=The provided column match expression was inva
 RouterIllegalLookupTableExpression=The provided table mapped router expression was invalid: %s.
 RouterNoColumnsToMatch=There were no columns to match for the data_id of %d
 RouterStats=Routing '%s' stat %s=%d
+RouterTimeForChannel=It took %d ms to route %d data for the '%s' channel
 RouterSkippingDataIdsGapExpired=Found a gap in data_id from %d to %d.  Skipping it because the gap expired.
 RouterSkippingDataIdsNoTransactions=Found a gap in data_id from %d to %d.  Skipping it because their are no pending transactions in the database.
 RouterRoutingFailed=Failed to route and batch data on '%s' channel