@@ -0,0 +1,1362 @@
+package gr.iti.mklab.manager.streams.monitors;%0A%0Aimport org.apache.log4j.Logger;%0A%0A/**%0A * @author manosetro%0A * @email manosetro@iti.gr%0A * This class implements a simple rate limits policy%0A */%0Apublic class RateLimitsMonitor %7B%0A%09%0A%09private int requests = 0;%0A%09private long lastCheckpoint = -1;%0A%09private int maxRequests;%0A%09private long minInterval;%0A%09%0A%09private Logger logger = Logger.getLogger(RateLimitsMonitor.class);%0A%09%0A%09public RateLimitsMonitor(int maxRequests, long minInterval) %7B%0A%09%09this.maxRequests = maxRequests;%0A%09%09this.minInterval = minInterval;%0A%09%7D%0A%09%0A%09/* %0A%09 * In a more sophisticated design this operation will be part of the %0A%09 * monitor module.%0A%09 */%0A%09synchronized public void check() %7B%0A%09%09if(lastCheckpoint %3C 0) %7B%0A%09%09%09lastCheckpoint = System.currentTimeMillis();%0A%09%09%7D%0A%09%09%0A%09%09requests++;%0A%09%09long interval = System.currentTimeMillis() - lastCheckpoint;%0A%09%09if(requests %3E= maxRequests) %7B%0A%09%09%09try %7B%0A%09%09%09%09long sleepTime = minInterval - interval;%0A%09%09%09%09if(sleepTime %3E 0) %7B%0A%09%09%09%09%09logger.info(%22Request limit reached. Wait for %22 + sleepTime/1000 + %22 seconds%22);%0A%09%09%09%09%09Thread.sleep(sleepTime);%0A%09%09%09%09%7D%0A%09%09%09%7D catch (InterruptedException e) %7B%0A%09%09%09%09logger.error(e.getMessage());%0A%09%09%09%09return;%0A%09%09%09%7D%0A%09%09%09finally %7B%0A%09%09%09%09requests = 0;%0A%09%09%09%09lastCheckpoint = System.currentTimeMillis();%0A%09%09%09%7D%0A%09%09%7D%0A%09%09else if(interval %3E minInterval) %7B%0A%09%09%09requests = 0;%0A%09%09%09lastCheckpoint = System.currentTimeMillis();%0A%09%09%7D%0A%09%7D%0A%7D%0A

Add rate limits monitor