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/18 04:41:07 UTC
[2/2] git commit: add threads back
add threads back
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/023572cd
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/023572cd
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/023572cd
Branch: refs/heads/two-dot-o
Commit: 023572cd85552b35d89135ee6f196d977baf2296
Parents: ac39bf2
Author: Shawn Feldman <sf...@apache.org>
Authored: Wed Sep 17 20:40:41 2014 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Wed Sep 17 20:40:41 2014 -0600
----------------------------------------------------------------------
.../notifications/ApplicationQueueManager.java | 30 ++++++++++----------
1 file changed, 15 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/023572cd/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 d13cf9f..47d5ffd 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 = 100;
+ public static int BATCH_SIZE = 1000;
public static final long MESSAGE_TRANSACTION_TIMEOUT = 5 * 60 * 1000;
private static final Logger LOG = LoggerFactory.getLogger(ApplicationQueueManager.class);
@@ -197,7 +197,7 @@ public class ApplicationQueueManager implements QueueManager {
return entity;
}
};
- if(jobExecution != 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>>() {
@@ -212,12 +212,12 @@ public class ApplicationQueueManager implements QueueManager {
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);
- }
+// }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);
}
@@ -363,14 +363,14 @@ public class ApplicationQueueManager implements QueueManager {
};
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).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)
+// rx.Observable.from(messages).subscribeOn(Schedulers.io()).map(func)
.buffer(BATCH_SIZE)
.map(new Func1<List<ApplicationQueueMessage>, HashMap<UUID, ApplicationQueueMessage>>() {
@Override