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/20 00:15:17 UTC
[1/2] git commit: smaller batches; property driven
Repository: incubator-usergrid
Updated Branches:
refs/heads/two-dot-o 3f694b8b5 -> bc6fbd561
smaller batches; property driven
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/810bc19a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/810bc19a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/810bc19a
Branch: refs/heads/two-dot-o
Commit: 810bc19a8db9822561a8eaa2188ee1bf8b6a425b
Parents: 3f694b8
Author: Shawn Feldman <sf...@apache.org>
Authored: Fri Sep 19 12:16:50 2014 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Fri Sep 19 12:16:50 2014 -0600
----------------------------------------------------------------------
.../notifications/ApplicationQueueManager.java | 9 +++++----
.../services/notifications/NotificationsService.java | 4 ++--
.../services/notifications/QueueListener.java | 15 ++++++++++-----
3 files changed, 17 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/810bc19a/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 c1c6960..71d6a04 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
@@ -52,8 +52,8 @@ import java.util.concurrent.atomic.AtomicInteger;
* Created by ApigeeCorporation on 8/27/14.
*/
public class ApplicationQueueManager implements QueueManager {
- public static String QUEUE_NAME = "notifications/queuelistenerv1_10";
- public static int BATCH_SIZE = 1000;
+ public static String QUEUE_NAME = "notifications/queuelistenerv1_11";
+ public static int BATCH_SIZE = 100;
public static final long MESSAGE_TRANSACTION_TIMEOUT = 5 * 1000;
private static final Logger LOG = LoggerFactory.getLogger(ApplicationQueueManager.class);
@@ -67,9 +67,9 @@ public class ApplicationQueueManager implements QueueManager {
private final org.apache.usergrid.mq.QueueManager qm;
private final JobScheduler jobScheduler;
private final MetricsFactory metricsFactory;
+ private final Properties properties;
HashMap<Object, Notifier> notifierHashMap; // only retrieve notifiers once
-
public final Map<String, ProviderAdapter> providerAdapters = new HashMap<String, ProviderAdapter>(3);
{
providerAdapters.put("apple", APNS_ADAPTER);
@@ -82,11 +82,12 @@ public class ApplicationQueueManager implements QueueManager {
public static ProviderAdapter TEST_ADAPTER = new TestAdapter();
- public ApplicationQueueManager(JobScheduler jobScheduler, EntityManager entityManager, org.apache.usergrid.mq.QueueManager queueManager, MetricsFactory metricsFactory){
+ public ApplicationQueueManager(JobScheduler jobScheduler, EntityManager entityManager, org.apache.usergrid.mq.QueueManager queueManager, MetricsFactory metricsFactory,Properties properties){
this.em = entityManager;
this.qm = queueManager;
this.jobScheduler = jobScheduler;
this.metricsFactory = metricsFactory;
+ this.properties = properties;
}
public static QueueResults getDeliveryBatch(org.apache.usergrid.mq.QueueManager queueManager) throws Exception {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/810bc19a/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 65baf8d..f45750c 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
@@ -94,12 +94,12 @@ public class NotificationsService extends AbstractCollectionService {
super.init(info);
smf = getApplicationContext().getBean(ServiceManagerFactory.class);
emf = getApplicationContext().getBean(EntityManagerFactory.class);
-
+ Properties props = (Properties)getApplicationContext().getBean("properties");
metricsService = getApplicationContext().getBean(MetricsFactory.class);
postMeter = metricsService.getMeter(NotificationsService.class, "requests");
postTimer = metricsService.getTimer(this.getClass(), "execution_rest");
JobScheduler jobScheduler = new JobScheduler(sm,em);
- notificationQueueManager = new ApplicationQueueManager(jobScheduler,em,smf.getServiceManager(smf.getManagementAppId()).getQueueManager(),metricsService);
+ notificationQueueManager = new ApplicationQueueManager(jobScheduler,em,smf.getServiceManager(smf.getManagementAppId()).getQueueManager(),metricsService,props);
gracePeriod = jobScheduler.SCHEDULER_GRACE_PERIOD;
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/810bc19a/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 0ca7b0f..5cb1bfe 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
@@ -63,7 +63,9 @@ public class QueueListener {
private ServiceManager svcMgr;
- private long sleepPeriod = 5000;
+ private long sleepWhenNoneFound = 5000;
+
+ private long sleepBetweenRuns = 5000;
ExecutorService pool;
List<Future> futures;
@@ -92,8 +94,10 @@ public class QueueListener {
if(shouldRun) {
LOG.info("QueueListener: starting.");
int threadCount = 0;
+
try {
- sleepPeriod = new Long(properties.getProperty("usergrid.notifications.listener.sleep", "5000")).longValue();
+ sleepBetweenRuns = new Long(properties.getProperty("usergrid.notifications.listener.sleep.between", "5000")).longValue();
+ sleepWhenNoneFound = new Long(properties.getProperty("usergrid.notifications.listener.sleep.after", "5000")).longValue();
int maxThreads = new Integer(properties.getProperty("usergrid.notifications.listener.maxThreads", MAX_THREADS));
futures = new ArrayList<Future>(maxThreads);
while (threadCount++ < maxThreads) {
@@ -160,7 +164,8 @@ public class QueueListener {
new JobScheduler(serviceManager, entityManager),
entityManager,
queueManager,
- metricsService
+ metricsService,
+ properties
);
return groupedObservable //buffer all of your notifications into a sender and send.
@@ -183,11 +188,11 @@ public class QueueListener {
.toBlocking()
.last();
LOG.info("QueueListener: Messages sent in batch");
-
+ Thread.sleep(sleepBetweenRuns);
}
else{
LOG.info("QueueListener: no messages...sleep...",results.size());
- Thread.sleep(sleepPeriod);
+ Thread.sleep(sleepWhenNoneFound);
}
//send to the providers
consecutiveExceptions.set(0);
[2/2] git commit: add optional sleep value
Posted by sf...@apache.org.
add optional sleep value
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/bc6fbd56
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/bc6fbd56
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/bc6fbd56
Branch: refs/heads/two-dot-o
Commit: bc6fbd561ea36a7a6b54d8e50cf443b781d5eef4
Parents: 810bc19
Author: Shawn Feldman <sf...@apache.org>
Authored: Fri Sep 19 16:14:58 2014 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Fri Sep 19 16:14:58 2014 -0600
----------------------------------------------------------------------
.../usergrid/services/notifications/QueueListener.java | 10 ++++++----
1 file changed, 6 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/bc6fbd56/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 5cb1bfe..ae54008 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
@@ -63,7 +63,7 @@ public class QueueListener {
private ServiceManager svcMgr;
- private long sleepWhenNoneFound = 5000;
+ private long sleepWhenNoneFound = 0;
private long sleepBetweenRuns = 5000;
@@ -96,7 +96,7 @@ public class QueueListener {
int threadCount = 0;
try {
- sleepBetweenRuns = new Long(properties.getProperty("usergrid.notifications.listener.sleep.between", "5000")).longValue();
+ sleepBetweenRuns = new Long(properties.getProperty("usergrid.notifications.listener.sleep.between", "0")).longValue();
sleepWhenNoneFound = new Long(properties.getProperty("usergrid.notifications.listener.sleep.after", "5000")).longValue();
int maxThreads = new Integer(properties.getProperty("usergrid.notifications.listener.maxThreads", MAX_THREADS));
futures = new ArrayList<Future>(maxThreads);
@@ -188,10 +188,12 @@ public class QueueListener {
.toBlocking()
.last();
LOG.info("QueueListener: Messages sent in batch");
- Thread.sleep(sleepBetweenRuns);
+ if(sleepBetweenRuns > 0) {
+ Thread.sleep(sleepBetweenRuns);
+ }
}
else{
- LOG.info("QueueListener: no messages...sleep...",results.size());
+ LOG.info("QueueListener: no messages...sleep...", results.size());
Thread.sleep(sleepWhenNoneFound);
}
//send to the providers