change queue impl
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/d536b5b4 Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/d536b5b4 Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/d536b5b4 Branch: refs/heads/master Commit: d536b5b4e3a377948e3427c95bedd5b178428850 Parents: 505b44a Author: Shawn Feldman <[email protected]> Authored: Wed Oct 14 15:38:25 2015 -0600 Committer: Shawn Feldman <[email protected]> Committed: Wed Oct 14 15:38:25 2015 -0600 ---------------------------------------------------------------------- .../usergrid/persistence/queue/DefaultQueueManager.java | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/usergrid/blob/d536b5b4/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java ---------------------------------------------------------------------- diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java index edd3b6b..d9a141e 100644 --- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java +++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java @@ -23,21 +23,24 @@ package org.apache.usergrid.persistence.queue; import rx.Observable; import java.io.IOException; +import java.util.AbstractQueue; import java.util.ArrayList; import java.util.List; import java.util.UUID; import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.PriorityBlockingQueue; /** * Default queue manager implementation, uses in memory linked queue */ public class DefaultQueueManager implements QueueManager { - public ArrayBlockingQueue<QueueMessage> queue = new ArrayBlockingQueue<>(10000000); + public LinkedBlockingQueue<QueueMessage> queue = new LinkedBlockingQueue<>(); @Override public synchronized Observable<QueueMessage> getMessages(int limit, int transactionTimeout, int waitTime, Class klass) { List<QueueMessage> returnQueue = new ArrayList<>(); - queue.drainTo(returnQueue,1); + queue.drainTo(returnQueue,10); return Observable.from( returnQueue); }
