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/17 18:18:05 UTC
git commit: moving back to single threaded; changing buffer size.
Repository: incubator-usergrid
Updated Branches:
refs/heads/two-dot-o bd74c860a -> 0cc97c6c8
moving back to single threaded; changing buffer size.
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/0cc97c6c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/0cc97c6c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/0cc97c6c
Branch: refs/heads/two-dot-o
Commit: 0cc97c6c8652006c4bcb6401b2ec4af8c92f7bc3
Parents: bd74c86
Author: Shawn Feldman <sf...@apache.org>
Authored: Wed Sep 17 10:07:54 2014 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Wed Sep 17 10:07:54 2014 -0600
----------------------------------------------------------------------
.../notifications/ApplicationQueueManager.java | 134 ++++++++++---------
.../services/notifications/QueueListener.java | 2 +-
2 files changed, 69 insertions(+), 67 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/0cc97c6c/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 3b0d5bd..d13cf9f 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
@@ -53,7 +53,7 @@ import java.util.concurrent.atomic.AtomicInteger;
*/
public class ApplicationQueueManager implements QueueManager {
public static String QUEUE_NAME = "notifications/queuelistenerv1";
- public static int BATCH_SIZE = 1000;
+ public static int BATCH_SIZE = 100;
public static final long MESSAGE_TRANSACTION_TIMEOUT = 5 * 60 * 1000;
private static final Logger LOG = LoggerFactory.getLogger(ApplicationQueueManager.class);
@@ -303,73 +303,76 @@ public class ApplicationQueueManager implements QueueManager {
final ConcurrentHashMap<UUID,SingleQueueTaskManager> taskMap = new ConcurrentHashMap<UUID, SingleQueueTaskManager>(messages.size());
final ConcurrentHashMap<UUID,Notification> notificationMap = new ConcurrentHashMap<UUID, Notification>(messages.size());
- return rx.Observable
- .from(messages)
- .parallel(new Func1<rx.Observable<ApplicationQueueMessage>, rx.Observable<ApplicationQueueMessage>>() {
- @Override
- public rx.Observable<ApplicationQueueMessage> call(rx.Observable<ApplicationQueueMessage> messageObservable) {
- return messageObservable.map(new Func1<ApplicationQueueMessage, ApplicationQueueMessage>() {
- @Override
- public ApplicationQueueMessage call(ApplicationQueueMessage message) {
- try {
+ final Func1<ApplicationQueueMessage, ApplicationQueueMessage> func = new Func1<ApplicationQueueMessage, ApplicationQueueMessage>() {
+ @Override
+ public ApplicationQueueMessage call(ApplicationQueueMessage message) {
+ try {
- UUID deviceUUID = message.getDeviceId();
-
- Notification notification = notificationMap.get(message.getNotificationId());
- if (notification == null) {
- notification = em.get(message.getNotificationId(), Notification.class);
- notificationMap.put(message.getNotificationId(), notification);
- }
- SingleQueueTaskManager taskManager;
- synchronized (taskMap) {
- taskManager = taskMap.get(message.getNotificationId());
- if (taskManager == null) {
- taskManager = new SingleQueueTaskManager(em, qm, proxy, notification);
- taskMap.put(message.getNotificationId(), taskManager);
- }
- }
-
- final Map<String, Object> payloads = notification.getPayloads();
- final Map<String, Object> translatedPayloads = translatePayloads(payloads, notifierMap);
- LOG.info("sending notification for device {} for Notification: {}", deviceUUID, notification.getUuid());
- if(!isOkToSend(notification)){
- return message;
- }
- taskManager.addMessage(deviceUUID,message);
- try {
- String notifierName = message.getNotifierKey().toLowerCase();
- Notifier notifier = notifierMap.get(notifierName.toLowerCase());
- Object payload = translatedPayloads.get(notifierName);
- Receipt receipt = new Receipt(notification.getUuid(), message.getNotifierId(), payload, deviceUUID);
- TaskTracker tracker = new TaskTracker(notifier, taskManager, receipt, deviceUUID);
-
- if (payload == null) {
- LOG.debug("selected device {} for notification {} doesn't have a valid payload. skipping.", deviceUUID, notification.getUuid());
- tracker.failed(0, "failed to match payload to " + message.getNotifierId() + " notifier");
-
- }else{
- try {
- ProviderAdapter providerAdapter = providerAdapters.get(notifier.getProvider());
- providerAdapter.sendNotification(message.getNotifierId(), notifier, payload, notification, tracker);
- } catch (Exception e) {
- tracker.failed(0, e.getMessage());
- }
- }
-
- } finally {
- sendMeter.mark();
- }
+ UUID deviceUUID = message.getDeviceId();
- } catch (Exception e) {
- LOG.error("Failure unknown",e);
- }
- return message;
+ Notification notification = notificationMap.get(message.getNotificationId());
+ if (notification == null) {
+ notification = em.get(message.getNotificationId(), Notification.class);
+ notificationMap.put(message.getNotificationId(), notification);
+ }
+ SingleQueueTaskManager taskManager;
+ synchronized (taskMap) {
+ taskManager = taskMap.get(message.getNotificationId());
+ if (taskManager == null) {
+ taskManager = new SingleQueueTaskManager(em, qm, proxy, notification);
+ taskMap.put(message.getNotificationId(), taskManager);
+ }
+ }
+
+ final Map<String, Object> payloads = notification.getPayloads();
+ final Map<String, Object> translatedPayloads = translatePayloads(payloads, notifierMap);
+ LOG.info("sending notification for device {} for Notification: {}", deviceUUID, notification.getUuid());
+ if(!isOkToSend(notification)){
+ return message;
+ }
+ taskManager.addMessage(deviceUUID,message);
+ try {
+ String notifierName = message.getNotifierKey().toLowerCase();
+ Notifier notifier = notifierMap.get(notifierName.toLowerCase());
+ Object payload = translatedPayloads.get(notifierName);
+ Receipt receipt = new Receipt(notification.getUuid(), message.getNotifierId(), payload, deviceUUID);
+ TaskTracker tracker = new TaskTracker(notifier, taskManager, receipt, deviceUUID);
+
+ if (payload == null) {
+ LOG.debug("selected device {} for notification {} doesn't have a valid payload. skipping.", deviceUUID, notification.getUuid());
+ tracker.failed(0, "failed to match payload to " + message.getNotifierId() + " notifier");
+
+ }else{
+ try {
+ ProviderAdapter providerAdapter = providerAdapters.get(notifier.getProvider());
+ providerAdapter.sendNotification(message.getNotifierId(), notifier, payload, notification, tracker);
+ } catch (Exception e) {
+ tracker.failed(0, e.getMessage());
}
- });
+ }
+
+ } finally {
+ sendMeter.mark();
}
- }, Schedulers.io())
- .buffer(BATCH_SIZE)
- .map(new Func1<List<ApplicationQueueMessage>, HashMap<UUID, ApplicationQueueMessage>>() {
+
+ } catch (Exception e) {
+ LOG.error("Failure unknown",e);
+ }
+ return message;
+ }
+ };
+ Observable o =
+
+// rx.Observable.from(messages).parallel(new Func1<rx.Observable<ApplicationQueueMessage>, rx.Observable<ApplicationQueueMessage>>() {
+// @Override
+// public rx.Observable<ApplicationQueueMessage> call(rx.Observable<ApplicationQueueMessage> messageObservable) {
+// return messageObservable.map(func);
+// }
+// }, Schedulers.io());
+
+ rx.Observable.from(messages).subscribeOn(Schedulers.io()).map(func)
+ .buffer(BATCH_SIZE)
+ .map(new Func1<List<ApplicationQueueMessage>, HashMap<UUID, ApplicationQueueMessage>>() {
@Override
public HashMap<UUID, ApplicationQueueMessage> call(List<ApplicationQueueMessage> queueMessages) {
//for gcm this will actually send notification
@@ -403,8 +406,7 @@ public class ApplicationQueueManager implements QueueManager {
LOG.error("Failed while sending",throwable);
}
});
-
-
+ return o;
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/0cc97c6c/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 bbaebcf..f9db3e4 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
@@ -68,7 +68,7 @@ public class QueueListener {
ExecutorService pool;
List<Future> futures;
- public static final String MAX_THREADS = "1";
+ public static final String MAX_THREADS = "4";
public QueueListener() {
pool = Executors.newFixedThreadPool(1);