usergrid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mru...@apache.org
Subject [2/7] usergrid git commit: First pass to allow more than 1k notifications to be sent for groups/queries targeting users and devices.
Date Mon, 11 Apr 2016 22:26:35 GMT
First pass to allow more than 1k notifications to be sent for groups/queries targeting users
and devices.


Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/8e4d7eef
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/8e4d7eef
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/8e4d7eef

Branch: refs/heads/release-2.1.1
Commit: 8e4d7eef2ca0967491e9ef863c78880e002575d1
Parents: 74de4bc
Author: Michael Russo <mrusso@apigee.com>
Authored: Sun Apr 10 23:20:10 2016 +0300
Committer: Michael Russo <mrusso@apigee.com>
Committed: Sun Apr 10 23:20:10 2016 +0300

----------------------------------------------------------------------
 .../corepersistence/CpRelationManager.java      |  34 ++++---
 .../services/notifications/TaskManager.java     |   7 +-
 .../impl/ApplicationQueueManagerImpl.java       | 100 ++++++++++++++++---
 3 files changed, 113 insertions(+), 28 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/usergrid/blob/8e4d7eef/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
index 67e92f8..b5a4107 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
@@ -17,14 +17,7 @@
 package org.apache.usergrid.corepersistence;
 
 
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.UUID;
+import java.util.*;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -318,14 +311,31 @@ public class CpRelationManager implements RelationManager {
 
         final String ql;
 
-        if ( startResult != null ) {
-            ql = "select * where created > " + startResult.timestamp();
-        }
-        else {
+
+        if (startResult != null ) {
+
+            // UUID timestamp is a different measure than 'created' field on entities
+            Calendar uuidEpoch = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
+            uuidEpoch.clear();
+            uuidEpoch.set(1582, 9, 15, 0, 0, 0); // 9 = October
+            long epochMillis = uuidEpoch.getTime().getTime();
+
+            long time = (startResult.timestamp() / 10000L) + epochMillis;
+
+            if ( !reversed ) {
+                ql = "select * where created > " + time;
+            } else {
+                ql = "select * where created < " + time;
+            }
+
+        } else {
             ql = "select *";
         }
 
         Query query = Query.fromQL( ql );
+        if(query == null ){
+            throw new RuntimeException("Unable to get data for collection: "+collectionName);
+        }
         query.setLimit( count );
         query.setReversed( reversed );
 

http://git-wip-us.apache.org/repos/asf/usergrid/blob/8e4d7eef/stack/services/src/main/java/org/apache/usergrid/services/notifications/TaskManager.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/TaskManager.java
b/stack/services/src/main/java/org/apache/usergrid/services/notifications/TaskManager.java
index 3e78210..954724f 100644
--- a/stack/services/src/main/java/org/apache/usergrid/services/notifications/TaskManager.java
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/TaskManager.java
@@ -131,13 +131,14 @@ public class TaskManager {
         }
 
         if ( debug || hasError) {
+
+            List<EntityRef> entities = Arrays.asList(notification, device);
+
             if (receipt.getUuid() == null) {
                 Receipt savedReceipt = em.create(receipt);
-                receipt.setUuid(savedReceipt.getUuid());
-                List<EntityRef> entities = Arrays.asList(notification, device);
                 em.addToCollections(entities, Notification.RECEIPTS_COLLECTION, savedReceipt);
             } else {
-                em.update(receipt);
+                em.addToCollections(entities, Notification.RECEIPTS_COLLECTION, receipt);
             }
         }
 

http://git-wip-us.apache.org/repos/asf/usergrid/blob/8e4d7eef/stack/services/src/main/java/org/apache/usergrid/services/notifications/impl/ApplicationQueueManagerImpl.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/impl/ApplicationQueueManagerImpl.java
b/stack/services/src/main/java/org/apache/usergrid/services/notifications/impl/ApplicationQueueManagerImpl.java
index 12a47b6..04e60b7 100644
--- a/stack/services/src/main/java/org/apache/usergrid/services/notifications/impl/ApplicationQueueManagerImpl.java
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/impl/ApplicationQueueManagerImpl.java
@@ -34,6 +34,7 @@ import rx.Observable;
 import rx.Subscriber;
 import rx.functions.Func1;
 
+import java.io.IOException;
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -119,7 +120,7 @@ public class ApplicationQueueManagerImpl implements ApplicationQueueManager
{
             final UUID appId = em.getApplication().getUuid();
             final Map<String, Object> payloads = notification.getPayloads();
 
-            final Func1<EntityRef, EntityRef> sendMessageFunction = deviceRef ->
{
+            final Func1<EntityRef, ApplicationQueueMessage> sendMessageFunction = deviceRef
-> {
                 try {
 
                     long now = System.currentTimeMillis();
@@ -143,7 +144,8 @@ public class ApplicationQueueManagerImpl implements ApplicationQueueManager
{
                     }
 
                     if (notifierId == null) {
-                        return deviceRef;
+                        //TODO need to leverage optional here
+                        //return deviceRef;
                     }
 
                     ApplicationQueueMessage message = new ApplicationQueueMessage(appId,
notification.getUuid(), deviceRef.getUuid(), notifierKey, notifierId);
@@ -153,16 +155,19 @@ public class ApplicationQueueManagerImpl implements ApplicationQueueManager
{
                         notification.setQueued(System.currentTimeMillis());
 
                     }
-                    qm.sendMessage(message);
                     deviceCount.incrementAndGet();
-                    queueMeter.mark();
+
+                    return message;
 
 
                 } catch (Exception deviceLoopException) {
                     logger.error("Failed to add device", deviceLoopException);
                     errorMessages.add("Failed to add device: " + deviceRef.getUuid() + ",
error:" + deviceLoopException);
+
+                    //TODO need an optional here
+                    return new ApplicationQueueMessage(appId, notification.getUuid(), deviceRef.getUuid(),
"test", "test");
                 }
-                return deviceRef;
+
             };
 
 
@@ -174,9 +179,28 @@ public class ApplicationQueueManagerImpl implements ApplicationQueueManager
{
                 }, 10)
                 .distinct(ref -> ref.getUuid())
                 .map(sendMessageFunction)
+                .buffer(100)
+                .doOnNext( applicationQueueMessages -> {
+
+                    applicationQueueMessages.forEach( message -> {
+
+                        try {
+
+                            qm.sendMessage( message );
+                            queueMeter.mark();
+
+                        } catch (IOException e) {
+                           logger.error("Unable to queue notification for notification UUID
{} and device UUID {} ",
+                               message.getNotificationId(), message.getDeviceId());
+                        }
+
+                    });
+
+
+                })
                 .doOnError(throwable -> logger.error("Failed while trying to send notification",
throwable));
 
-            processMessagesObservable.toBlocking().lastOrDefault(null);
+            processMessagesObservable.toBlocking(); // let this run and block the async thread,
messages are queued
 
         }
 
@@ -487,20 +511,70 @@ public class ApplicationQueueManagerImpl implements ApplicationQueueManager
{
 
     private List<EntityRef> getDevices(EntityRef ref) {
 
-        List<EntityRef> devices = Collections.EMPTY_LIST;
+        List<EntityRef> devices = new ArrayList<>();
+
 
         try {
+
             if ("device".equals(ref.getType())) {
+
                 devices = Collections.singletonList(ref);
+
             } else if ("user".equals(ref.getType())) {
-                devices = em.getCollection(ref, "devices", null, Query.MAX_LIMIT,
-                    Query.Level.REFS, false).getRefs();
+
+                UUID start = null;
+                boolean initial = true;
+                int resultSize = 0;
+                while( initial || resultSize >= Query.DEFAULT_LIMIT) {
+
+                    initial = false;
+
+                    final List<EntityRef> mydevices = em.getCollection(ref, "devices",
start, Query.DEFAULT_LIMIT,
+                        Query.Level.REFS, true).getRefs();
+
+                    resultSize = mydevices.size();
+                    if(mydevices.size() > 0){
+                        start = mydevices.get(mydevices.size() - 1 ).getUuid();
+                    }
+
+
+                    devices.addAll( mydevices  );
+
+
+                }
+
             } else if ("group".equals(ref.getType())) {
-                devices = new ArrayList<>();
-                for (EntityRef r : em.getCollection(ref, "users", null,
-                    Query.MAX_LIMIT, Query.Level.REFS, false).getRefs()) {
-                    devices.addAll(getDevices(r));
+
+                //devices = new ArrayList<>();
+                UUID start = null;
+                boolean initial = true;
+                int resultSize = 0;
+
+                while( initial || resultSize >= Query.DEFAULT_LIMIT){
+
+                        initial = false;
+                        final List<EntityRef> myusers =  em.getCollection(ref, "users",
start,
+                            Query.DEFAULT_LIMIT, Query.Level.REFS, true).getRefs();
+
+                        resultSize = myusers.size();
+                        if(myusers.size() > 0){
+                            start = myusers.get(myusers.size() - 1 ).getUuid();
+                        }
+
+
+                        // don't allow a single user to have more than 100 devices?
+                        for (EntityRef user : myusers) {
+
+                            devices.addAll( em.getCollection(user, "devices", null, 100,
+                                Query.Level.REFS, true).getRefs() );
+
+
+                        }
+
                 }
+
+
+
             }
         } catch (Exception e) {
 


Mime
View raw message