add support to StatisticsTracker to keep a CrawledBytesHistotable per source tag when trackSources is enabled; integration test for this functionality

engine/src/main/java/org/archive/crawler/reporting/StatisticsTracker.java

@@ -288,6 +288,12 @@ public class StatisticsTracker
     // TODO: restore spill-to-disk, like with processedSeedsRecords
     protected ConcurrentHashMap<String, ConcurrentMap<String, AtomicLong>> sourceHostDistribution = 
         new ConcurrentHashMap<String, ConcurrentMap<String,AtomicLong>>(); 
+    
+    /** Keep track of crawled bytes stats per seed */
+    // TODO: spill-to-disk (requires bdb replacement for Histotable, or some
+    // other refactoring)
+    protected ConcurrentHashMap<String, CrawledBytesHistotable> statsBySource =
+            new ConcurrentHashMap<String, CrawledBytesHistotable>();
 
     /* Keep track of 'top' hosts for live reports */
     protected TopNSet hostsDistributionTop;
@@ -396,6 +402,15 @@ public class StatisticsTracker
                     sourceHostDistribution.put(source, hostUriCount);
                 }
                 
+                JSONObject ss = json.getJSONObject("sourceStats");
+                keyIter = ss.keys();
+                for(; keyIter.hasNext();) {
+                    String source = keyIter.next();
+                    CrawledBytesHistotable cb = new CrawledBytesHistotable();
+                    JSONUtils.putAllLongs(cb, ss.getJSONObject(source));
+                    statsBySource.put(source, cb);
+                }
+                
                 JSONUtils.putAllLongs(
                     crawledBytes,
                     json.getJSONObject("crawledBytes"));
@@ -768,9 +783,9 @@ public class StatisticsTracker
                 curi.getContentSize());
         
         if (getTrackSources() && curi.getData().containsKey(A_SOURCE_TAG)) {
-        	saveSourceStats((String)curi.getData().get(A_SOURCE_TAG),
-                        sc.getHostFor(curi.getUURI()).
-                    getHostName()); 
+        	saveSourceStats(curi.getSourceTag(), 
+        	        sc.getHostFor(curi.getUURI()).getHostName());
+        	tallySourceStats(curi);
         }
     }
          
@@ -784,7 +799,16 @@ public class StatisticsTracker
             }
         }
         incrementMapCount(hostUriCount, hostname);
+    }
 
+    protected void tallySourceStats(CrawlURI curi) {
+        String source = curi.getSourceTag();
+        CrawledBytesHistotable sourceStats = statsBySource.get(source);
+        if (sourceStats == null) {
+            sourceStats = new CrawledBytesHistotable();
+            statsBySource.put(source, sourceStats);
+        }
+        sourceStats.accumulate(curi);
     }
     
     /**
@@ -1087,6 +1111,7 @@ public class StatisticsTracker
             json.put("statusCodeDistribution", statusCodeDistribution);
 
             json.put("sourceHostDistribution", sourceHostDistribution);
+            json.put("sourceStats", statsBySource);
             
             json.put("crawledBytes", crawledBytes);
 
@@ -1103,4 +1128,10 @@ public class StatisticsTracker
         this.recoveryCheckpoint = recoveryCheckpoint;
     }
     
+    public CrawledBytesHistotable getSourceStats(String source) {
+        return statsBySource.get(source);
+    }
+    
+    
+    
 }