Mongo: Handling of duplicate events

Mongo: Added action filter in getTargetStats
Added Serializable interface to model objects
Various fixes

src/main/java/org/mongoste/core/DuplicateEventException.java

@@ -0,0 +1,40 @@
+/*
+ *    Copyright (c) 2010-2011 Manuel Polo (mrmx.org)
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.mongoste.core;
+
+/**
+ * Duplicate event stats engine exception
+ * @author mrmx
+ */
+public class DuplicateEventException extends StatsEngineException {
+
+	/**
+	 * Serial version
+	 */
+	private static final long serialVersionUID = -8318719826740661005L;
+
+	public DuplicateEventException() {
+    }
+
+    public DuplicateEventException(String msg) {
+        super(msg);
+    }
+
+    public DuplicateEventException(String msg, Throwable cause) {
+        super(msg,cause);
+    }
+
+}

src/main/java/org/mongoste/core/StatsEngineException.java

@@ -21,7 +21,12 @@
  */
 public class StatsEngineException extends RuntimeException {
 
-    public StatsEngineException() {
+    /**
+	 * Serial version
+	 */
+	private static final long serialVersionUID = 3863955288683082759L;
+
+	public StatsEngineException() {
     }
 
     public StatsEngineException(String msg) {

src/main/java/org/mongoste/core/impl/mongodb/MongoStatsEngine.java

@@ -16,6 +16,7 @@
 package org.mongoste.core.impl.mongodb;
 
 import org.mongoste.core.AbstractStatsEngine;
+import org.mongoste.core.DuplicateEventException;
 import org.mongoste.core.StatsEngineException;
 import org.mongoste.core.TimeScope;
 import org.mongoste.model.StatEvent;
@@ -30,6 +31,7 @@
 import com.mongodb.DBObject;
 import com.mongodb.Mongo;
 import com.mongodb.MongoException;
+import com.mongodb.WriteConcern;
 import com.mongodb.WriteResult;
 
 import org.apache.commons.io.IOUtils;
@@ -67,7 +69,9 @@ public class MongoStatsEngine extends AbstractStatsEngine {
     private Map<String,DBCollection> collectionMap;
     private Map<String,String> functionMap;
     private static final DBObject EMPTY_DOC = new BasicDBObject();
-
+    private static final int ERROR_DUPKEY 			= 11000;
+    private static final int ERROR_DUPKEY_INSERT	= 11001;
+
     public static final String DB_NAME              = "dbname";
     protected static final String DEFAULT_DB_NAME     = "mongoste";
     protected static final String EVENT_CLIENT_ID     = "_idc";
@@ -104,6 +108,8 @@ public class MongoStatsEngine extends AbstractStatsEngine {
     protected static final String METAKEY_IP             = "ip";
 
     protected static final TimeScope DEFAULT_TIMESCOPE_PRECISION = TimeScope.DAILY;
+
+
 
     private boolean resetCollections = false; //For testing debug!!!!
     private boolean countEvents = true;
@@ -172,7 +178,7 @@ public void handleEvent(StatEvent event) throws StatsEngineException {
                 countTarget(event);
                 //Count total actions/targets
                 countTargetActions(event);
-            } else throw new StatsEngineException("Raw count failed with event "+ event);
+            } 
         }
     }
 
@@ -310,6 +316,10 @@ public List<StatAction> getTargetStats(Query query) throws StatsEngineException
             EVENT_TARGET_TYPE,getQueryValue(query,QueryField.TARGET_TYPE),
             EVENT_TARGET,     getQueryValue(query,QueryField.TARGET)
         );
+        QueryFilter actionFilter = query.getFilter(QueryField.ACTION);
+        if(actionFilter != null) {
+        	queryDoc.put(EVENT_ACTION,getQueryValue(query,QueryField.ACTION));
+        }
         QueryFilter dateFromFilter = query.getFilter(QueryField.DATE_FROM);
         DateTime dtFrom = null;
         if(dateFromFilter != null) {            
@@ -338,7 +348,9 @@ EVENT_TARGET,     getQueryValue(query,QueryField.TARGET)
                         new BasicDBObject("$gte", dtFrom.toDate()).append("$lte",dtTo.toDate())
                 );
             }
-        }
+        }        
+        //TODO getTargetStats(queryDoc,query.getPrecision() == TimeScope.DAILY) to handle day level
+        //TODO or better: getTargetStats(queryDoc,query.getPrecision()) to handle hourly,daily,monthly (default) precision
         return getTargetStats(queryDoc);
     }
 
