You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sf...@apache.org on 2014/09/15 22:23:10 UTC
git commit: adding debug code
Repository: incubator-usergrid
Updated Branches:
refs/heads/two-dot-o 983ba6996 -> 1f8f35951
adding debug code
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/1f8f3595
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/1f8f3595
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/1f8f3595
Branch: refs/heads/two-dot-o
Commit: 1f8f3595197b57257cedbf36ca7a354aefd70922
Parents: 983ba69
Author: Shawn Feldman <sf...@apache.org>
Authored: Mon Sep 15 14:22:52 2014 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Sep 15 14:22:52 2014 -0600
----------------------------------------------------------------------
.../notifications/ApplicationQueueManager.java | 46 ++++++++++++++------
.../services/notifications/JobScheduler.java | 4 +-
.../services/notifications/QueueListener.java | 12 ++++-
3 files changed, 47 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/1f8f3595/stack/services/src/main/java/org/apache/usergrid/services/notifications/ApplicationQueueManager.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/ApplicationQueueManager.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/ApplicationQueueManager.java
index ed37d51..70dbdfc 100644
--- a/stack/services/src/main/java/org/apache/usergrid/services/notifications/ApplicationQueueManager.java
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/ApplicationQueueManager.java
@@ -107,7 +107,7 @@ public class ApplicationQueueManager implements QueueManager {
public void queueNotification(final Notification notification, final JobExecution jobExecution) throws Exception {
if (notification.getCanceled() == Boolean.TRUE) {
- LOG.info("notification " + notification.getUuid() + " canceled");
+ LOG.info("ApplicationsQueueMessage: notification " + notification.getUuid() + " canceled");
if (jobExecution != null) {
jobExecution.killed();
}
@@ -115,7 +115,7 @@ public class ApplicationQueueManager implements QueueManager {
}
long startTime = System.currentTimeMillis();
- LOG.info("notification {} start queuing", notification.getUuid());
+ LOG.info("ApplicationsQueueMessage: notification {} start queuing", notification.getUuid());
final PathQuery<Device> pathQuery = notification.getPathQuery() ; //devices query
final AtomicInteger deviceCount = new AtomicInteger(); //count devices so you can make a judgement on batching
final ConcurrentLinkedQueue<String> errorMessages = new ConcurrentLinkedQueue<String>(); //build up list of issues
@@ -124,6 +124,7 @@ public class ApplicationQueueManager implements QueueManager {
//get devices in querystring, and make sure you have access
if (pathQuery != null) {
+ LOG.info("ApplicationsQueueMessage: notification {} start query", notification.getUuid());
final Iterator<Device> iterator = pathQuery.iterator(em);
//if there are more pages (defined by PAGE_SIZE) you probably want this to be async, also if this is already a job then don't reschedule
if (iterator instanceof ResultsIterator && ((ResultsIterator) iterator).hasPages() && jobExecution == null) {
@@ -131,7 +132,9 @@ public class ApplicationQueueManager implements QueueManager {
return;
}
final CountMinSketch sketch = new CountMinSketch(0.0001,.99,7364181); //add probablistic counter to find dups
+ final UUID appId = em.getApplication().getUuid();
+ LOG.info("ApplicationsQueueMessage: notification {} start threading", notification.getUuid());
rx.Observable.create(new IteratorObservable<Entity>(iterator)).parallel(new Func1<Observable<Entity>, Observable<Entity>>() {
@Override
public rx.Observable<Entity> call(rx.Observable<Entity> deviceObservable) {
@@ -139,12 +142,19 @@ public class ApplicationQueueManager implements QueueManager {
@Override
public Entity call(Entity entity) {
try {
+ LOG.info("ApplicationsQueueMessage: notification {} send to entity {}", notification.getUuid(), entity.getUuid());
+
List<EntityRef> devicesRef = getDevices(entity); // resolve group
+ LOG.info("ApplicationsQueueMessage: notification {} send to {} devices", notification.getUuid(), devicesRef.size());
+
for (EntityRef deviceRef : devicesRef) {
+ if(LOG.isDebugEnabled()){
+ LOG.info("ApplicationsQueueMessage: notification {} starting to queue device {} ", notification.getUuid(), deviceRef.getUuid());
+ }
long hash = MurmurHash.hash(deviceRef.getUuid());
if (sketch.estimateCount(hash) > 0) { //look for duplicates
- LOG.debug("Maybe Found duplicate device: {}", deviceRef.getUuid());
+ LOG.debug("ApplicationsQueueMessage: Maybe Found duplicate device: {}", deviceRef.getUuid());
continue;
} else {
sketch.add(hash, 1);
@@ -164,16 +174,26 @@ public class ApplicationQueueManager implements QueueManager {
}
if (notifierId == null) {
- LOG.debug("Notifier did not match for device {} ", deviceRef);
+ LOG.debug("ApplicationsQueueMessage: Notifier did not match for device {} ", deviceRef);
continue;
}
- ApplicationQueueMessage message = new ApplicationQueueMessage(em.getApplication().getUuid(), notification.getUuid(), deviceRef.getUuid(), notifierKey, notifierId);
+ ApplicationQueueMessage message = new ApplicationQueueMessage(appId, notification.getUuid(), deviceRef.getUuid(), notifierKey, notifierId);
+ if(LOG.isDebugEnabled()){
+ LOG.info("ApplicationsQueueMessage: notification {} pre-queue to device {} ", notification.getUuid(), deviceRef.getUuid());
+ }
qm.postToQueue(QUEUE_NAME, message);
+ if(LOG.isDebugEnabled()){
+ LOG.info("ApplicationsQueueMessage: notification {} post-queue to device {} ", notification.getUuid(), deviceRef.getUuid());
+ }
if (notification.getQueued() == null) {
- // update queued time
- notification.setQueued(System.currentTimeMillis());
- em.update(notification);
+ synchronized (notification) {
+ // update queued time
+ notification.setQueued(System.currentTimeMillis());
+ em.update(notification);
+ LOG.info("ApplicationsQueueMessage: notification {} queue time set.", notification.getUuid(), deviceRef.getUuid());
+ }
+
}
deviceCount.incrementAndGet();
}
@@ -222,11 +242,9 @@ public class ApplicationQueueManager implements QueueManager {
if (LOG.isInfoEnabled()) {
long elapsed = notification.getQueued() != null ? notification.getQueued() - startTime : 0;
- StringBuilder sb = new StringBuilder();
- sb.append("notification ").append(notification.getUuid());
- sb.append(" done queuing to ").append(deviceCount);
- sb.append(" devices in ").append(elapsed).append(" ms");
- LOG.info(sb.toString());
+ LOG.info("ApplicationsQueueMessage: notification {} done queuing to {} devices in "+elapsed+" ms",notification.getUuid().toString(),deviceCount.get());
+ LOG.info("ApplicationsQueueMessage: notification {} finished in {} ms",notification.getUuid().toString(),elapsed);
+
}
}
@@ -261,6 +279,8 @@ public class ApplicationQueueManager implements QueueManager {
* @throws Exception
*/
public Observable sendBatchToProviders( final List<ApplicationQueueMessage> messages) {
+ LOG.info("sending batch of {} notifications.", messages.size());
+
final Map<Object, Notifier> notifierMap = getNotifierMap();
final QueueManager proxy = this;
final ConcurrentHashMap<UUID,SingleQueueTaskManager> taskMap = new ConcurrentHashMap<UUID, SingleQueueTaskManager>(messages.size());
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/1f8f3595/stack/services/src/main/java/org/apache/usergrid/services/notifications/JobScheduler.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/JobScheduler.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/JobScheduler.java
index 54b4434..fa293ed 100644
--- a/stack/services/src/main/java/org/apache/usergrid/services/notifications/JobScheduler.java
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/JobScheduler.java
@@ -82,7 +82,9 @@ public class JobScheduler{
SchedulerService scheduler = getSchedulerService();
scheduler.createJob("queueJob", scheduleAt, jobData);
}
- LOG.info("notification {} scheduled for queuing", notification.getUuid());
+ if(scheduled){
+ LOG.info("notification {} scheduled for queuing", notification.getUuid());
+ }
return scheduled;
}
private SchedulerService getSchedulerService() {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/1f8f3595/stack/services/src/main/java/org/apache/usergrid/services/notifications/QueueListener.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/QueueListener.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/QueueListener.java
index 29e62c2..01c02e3 100644
--- a/stack/services/src/main/java/org/apache/usergrid/services/notifications/QueueListener.java
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/QueueListener.java
@@ -87,12 +87,15 @@ public class QueueListener {
}
public void run(){
+ LOG.info("QueueListener: starting.");
+
int threadCount = 0;
try {
sleepPeriod = new Long(properties.getProperty("usergrid.notifications.listener.sleep", "5000")).longValue();
int maxThreads = new Integer(properties.getProperty("usergrid.notifications.listener.maxThreads", MAX_THREADS));
futures = new ArrayList<Future>(maxThreads);
while (threadCount++ < maxThreads) {
+ LOG.info("QueueListener: Starting thread {}.", threadCount);
futures.add(
pool.submit(new Runnable() {
@Override
@@ -107,18 +110,25 @@ public class QueueListener {
);
}
}catch (Exception e){
- LOG.error("QueueListener failed to start:", e);
+ LOG.error("QueueListener: failed to start:", e);
}
+ LOG.info("QueueListener: done starting.");
+
}
private void execute(){
+
svcMgr = smf.getServiceManager(smf.getManagementAppId());
queueManager = svcMgr.getQueueManager();
final AtomicInteger consecutiveExceptions = new AtomicInteger();
+ LOG.info("QueueListener: Starting execute process.");
+
// run until there are no more active jobs
while ( true ) {
try {
QueueResults results = ApplicationQueueManager.getDeliveryBatch(queueManager);
+ LOG.info("QueueListener: retreived batch of {} messages",results.size());
+
List<Message> messages = results.getMessages();
if(messages.size()>0) {
Observable.from(messages) //observe all messages