Return the number of resource events processed

diff --git a/src/main/scala/gr/grnet/aquarium/actor/service/user/UserActor.scala b/src/main/scala/gr/grnet/aquarium/actor/service/user/UserActor.scala
index 12006a1..0503b97 100644 (file)
--- a/src/main/scala/gr/grnet/aquarium/actor/service/user/UserActor.scala
+++ b/src/main/scala/gr/grnet/aquarium/actor/service/user/UserActor.scala
@@ -190,7 +190,6 @@ class UserActor extends ReflectiveRoleableActor {
 
   private[this] def processResourceEventsAfterLastKnownUserState() {
     // Update the user state snapshot with fresh (ie not previously processed) events.
-
   }
 
   private[this] def makeUserStateMsgUpToDate() {

diff --git a/src/main/scala/gr/grnet/aquarium/store/ResourceEventStore.scala b/src/main/scala/gr/grnet/aquarium/store/ResourceEventStore.scala
index 4650285..a5d657b 100644 (file)
--- a/src/main/scala/gr/grnet/aquarium/store/ResourceEventStore.scala
+++ b/src/main/scala/gr/grnet/aquarium/store/ResourceEventStore.scala
@@ -57,9 +57,17 @@ trait ResourceEventStore {
    */
   def countOutOfSyncResourceEventsForBillingPeriod(userID: String, startMillis: Long, stopMillis: Long): Long
 
+  /**
+   *
+   * @param userID
+   * @param startMillis
+   * @param stopMillis
+   * @param f
+   * @return The number of resource events processed.
+   */
   def foreachResourceEventOccurredInPeriod(
       userID: String,
       startMillis: Long,
       stopMillis: Long
-  )(f: ResourceEventMsg ⇒ Unit): Unit
+  )(f: ResourceEventMsg ⇒ Unit): Long
 }
\ No newline at end of file

diff --git a/src/main/scala/gr/grnet/aquarium/store/memory/MemStoreProvider.scala b/src/main/scala/gr/grnet/aquarium/store/memory/MemStoreProvider.scala
index 9f0a66a..0746916 100644 (file)
--- a/src/main/scala/gr/grnet/aquarium/store/memory/MemStoreProvider.scala
+++ b/src/main/scala/gr/grnet/aquarium/store/memory/MemStoreProvider.scala
@@ -152,11 +152,17 @@ extends StoreProvider
       userID: String,
       startMillis: Long,
       stopMillis: Long
-  )(f: ResourceEventMsg ⇒ Unit): Unit = {
+  )(f: ResourceEventMsg ⇒ Unit): Long = {
+    var _counter= 0L
     _resourceEvents.filter { case ev ⇒
       ev.getUserID == userID &&
       MessageHelpers.isOccurredWithinMillis(ev, startMillis, stopMillis)
-    }.foreach(f)
+    } foreach { rcEvent ⇒
+      f(rcEvent)
+      _counter += 1
+    }
+
+    _counter
   }
 
   //+ IMEventStore

diff --git a/src/main/scala/gr/grnet/aquarium/store/mongodb/MongoDBStore.scala b/src/main/scala/gr/grnet/aquarium/store/mongodb/MongoDBStore.scala
index 1907ec7..b2d549e 100644 (file)
--- a/src/main/scala/gr/grnet/aquarium/store/mongodb/MongoDBStore.scala
+++ b/src/main/scala/gr/grnet/aquarium/store/mongodb/MongoDBStore.scala
@@ -139,8 +139,8 @@ class MongoDBStore(
       userID: String,
       startMillis: Long,
       stopMillis: Long
-  )(f: ResourceEventMsg ⇒ Unit): Unit = {
-
+  )(f: ResourceEventMsg ⇒ Unit): Long = {
+    var _counter= 0L
     val query = new BasicDBObjectBuilder().
       add(MongoDBStore.JsonNames.userID, userID).
       add(MongoDBStore.JsonNames.occurredMillis, new BasicDBObject("$gte", startMillis)).
@@ -157,8 +157,11 @@ class MongoDBStore(
         val nextEvent = AvroHelpers.specificRecordOfBytes(payload, new ResourceEventMsg)
 
         f(nextEvent)
+        _counter += 1
       }
     }
+
+    _counter
   }
   //-ResourceEventStore