RateLimitedPushManager.java

package com.swrve.push_system;

import com.google.common.util.concurrent.RateLimiter;
import com.relayrides.pushy.apns.ApnsEnvironment;
import com.relayrides.pushy.apns.ApnsPushNotification;
import com.relayrides.pushy.apns.PushManager;
import java.security.KeyStore;
import java.util.Collection;
import scala.actors.threadpool.Arrays;

public class RateLimitedPushManager<T extends ApnsPushNotification> extends PushManager<T> {
    
    private final RateLimiter rateLimiter;
    
    public RateLimitedPushManager(RateLimiter rateLimiter, ApnsEnvironment environment, KeyStore keyStore, char[] keyStorePassword) {
        super(environment, keyStore, keyStorePassword);
        this.rateLimiter = rateLimiter;
    }

    public RateLimitedPushManager(RateLimiter rateLimiter, ApnsEnvironment environment, KeyStore keyStore, char[] keyStorePassword, int concurrentConnections) {
        super(environment, keyStore, keyStorePassword, concurrentConnections);
        this.rateLimiter = rateLimiter;
    }
    
    @Override
    public void enqueuePushNotification(T notification) {
        rateLimiter.acquire();
        super.enqueuePushNotification(notification);
    }
    
    @Override
    public void enqueueAllNotifications(Collection<T> notifications) {
        rateLimiter.acquire(notifications.size());
        super.enqueueAllNotifications(notifications);
    }
    
    public void enqueueAllNotificationsInBatches(Collection<T> notifications, int batchSize) {
        Object[] array = notifications.toArray();
        for (int i=0; i<notifications.size(); i += batchSize) {
            rateLimiter.acquire(batchSize);
            super.enqueueAllNotifications(Arrays.asList(Arrays.copyOfRange(array, i, i+batchSize)));
        }
    }
}