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/16 23:19:44 UTC
[1/2] git commit: single thread calls if they are not in a job context
Repository: incubator-usergrid
Updated Branches:
refs/heads/two-dot-o 32455d8fd -> c70920195
single thread calls if they are not in a job context
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/7cd7175b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/7cd7175b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/7cd7175b
Branch: refs/heads/two-dot-o
Commit: 7cd7175bfd909d6c832b2dae40513a61ecab1168
Parents: 3187097
Author: Shawn Feldman <sf...@apache.org>
Authored: Tue Sep 16 15:18:29 2014 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Tue Sep 16 15:18:29 2014 -0600
----------------------------------------------------------------------
.../notifications/ApplicationQueueManager.java | 142 ++++++++++---------
.../notifications/NotificationsService.java | 3 +
2 files changed, 76 insertions(+), 69 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7cd7175b/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 d25c9d9..1f103a7 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
@@ -136,85 +136,89 @@ public class ApplicationQueueManager implements QueueManager {
final UUID appId = em.getApplication().getUuid();
final Map<String,Object> payloads = notification.getPayloads();
- LOG.info("ApplicationQueueMessage: notification {} start threading", notification.getUuid());
- rx.Observable.create(new IteratorObservable<Entity>(iterator)).parallel(new Func1<Observable<Entity>, Observable<Entity>>() {
+ Observable o;
+ final Func1<Entity,Entity> entityListFunct = new Func1<Entity, Entity>() {
@Override
- public rx.Observable<Entity> call(rx.Observable<Entity> deviceObservable) {
- return deviceObservable.map(new Func1<Entity, Entity>() {
- @Override
- public Entity call(Entity entity) {
- try {
- LOG.info("ApplicationQueueMessage: notification {} send to entity {}", notification.getUuid(), entity.getUuid());
+ public Entity call(Entity entity) {
+ try {
- List<EntityRef> devicesRef = getDevices(entity); // resolve group
+ LOG.info("ApplicationQueueMessage: notification {} queue entity {}", notification.getUuid(), entity.getUuid());
- LOG.info("ApplicationQueueMessage: notification {} send to {} devices", notification.getUuid(), devicesRef.size());
+ List<EntityRef> devicesRef = getDevices(entity); // resolve group
- for (EntityRef deviceRef : devicesRef) {
- if(LOG.isDebugEnabled()){
- LOG.info("ApplicationQueueMessage: 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("ApplicationQueueMessage: Maybe Found duplicate device: {}", deviceRef.getUuid());
- continue;
- } else {
- sketch.add(hash, 1);
- }
- String notifierId = null;
- String notifierKey = null;
-
- //find the device notifier info, match it to the payload
- for (Map.Entry<String, Object> entry : payloads.entrySet()) {
- Notifier notifier = notifierMap.get(entry.getKey().toLowerCase());
- String providerId = getProviderId(deviceRef, notifier);
- if (providerId != null) {
- notifierId = providerId;
- notifierKey = entry.getKey().toLowerCase();
- break;
- }
- }
+ LOG.info("ApplicationQueueMessage: notification {} queue {} devices", notification.getUuid(), devicesRef.size());
- if (notifierId == null) {
- LOG.debug("ApplicationQueueMessage: Notifier did not match for device {} ", deviceRef);
- continue;
- }
-
- ApplicationQueueMessage message = new ApplicationQueueMessage(appId, notification.getUuid(), deviceRef.getUuid(), notifierKey, notifierId);
- if(LOG.isDebugEnabled()){
- LOG.info("ApplicationQueueMessage: notification {} pre-queue to device {} ", notification.getUuid(), deviceRef.getUuid());
- }
- qm.postToQueue(QUEUE_NAME, message);
- if(LOG.isDebugEnabled()){
- LOG.info("ApplicationQueueMessage: notification {} post-queue to device {} ", notification.getUuid(), deviceRef.getUuid());
- }
- if (notification.getQueued() == null) {
- // update queued time
- notification.setQueued(System.currentTimeMillis());
- em.update(notification);
- LOG.info("ApplicationQueueMessage: notification {} queue time set.", notification.getUuid(), deviceRef.getUuid());
- }
- deviceCount.incrementAndGet();
- queueMeter.mark();
+ for (EntityRef deviceRef : devicesRef) {
+ LOG.info("ApplicationQueueMessage: 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("ApplicationQueueMessage: Maybe Found duplicate device: {}", deviceRef.getUuid());
+ continue;
+ } else {
+ sketch.add(hash, 1);
+ }
+ String notifierId = null;
+ String notifierKey = null;
+
+ //find the device notifier info, match it to the payload
+ for (Map.Entry<String, Object> entry : payloads.entrySet()) {
+ Notifier notifier = notifierMap.get(entry.getKey().toLowerCase());
+ String providerId = getProviderId(deviceRef, notifier);
+ if (providerId != null) {
+ notifierId = providerId;
+ notifierKey = entry.getKey().toLowerCase();
+ break;
}
+ }
- } catch (Exception deviceLoopException) {
- LOG.error("Failed to add devices", deviceLoopException);
- errorMessages.add("Failed to add devices for entity: " + entity.getUuid() + " error:" + deviceLoopException);
+ if (notifierId == null) {
+ LOG.debug("ApplicationQueueMessage: Notifier did not match for device {} ", deviceRef);
+ continue;
}
- return entity;
+
+ ApplicationQueueMessage message = new ApplicationQueueMessage(appId, notification.getUuid(), deviceRef.getUuid(), notifierKey, notifierId);
+ LOG.info("ApplicationQueueMessage: notification {} pre-queue to device {} ", notification.getUuid(), deviceRef.getUuid());
+ qm.postToQueue(QUEUE_NAME, message);
+ LOG.info("ApplicationQueueMessage: notification {} post-queue to device {} ", notification.getUuid(), deviceRef.getUuid());
+ if (notification.getQueued() == null) {
+ // update queued time
+ notification.setQueued(System.currentTimeMillis());
+ em.update(notification);
+ LOG.info("ApplicationQueueMessage: notification {} queue time set.", notification.getUuid(), deviceRef.getUuid());
+ }
+ deviceCount.incrementAndGet();
+ queueMeter.mark();
}
- });
+ } catch (Exception deviceLoopException) {
+ LOG.error("Failed to add devices", deviceLoopException);
+ errorMessages.add("Failed to add devices for entity: " + entity.getUuid() + " error:" + deviceLoopException);
+ }
+ return entity;
}
- }, Schedulers.io())
- .doOnError(new Action1<Throwable>() {
- @Override
- public void call(Throwable throwable) {
- LOG.error("Failed while writing", throwable);
- }
- })
- .toBlocking()
- .lastOrDefault(null);
+ };
+ if(jobExecution != null) {
+ LOG.info("ApplicationQueueMessage: notification {} start threading", notification.getUuid());
+
+ o = rx.Observable.create(new IteratorObservable<Entity>(iterator)).parallel(new Func1<Observable<Entity>, Observable<Entity>>() {
+ @Override
+ public rx.Observable<Entity> call(rx.Observable<Entity> deviceObservable) {
+ return deviceObservable.map(entityListFunct);
+ }
+ }, Schedulers.io())
+ .doOnError(new Action1<Throwable>() {
+ @Override
+ public void call(Throwable throwable) {
+ LOG.error("Failed while writing", throwable);
+ }
+ });
+ }else{
+ LOG.info("ApplicationQueueMessage: notification {} start single threaded", notification.getUuid());
+ o = rx.Observable.create(new IteratorObservable(iterator))
+ .subscribeOn(Schedulers.io())
+ .map(entityListFunct);
+ }
+ o.toBlocking().lastOrDefault(null);
+
}
// update queued time
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7cd7175b/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationsService.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationsService.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationsService.java
index 17e3d15..b3e0188 100644
--- a/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationsService.java
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationsService.java
@@ -130,6 +130,7 @@ public class NotificationsService extends AbstractCollectionService {
@Override
public ServiceResults postCollection(ServiceContext context) throws Exception {
+ LOG.info("NotificationService: start request.");
Timer.Context timer = postTimer.time();
postMeter.mark();
try {
@@ -140,10 +141,12 @@ public class NotificationsService extends AbstractCollectionService {
context.setOwner(sm.getApplication());
ServiceResults results = super.postCollection(context);
Notification notification = (Notification) results.getEntity();
+ LOG.info("NotificationService: notification {} pre queue ", notification.getUuid());
if(!notificationQueueManager.scheduleQueueJob(notification)){
notificationQueueManager.queueNotification(notification, null);
}
outstandingQueue.inc();
+ LOG.info("NotificationService: notification {} post queue ", notification.getUuid());
// future: somehow return 202?
return results;
}finally {
[2/2] git commit: Merge branch 'two-dot-o' of
https://git-wip-us.apache.org/repos/asf/incubator-usergrid into two-dot-o
Posted by sf...@apache.org.
Merge branch 'two-dot-o' of https://git-wip-us.apache.org/repos/asf/incubator-usergrid into two-dot-o
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/c7092019
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/c7092019
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/c7092019
Branch: refs/heads/two-dot-o
Commit: c70920195bc19159d6103ed41d070e47b13b5a4a
Parents: 7cd7175 32455d8
Author: Shawn Feldman <sf...@apache.org>
Authored: Tue Sep 16 15:18:43 2014 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Tue Sep 16 15:18:43 2014 -0600
----------------------------------------------------------------------
.../organizations/users/UsersResource.java | 20 ++---
.../rest/management/users/UsersResource.java | 28 +++----
.../apache/usergrid/rest/PartialUpdateTest.java | 85 ++++++++++++++++++++
.../collection/activities/PutTest.java | 51 +-----------
4 files changed, 105 insertions(+), 79 deletions(-)
----------------------------------------------------------------------