@@ -348,7 +360,7 @@ private List<StatAction> getTargetStats(DBObject query) throws StatsEngineExcept
         try {
             log.debug("Querying targets");
             DBCollection targets = getTargetCollection();
-            long t = System.currentTimeMillis();
+            long t = System.currentTimeMillis();            
             DBObject fields = MongoUtil.createDoc(
                     EVENT_ACTION,1,
                     FIELD_COUNT,1,
@@ -568,7 +580,9 @@ private void saveEvent(StatEvent event) throws StatsEngineException {
         log.debug("saveEvent result: {}",ws.getLastError());
     }
 
-    private boolean countRawTarget(StatEvent event)  {
+    @SuppressWarnings("finally")
+	private boolean countRawTarget(StatEvent event)  throws StatsEngineException {
+    	boolean processed = false;    	
         try {
             BasicDBObject q = new BasicDBObject();
             q.put(EVENT_CLIENT_ID,event.getClientId());
@@ -605,13 +619,17 @@ private boolean countRawTarget(StatEvent event)  {
             }
             doc.put("$inc",incDoc);
             DBCollection targets = getTargetCollection(event,TimeScope.GLOBAL);
-            WriteResult ws = targets.update(q,doc,true,true);
-            //log.debug("countRawTarget result: {}",ws.getLastError());
-            return ws.getN() > 0;
-        }catch(Exception ex) {
-            log.error("countRawTarget failed",ex);
-        }
-        return false;
+            //TODO externalize write concern to configuration properties:
+            WriteResult wr = targets.update(q,doc,true,true,WriteConcern.FSYNC_SAFE); 
+            processed = wr.getN() > 0;
+        }catch(MongoException ex) {
+        	int errorCode = ex.getCode();
+        	if(errorCode == ERROR_DUPKEY || errorCode == ERROR_DUPKEY_INSERT) {
+        		throw new DuplicateEventException("Duplicate event " + event);
+        	}    	
+        	throw new StatsEngineException("countRawTarget failed",ex);
+        } 
+        return processed;        
     }
 
     private void countTarget(StatEvent event) throws StatsEngineException {
@@ -738,7 +756,7 @@ protected DBCollection getTargetCollection(StatEvent event, TimeScope timeScope)
                         EVENT_TARGET_TYPE,1,
                         TARGET_YEAR,1,
                         TARGET_MONTH,1
-                        ),"targetYearMonth",true);
+                        ),"targetYearMonth",false);
                 target.ensureIndex(MongoUtil.createDoc(
                         EVENT_CLIENT_ID,1,
                         EVENT_TARGET,1,
@@ -747,12 +765,19 @@ protected DBCollection getTargetCollection(StatEvent event, TimeScope timeScope)
                         TARGET_YEAR,1,
                         TARGET_MONTH,1
                         ),"targetActionYearMonth",true);
+                target.ensureIndex(MongoUtil.createDoc(
+                        EVENT_CLIENT_ID,1,
+                        EVENT_TARGET,1,
+                        EVENT_TARGET_TYPE,1,
+                        EVENT_ACTION,1,
+                        EVENT_DATE,1
+                        ),"targetActionDate",true);
                 target.ensureIndex(MongoUtil.createDoc(
                         EVENT_CLIENT_ID,1,
                         EVENT_TARGET,1,
                         EVENT_TARGET_TYPE,1,
                         EVENT_DATE,1
-                        ),"targetDate",true);
+                        ),"targetDate",false);
             }catch(MongoException ex) {
                 throw new StatsEngineException("creating target " + name + " indexes", ex);
             }

src/main/java/org/mongoste/model/StatAction.java

@@ -24,7 +24,11 @@
  * @author mrmx
  */
 public class StatAction extends StatCounter {
-    private List<StatCounter> targets;
+    /**
+	 * Serial version
+	 */
+	private static final long serialVersionUID = 3382251251144735716L;
+	private List<StatCounter> targets;
 
     public StatAction(String name, long count) {
         this(name,count,null);

src/main/java/org/mongoste/model/StatCounter.java

@@ -18,13 +18,19 @@
 import org.joda.time.DateTime;
 
 import java.util.Date;
+import java.io.Serializable;
 
 /**
  * A stat counter
  * @author mrmx
  */
-public class StatCounter {
-    private String name;
+public class StatCounter implements Serializable {
+
+    /**
+	 * Serial version
+	 */
+	private static final long serialVersionUID = -3881894514971648160L;
+	private String name;
     private long count;
     private DateTime dateTime;
 

src/main/java/org/mongoste/model/StatEvent.java

@@ -17,20 +17,26 @@
 
 import org.mongoste.util.DateUtil;
 
-import org.joda.time.DateTime;
-import org.joda.time.MutableDateTime;
-
+import java.io.Serializable;
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
 
+import org.joda.time.DateTime;
+import org.joda.time.MutableDateTime;
+
+
 /**
  * Main stat event object
  * @author mrmx
  */
-public class StatEvent {
-    private String clientId;
+public class StatEvent implements Serializable {
+    /**
+	 * Serial version
+	 */
+	private static final long serialVersionUID = 4528657092049273953L;
+	private String clientId;
     private String target;
     private String targetType;
     private List<String> targetOwners;