public enum PlatformType {
    @EnumValue("GCM")
    SERVICE_GCM("GCM", "https://fcm.googleapis.com/fcm/send"),

    @EnumValue( "APNS")
    SERVICE_APNS("APNS", "https://gateway.sandbox.push.apple.com");

    public String name;
    public String url;
    PlatformType(String name, String url) {
        this.name = name;
        this.url = url;
    }
} public class MessageValidationException extends Throwable {

    public MessageValidationException(String message) {
        super(message);
    }
} public class MessageBuilder {
    public static class Builder {
        private final int ONE_WEEK_IN_SECONDS = 60 * 60 * 24 * 7;
        private final String DEFAULT_COLLAPSE_KEY = "default_collapse";
        private Credentials mCredentials;
        private Set<String> mMessageTokens = new HashSet<>();
        private Map<String, String> mMessageData = new HashMap<>();
        private MessagePriority mMessagePriority = MessagePriority.PRIORITY_LOW;
        private int mTtl = ONE_WEEK_IN_SECONDS;
        private boolean mShouldDelayWhileIdle = true;
        private boolean mIsDryRun = false;
        private String mCollapseKey;
        private int mMaxMessageRetries = 3;

        /**
         * Build the platform raw push service platform message.
         *
         * @return Push Service {@link Message} for a push platform.
         */
        @Nullable
        public Message build() throws MessageValidationException {
            List<Recipient> recipients = new ArrayList<>();
            for (String token : mMessageTokens) {
                Recipient recipient = new Recipient(token);
                recipients.add(recipient);
            }

            List<PayloadElement> payload = new ArrayList<>();
            for (Map.Entry<String, String> datum : mMessageData.entrySet()) {
                payload.add(new PayloadElement(datum.getKey(), datum.getValue()));
            }

            Message message = new Message();
            message.setRecipients(recipients);
            message.setCredentials(mCredentials);
            message.setCollapseKey(mCollapseKey != null ? mCollapseKey : DEFAULT_COLLAPSE_KEY);
            message.setTtlSeconds(mTtl);
            message.setDryRun(mIsDryRun);
            message.setPayloadData(payload);
            message.setMaximumRetries(mMaxMessageRetries);
            message.setShouldDelayWhileIdle(mShouldDelayWhileIdle);
            message.setMessagePriority(mMessagePriority);

            MessageHelper.verifyMessage(message);
            return message;
        }

        /**
         * Set a list of platform device token for the message to be sent to.
         *
         * @param tokens device tokens to send message to.
         */
        public Builder setDeviceTokens(@Nonnull Set<String> tokens) {
            mMessageTokens.addAll(tokens);
            return this;
        }

        /**
         * Add platform device token for the message to be sent to.
         *
         * @param token device token tokens to send message to.
         */
        public Builder addDeviceToken(@Nonnull String... token) {
            mMessageTokens.addAll(Arrays.asList(token));
            return this;
        }


        /**
         * Setting this to true allows a response back from the server for a sent message,
         * but it not to end up on the devices.
         *
         * @param isDryRun true if the message should not actually be sent. defaults to false.
         */
        public Builder setIsDryRun(boolean isDryRun) {
            mIsDryRun = isDryRun;
            return this;
        }

        /**
         * Add a set of key/value payload attributes to the message. This can be anything you want,
         * but is limited to 4096 bytes for all payload.
         *
         * @param attributeKey   key of message attribute.
         * @param attributeValue value of message attribute.
         */
        public Builder addData(String attributeKey, String attributeValue) {
            if (attributeKey != null && attributeValue != null) {
                mMessageData.put(attributeKey, attributeValue);
            }
            return this;
        }

        /**
         * Add a set of key/value payload attributes to the message. This can be anything you want,
         * but is limited to 4096 bytes for all payload.
         *
         * @param messageDataAttributes attribute keys and values for message payload.
         */
        public Builder setData(@Nonnull Map<String, String> messageDataAttributes) {
            mMessageData = messageDataAttributes;
            return this;
        }

        /**
         * Set the platform account (push services) that contains endpoints, platform type,
         * and authorisation keys, etc for that provider.
         *
         * @param credentials push service platform credentials.
         */
        public Builder setPlatformCredentials(@Nonnull Credentials credentials) {
            Credentials credentialsCopy = new Credentials(credentials.getPlatformType());
            credentialsCopy.setAuthKey(credentials.getAuthKey());
            credentialsCopy.setCertBody(credentials.getCertBody());
            credentialsCopy.setPackageUri(credentials.getPackageUri());
            mCredentials = credentialsCopy;
            return this;
        }

        /**
         * Set the time-to-live for a push message (can differs between platforms, but is usually
         * the time an unread or unsent message is kept on the server before giving up sending.
         *
         * @param ttl time to live in seconds. Default is one week.
         */
        public Builder setTimeToLiveSeconds(int ttl) {
            mTtl = ttl;
            return this;
        }

        /**
         * Sets the number of attempts the Message Dispatcher will try sending a message
         * to a recipient that hasn't fatally failed (for example, when we are told to back-off
         * from the provider). Default is 3.
         */
        public Builder setSoftFailRetries(int maxRetries) {
            mMaxMessageRetries = maxRetries;
            return this;
        }

        /**
         * Sets the importance of the platform Message. Sometimes has an effect on either delivery
         * or device notification timing. For example, {@link MessagePriority} PRIORITY_HIGH will
         * also set the setShouldDelayWhileIdle attribute to false. and on GCM will ignore Doze mode.
         * <p>
         * Note: Be considerate of users' time. Set the priority to use the lowest priority the
         * message data can afford.
         *
         * @param messagePriority Priority of platform message.
         */
        public Builder setMessagePriority(@Nonnull MessagePriority messagePriority) {
            mMessagePriority = messagePriority;
            return this;
        }

        /**
         * Should the message Wake up the device (and often screen, sound etc) or is it okay to only
         * alert when the user is engaging with device.
         * <p>
         * Setting to false will override the {@link MessagePriority} and set it
         * to MessagePriority.HIGH.
         *
         * @param shouldDelayWhileIdle default is true.
         * @deprecated Google have deprecated this feature in favor of setMessagePriority
         */
        @Deprecated
        public Builder setShouldDelayWhileIdle(boolean shouldDelayWhileIdle) {
            mShouldDelayWhileIdle = shouldDelayWhileIdle;
            return this;
        }

        /**
         * Set the collapse key. If two messages are sent to a device with the same key, the first message
         * (alert, etc) will be overwritten by the second.
         *
         * @param collapseKey collapse key string.
         */
        public Builder setCollapseKey(@Nonnull String collapseKey) {
            mCollapseKey = collapseKey;
            return this;
        }

    }
} public interface TaskQueueListener {

    /**
     * Invoked with a collection of updated {@link Recipient} registrations.
     *
     * @param recipients A an old and new {@link Recipient} in a Map of stale and updated
     *                   registrations s in the format:
     *                   <Registration staleRegistration, Registration updatedRegistration>
     */
    void updatedRecipients(@Nonnull List<UpdatedRecipient> recipients);

    /**
     * Notification of a failure for a particular message recipient. This does not mean
     * the message as a whole has fully completed or failed yet.
     *
     * @param recipients Recipients that a message was undeliverable to.
     */
    void failedRecipients(@Nonnull List<Recipient> recipients);

    /**
     * All recipients in a Message were processed and sent, but some recipients may
     * still have failures.
     *
     * @param message The taskCompleted Task.
     */
    void messageCompleted(@Nonnull Message message);

    /**
     * All recipients in a message failed due to an unrecoverable failure, and
     * will not be retried.
     *
     * @param message Message which failed.
     * @param failure Failure details
     */
    void messageFailed(@Nonnull Message message, PlatformFailure failure);
} public class UpdatedRecipient {
    private Recipient mStaleRecipient;
    private Recipient mUpdatedRecipient;

    // Private access for immutability.
    private UpdatedRecipient() {
    }

    public UpdatedRecipient(@Nonnull Recipient staleRecipient, @Nonnull Recipient updatedRecipient) {
        mStaleRecipient = staleRecipient;
        mUpdatedRecipient = updatedRecipient;
    }

    @Nonnull
    public Recipient getStaleRecipient() {
        return mStaleRecipient;
    }

    @Nonnull
    public Recipient getUpdatedRecipient() {
        return mUpdatedRecipient;
    }
} @Entity
@Table(name = "credentials", schema = "pushservices")
public class Credentials extends Model {
    public static Finder<Long, Credentials> find = new Finder<>(Credentials.class);

    @Id
    @JsonIgnore
    @Column(name = "id")
    @SequenceGenerator(name = "gen", sequenceName = "pushservices.credentials_id_seq", allocationSize = 1)
    @GeneratedValue(strategy = GenerationType.SEQUENCE, generator = "gen")
    protected Long id;

    @OneToOne(mappedBy = "credentials")
    private Message message;

    @Enumerated(EnumType.STRING)
    @Column(name = "platform")
    private PlatformType platformType;

    @Column(name = "authorisation_key", columnDefinition = "TEXT")
    private String authKey;

    @Column(name = "certificate_body", columnDefinition = "TEXT")
    private String certBody;

    @Column(name = "package_uri", columnDefinition = "TEXT")
    private String packageUri;

    public Credentials(PlatformType platformType) {
        setPlatformType(platformType);
    }

    public Long getId() {
        return id;
    }

    public Message getMessage() {
        return message;
    }

    public void setMessage(Message message) {
        this.message = message;
    }

    public PlatformType getPlatformType() {
        return platformType;
    }

    private void setPlatformType(PlatformType platformType) {
        this.platformType = platformType;
    }

    public String getAuthKey() {
        return authKey;
    }

    public void setAuthKey(String authKey) {
        this.authKey = authKey;
    }

    public String getCertBody() {
        return certBody;
    }

    public void setCertBody(String certBody) {
        this.certBody = certBody;
    }

    public String getPackageUri() {
        return packageUri;
    }

    public void setPackageUri(String packageUri) {
        this.packageUri = packageUri;
    }
} @Entity
@Table(name = "messages", schema = "pushservices")
public class Message extends Model {
    public static Finder<Long, Message> find = new Finder<>(Message.class);
    private static final int TTL_SECONDS_DEFAULT = 60 * 60 * 24 * 7;

    @Id
    @JsonIgnore
    @Column(name = "id")
    @SequenceGenerator(name = "gen", sequenceName = "pushservices.message_id_seq", allocationSize = 1)
    @GeneratedValue(strategy = GenerationType.SEQUENCE, generator = "gen")
    private Long id;

    @OneToMany(cascade = CascadeType.ALL, fetch = FetchType.EAGER)
    private List<Recipient> recipients;

    @OneToOne(cascade = CascadeType.ALL, fetch = FetchType.EAGER)
    @JoinColumn(name = "credentials_id")
    private Credentials credentials;

    @OneToMany(cascade = CascadeType.ALL, fetch = FetchType.EAGER)
    private List<PayloadElement> payloadData;

    @Column(name = "collapse_key")
    private String collapseKey;

    @Column(name = "priority")
    @Enumerated(EnumType.STRING)
    private MessagePriority messagePriority = MessagePriority.PRIORITY_NORMAL;

    @Column(name = "ttl_seconds")
    private int ttlSeconds = TTL_SECONDS_DEFAULT;

    @Column(name = "delay_while_idle")
    private boolean shouldDelayWhileIdle;

    @Column(name = "dry_run")
    private boolean dryRun;

    @Column(name = "maximum_retries")
    private int maximumRetries = 10;

    @Basic
    @Column(name = "added_time", columnDefinition = "timestamp without time zone")
    @Temporal(TemporalType.TIMESTAMP)
    private Date addedTime;

    @PrePersist
    public void updatedTime() {
        setAddedTime(new Date());
    }

    public Long getId() {
        return id;
    }

    public List<Recipient> getRecipients() {
        return recipients;
    }

    public void setRecipients(List<Recipient> recipients) {
        this.recipients = recipients;
    }

    public Credentials getCredentials() {
        return credentials;
    }

    public void setCredentials(Credentials credentials) {
        this.credentials = credentials;
    }

    public List<PayloadElement> getPayloadData() {
        return payloadData;
    }

    public void setPayloadData(List<PayloadElement> payloadData) {
        this.payloadData = payloadData;
    }

    public String getCollapseKey() {
        return collapseKey;
    }

    public void setCollapseKey(String collapseKey) {
        this.collapseKey = collapseKey;
    }

    public MessagePriority getMessagePriority() {
        return messagePriority;
    }

    public void setMessagePriority(MessagePriority messagePriority) {
        this.messagePriority = messagePriority;
    }

    public int getTtlSeconds() {
        return ttlSeconds;
    }

    public void setTtlSeconds(int ttlSeconds) {
        this.ttlSeconds = ttlSeconds;
    }

    public boolean isShouldDelayWhileIdle() {
        return shouldDelayWhileIdle;
    }

    public void setShouldDelayWhileIdle(boolean shouldDelayWhileIdle) {
        this.shouldDelayWhileIdle = shouldDelayWhileIdle;
    }

    public boolean isDryRun() {
        return dryRun;
    }

    public void setDryRun(boolean dryRun) {
        this.dryRun = dryRun;
    }

    public int getMaximumRetries() {
        return maximumRetries;
    }

    public void setMaximumRetries(int maximumRetries) {
        this.maximumRetries = maximumRetries;
    }

    public Date getAddedTime() {
        return addedTime;
    }

    private void setAddedTime(Date addedTime) {
        this.addedTime = addedTime;
    }
} @Entity
@Table(name = "recipient_failures", schema = "pushservices")
public class PlatformFailure extends Model {
    public static Finder<Long, PlatformFailure> find = new Finder<>(PlatformFailure.class);

    @Id
    @JsonIgnore
    @Column(name = "id")
    @SequenceGenerator(name = "gen", sequenceName = "pushservices.failure_id_seq", allocationSize = 1)
    @GeneratedValue(strategy = GenerationType.SEQUENCE, generator = "gen")
    private Long id;

    @OneToOne(mappedBy = "failure")
    private Recipient recipient;

    @Column(name = "type")
    @Enumerated(EnumType.STRING)
    private FailureType failureType;

    @Column(name = "message")
    private String failureMessage;

    @Basic
    @Column(name = "fail_time", columnDefinition = "timestamp without time zone")
    @Temporal(TemporalType.TIMESTAMP)
    private Date failTime;

    public PlatformFailure(String failureMessage) {
        setFailureMessage(failureMessage);
    }

    public PlatformFailure(FailureType failureType, String failureMessage) {
        setFailureType(failureType);
        setFailureMessage(failureMessage);
    }

    public PlatformFailure(FailureType failureType, String failureMessage, Date failTime) {
        setFailureType(failureType);
        setFailureMessage(failureMessage);
        setFailTime(failTime);
    }

    @PrePersist
    public void prePersist() {
        if (failTime == null) {
            setFailTime(new Date());
        }
    }

    public Long getId() {
        return id;
    }

    public Recipient getRecipient() {
        return recipient;
    }

    public void setRecipient(Recipient recipient) {
        this.recipient = recipient;
    }

    public FailureType getFailureType() {
        return failureType;
    }

    public void setFailureType(FailureType failureType) {
        this.failureType = failureType;
    }

    public String getFailureMessage() {
        return failureMessage;
    }

    public void setFailureMessage(String failureMessage) {
        this.failureMessage = failureMessage;
    }

    public Date getFailTime() {
        return failTime;
    }

    public void setFailTime(Date failTime) {
        this.failTime = failTime;
    }
} @Entity
@Table(name = "recipients", schema = "pushservices")
public class Recipient extends Model {
    public static Finder<Long, Recipient> find = new Finder<>(Recipient.class);

    @Id
    @JsonIgnore
    @Column(name = "id")
    @SequenceGenerator(name = "gen", sequenceName = "pushservices.recipient_id_seq", allocationSize = 1)
    @GeneratedValue(strategy = GenerationType.SEQUENCE, generator = "gen")
    private Long id;

    @OneToOne(fetch = FetchType.EAGER, cascade = CascadeType.ALL)
    private PlatformFailure failure;

    @Column(name = "token", columnDefinition = "TEXT")
    private String token;

    @ManyToOne
    private Message message;

    @Column(name = "state")
    private RecipientState state;

    @Column(name = "send_attempts")
    private int sendAttemptCount;

    @Basic
    @Temporal(TemporalType.TIMESTAMP)
    @Column(name = "last_attempt", columnDefinition = "timestamp without time zone")
    private Date lastSendAttempt;

    @Basic
    @Temporal(TemporalType.TIMESTAMP)
    @Column(name = "next_attempt", columnDefinition = "timestamp without time zone")
    private Date nextAttempt;

    @PrePersist
    public void prePersist() {
        state = RecipientState.STATE_IDLE;
        sendAttemptCount = 0;
    }

    public Recipient(@Nonnull String token) {
        setToken(token);
    }

    public Long getId() {
        return id;
    }

    public PlatformFailure getPlatformFailure() {
        return failure;
    }

    public void setFailure(PlatformFailure failure) {
        this.failure = failure;
    }

    public String getToken() {
        return token;
    }

    private void setToken(String token) {
        this.token = token;
    }

    public Message getMessage() {
        return message;
    }

    public void setMessage(Message message) {
        this.message = message;
    }

    public RecipientState getState() {
        return state;
    }

    public void setState(RecipientState state) {
        this.state = state;
    }

    public int getSendAttemptCount() {
        return sendAttemptCount;
    }

    public void setSendAttemptCount(int sendAttemptCount) {
        this.sendAttemptCount = sendAttemptCount;
    }

    public Date getLastSendAttempt() {
        return lastSendAttempt;
    }

    public void setLastSendAttempt(Date lastSendAttempt) {
        this.lastSendAttempt = lastSendAttempt;
    }

    public Date getNextAttempt() {
        return nextAttempt;
    }

    public void setNextAttempt(Date nextAttempt) {
        this.nextAttempt = nextAttempt;
    }
} @Singleton
public class TaskQueue {
    // Delay each message polling interval to a 1/4 second so the server isn't flooded.
    private static final long TASKQUEUE_POLL_INTERVAL_MS = 500;

    // Collection containing all messages that have not returned from the provider with a success or fail.
    private Set<Long> mActiveMessages = new HashSet<>();
    private Map<Long, PlatformResponseCallback> mInternalListeners = new HashMap<>();
    private Map<Long, TaskQueueListener> mExternalListeners = new HashMap<>();

    // TaskQueue thread and queue related members.
    private BlockingQueue<Message> mMessageProcessQueue = new ArrayBlockingQueue<>(5000);
    private MessageProducerThread mQueueProducerThread;
    private MessageConsumerThread mQueueConsumerThread;

    private GcmMessageDispatcher mGcmMessageDispatcher;
    private MessagesDao mMessagesDao;

    /**
     * Privately instantiate the TaskQueue with required Dependencies.
     *
     * @param messagesDao          Message persistence.
     * @param gcmMessageDispatcher GCM Google message dispatcher.
     */
    @Inject
    protected TaskQueue(MessagesDao messagesDao, GcmMessageDispatcher gcmMessageDispatcher) {
        mMessagesDao = messagesDao;
        mGcmMessageDispatcher = gcmMessageDispatcher;
    }

    @SuppressWarnings("unused")
    private TaskQueue() {
    }

    /**
     * Checks that the message consumer process is active and running, and starts the
     * TaskQueue {@link Message} polling process if it is not.
     */
    public synchronized void startup() {
        Logger.info("TaskQueue Startup");

        // Start the message producer and consumer queues.
        startProducerQueue();
        startConsumerQueue();

        // Check and re-queue existing pending messages.
        queuePendingMessages();
    }

    private synchronized void startProducerQueue() {
        if (mQueueProducerThread == null || !mQueueProducerThread.isAlive()) {
            mQueueProducerThread = new MessageProducerThread(mMessageProcessQueue);
            Logger.debug("Starting the TaskQueue Producer process.");
            mQueueProducerThread.start();
        }
    }

    private synchronized void startConsumerQueue() {
        if (mQueueConsumerThread == null || !mQueueConsumerThread.isAlive()) {
            mQueueConsumerThread = new MessageConsumerThread();
            Logger.debug("Starting the TaskQueue Consumer process.");
            mQueueConsumerThread.start();
        }
    }

    /**
     * Checks that the message ask consumer process is active and running, and starts the
     * TaskQueue {@link Message} polling process if it is not.
     */
    public synchronized void shutdown() {
        if (mQueueProducerThread != null) {
            Logger.debug("Shutting down the TaskQueue Producer process.");
            mQueueProducerThread.interrupt();
        }

        if (mQueueConsumerThread != null) {
            Logger.debug("Shutting down the TaskQueue Consumer process.");
            mQueueConsumerThread.interrupt();
        }
    }

    /**
     * Queue any messages that have pending recipients. If you call this method, be sure that the messages
     * in question are not being processed by the ConsumerThread, or are awaiting results
     * from the dispatcher. Ideally, the TaskQueue should not be started.
     */
    private void queuePendingMessages() {
        // Get outstanding incomplete message messages and startup queue producer thread.
        List<Message> pendingMessages = mMessagesDao.fetchPendingMessages();
        Logger.info(String.format("Pending Message check. Active queue: %d.", pendingMessages.size()));

        for (Message message : pendingMessages) {
            if (!isMessageInQueue(message)) {
                queueMessage(message);
            }
        }
    }

    /**
     * Add new messages to the TaskQueue.
     *
     * @param messages message to add to TaskQueue and dispatch.
     * @param callback TaskQueue callback to get processing updates..
     */
    public synchronized void queueMessages(@Nonnull List<Message> messages, TaskQueueListener callback) throws MessageValidationException {
        Logger.debug("Retrieved a message from the client to queue.");

        for (Message message : messages) {
            // Verify the Message has all required attributes.
            MessageHelper.verifyMessage(message);

            if (!mMessagesDao.saveMessage(message)) {
                throw new MessageValidationException("Error saving message. Check persistence settings.");
            }

            // Add client TaskQueue listener.
            if (mExternalListeners.get(message.getId()) == null && callback != null) {
                mExternalListeners.put(message.getId(), callback);
            }

            // Queue the message if it is not active.
            if (!isMessageInQueue(message)) {
                queueMessage(message);
            }
        }
    }

    /**
     * Add a message to the MessageQueue process.
     *
     * @param message Message to add to the queue.
     */
    private void queueMessage(Message message) {
        if (message != null && message.getId() != null) {
            // Add to processing messages collection.
            mActiveMessages.add(message.getId());

            //  Add the internal provider dispatcher listener.
            if (mInternalListeners.get(message.getId()) == null) {
                mInternalListeners.put(message.getId(), new PlatformResponseCallback());
            }

            // Start the message queues if they are not already active and add to staging collection
            // to be picked up by the consumer
            startProducerQueue();
            startConsumerQueue();
            mQueueProducerThread.dispatchQueue.add(message);
        }
    }

    private void removeMessageFromQueue(Message message) {
        if (message != null && message.getId() != null) {
            mActiveMessages.remove(message.getId());
            mMessageProcessQueue.remove(message);

            mInternalListeners.remove(message.getId());
            mExternalListeners.remove(message.getId());
        }
    }

    /**
     * Ascertain whether the current message is being worked on or queued, and return true
     * if it is.
     *
     * @param message Message to check.
     * @return true if the message is currently active.
     */
    private boolean isMessageInQueue(@Nonnull Message message) {
        if (mActiveMessages.contains(message.getId())) {
            Logger.debug(String.format("Message %d found in active process queue.", message.getId()));
            return true;

        } else {
            return false;
        }
    }

    /**
     * Dispatch a Message {@link Message}
     *
     * @param message The {@link Message} to dispatch.
     */
    private void dispatchMessage(@Nonnull Message message) {
        int messageRecipientCount = 0;

        try {
            if (!MessageHelper.isMessageReady(message)) {
                Logger.warn(String.format("Message %d has already finished and won't be dispatched.", message.getId()));
                return;
            }

        } catch (MessageValidationException e) {
            Logger.error(String.format("Message %d invalid: %s", message.getId(), e.getMessage()));
            PlatformFailure failure = new PlatformFailure(FailureType.MESSAGE_PAYLOAD_INVALID, e.getMessage(), new Date());
            MessageHelper.failMessage(message, failure);
            mMessagesDao.saveMessage(message);

            removeMessageFromQueue(message);
            return;
        }

        // Set recipient states to processing for ready recipients.
        for (Recipient recipient : message.getRecipients()) {
            // The recipient is out of the cooling off period.
            if (!MessageHelper.isRecipientCoolingOff(recipient)) {
                recipient.setState(RecipientState.STATE_PROCESSING);
                recipient.setLastSendAttempt(new Date());
                messageRecipientCount += 1;
            } else {
                Logger.debug(String.format("Recipient %d still within cooling down period", recipient.getId()));
            }
        }

        // If there are pendingRecipients, dispatch the message.
        if (messageRecipientCount > 0) {
            if (!mMessagesDao.saveMessage(message)) {
                removeMessageFromQueue(message);
                return;
            }

            PlatformResponseCallback platformResponse = mInternalListeners.get(message.getId());
            if (platformResponse == null) {
                platformResponse = new PlatformResponseCallback();
                mInternalListeners.put(message.getId(), platformResponse);
            }

            // Dispatch the message.
            Logger.debug(String.format("Dispatching message %d", message.getId()));
            mGcmMessageDispatcher.dispatchMessage(message, platformResponse);

        } else {
            // At least one recipient in the message is pending AND cooling off. Requeue message.
            Logger.debug(String.format("Only cooling down recipients ready in message %d. Requeuing", message.getId()));
            queueMessage(message);
        }
    }

    /*
     * Response back from the push message push-services (APNS or GCM) for the sent message.
     * The message may have either succeeded or failed.
     */
    private class PlatformResponseCallback implements PlatformResponse {
        /**
         * A raw, unsorted callback for results returned for a message send from the push-services.
         *
         * @param message           The original message sent to the push-services
         * @param successRecipients The list of completed Recipients
         * @param failedRecipients  The map of failed Recipients with the Failure.
         * @param recipientsToRetry The list of recipients to exponentially retry.
         **/
        @Override
        public void messageSuccess(@Nonnull Message message, @Nonnull List<Recipient> successRecipients, @Nonnull List<Recipient> failedRecipients,
                                   @Nonnull List<UpdatedRecipient> recipientsToUpdate, @Nonnull List<Recipient> recipientsToRetry) {
            mMessagesDao.saveMessage(message);

            if (!MessageHelper.hasMessageCompleted(message)) {
                queueMessage(message);
                return;
            }

            // Client responses:
            TaskQueueListener messageCallback = mExternalListeners.get(message.getId());
            if (messageCallback != null) {

                // Invoke updatedRecipients() callback.
                if (!recipientsToUpdate.isEmpty()) {
                    Logger.debug(String.format("[%d] recipients requiring token change", recipientsToUpdate.size()));
                    messageCallback.updatedRecipients(recipientsToUpdate);
                }

                // Invoke individual recipient failure callback.
                if (!failedRecipients.isEmpty()) {
                    Logger.debug(String.format("[%d] failed recipients", failedRecipients.size()));
                    messageCallback.failedRecipients(failedRecipients);
                }

                // Invoke messageCompleted() callback.
                if (MessageHelper.hasMessageCompleted(message)) {
                    Logger.debug(String.format("[%d] successful recipients", successRecipients.size()));
                    messageCallback.messageCompleted(message);
                }
            }

            Logger.info(String.format("Message %d completed.", message.getId()));
            removeMessageFromQueue(message);
        }

        @Override
        public void messageFailure(@Nonnull Message message, @Nonnull PlatformFailure failure) {
            Logger.error(String.format("Platform error '%1$s' from provider for message %2$d",
                    failure.getFailureType().name(), message.getId()));

            // Update the message entry.
            mMessagesDao.saveMessage(message);

            if (!MessageHelper.hasMessageCompleted(message)) {
                queueMessage(message);
                return;
            }

            TaskQueueListener messageCallback = mExternalListeners.get(message.getId());
            if (messageCallback != null) {
                messageCallback.messageFailed(message, failure);
            }

            removeMessageFromQueue(message);
        }
    }

    /**
     * Thread that loops through the blocking queue and sends
     * messages to the platform push provider. This will block if there are no messages to take,
     * hence the Runnable.
     */
    private class MessageProducerThread extends Thread {
        private BlockingQueue<Message> dispatchQueue;

        MessageProducerThread(BlockingQueue<Message> blockingQueue) {
            dispatchQueue = blockingQueue;
        }

        @Override
        public void run() {
            while (true) {
                try {
                    Message waitingMessage = dispatchQueue.take();
                    Logger.debug("Adding item from Producer into MessageQueue");

                    mMessageProcessQueue.offer(waitingMessage);

                    // Sleep on while so often so the server isn't hammered.
                    Thread.sleep(TASKQUEUE_POLL_INTERVAL_MS);

                } catch (InterruptedException e) {
                    Logger.debug("InterruptedException was invoked in the MessageProducerThread");
                }
            }
        }
    }

    /**
     * Thread that loops through the blocking queue and sends
     * messages to the platform push provider. This will block if there are no messages to take,
     * hence the Runnable.
     */
    private class MessageConsumerThread extends Thread {
        @Override
        public void run() {
            try {
                while (true) {
                    // Take and remove the task from queue.
                    Message message = mMessageProcessQueue.take();
                    Logger.debug(String.format("TaskQueue processing queued message %d", message.getId()));

                    // Dispatch the queued message.
                    dispatchMessage(message);
                    Thread.sleep(TASKQUEUE_POLL_INTERVAL_MS);
                }

            } catch (InterruptedException e) {
                Logger.debug("InterruptedException was invoked in the MessageConsumeThread");
            }
        }
    }
}
import enums.pushservices.PlatformType;
import exceptions.pushservices.MessageValidationException;
import helpers.pushservices.MessageBuilder;
import interfaces.pushservices.TaskQueueListener;
import models.pushservices.app.UpdatedRecipient;
import models.pushservices.db.Credentials;
import models.pushservices.db.Message;
import models.pushservices.db.PlatformFailure;
import models.pushservices.db.Recipient;
import play.Logger;
import play.mvc.Controller;
import play.mvc.Result;
import services.pushservices.TaskQueue;
import javax.annotation.Nonnull;
import javax.inject.Inject;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
package controllers;



/**
 * Illustrates the push-services module API.
 */
public class SampleController extends Controller {
    private TaskQueue mTaskQueue;

    @Inject
    public SampleController(TaskQueue taskQueue) {
        mTaskQueue = taskQueue;
    }

    public Result robots() {
        return ok("User-agent: *\nDisallow: /");
    }

    public CompletionStage<Result> index() {
        return CompletableFuture.supplyAsync(() -> {
            /*
             * Send a push message to a set of recipients.
             * NOTE: Contains sample tokens / credentials that will fail. Replace with your own.
             */
            Credentials sampleCredentials = new Credentials(PlatformType.SERVICE_GCM);
            sampleCredentials.setPackageUri("com.company.app");
            sampleCredentials.setAuthKey("AEFbawuefAWEFwaEFea9OAKFAEWfeawKk");

            Set<String> sampleDeviceTokens = new HashSet<>();
            sampleDeviceTokens.add("d1NqItxeuykwdWQaefeAWEFwawefEWEDgregsdKXHfxi4j2VHfLv9TcE14QwjckJ3qB4gm6");
            sampleDeviceTokens.add("7RdoKVpIClSiijili1DUTtUDYJv00rBLTBf0nDsfdiktPsnZDU:APA91bEHetahBunbXxD");
            sampleDeviceTokens.add("fdiktPsnZDU:APA91bEHetahBunbXST6jfiAGBd0TnDcZsFFEZfvbbuT8HHcxeCaMhWTLT");

            Map<String, String> sampleData = new HashMap<>();
            sampleData.put("sample_first_key", "sample_first_value");
            sampleData.put("sample_second_key", "sample_second_value");

            MessageBuilder.Builder messageBuilder = new MessageBuilder.Builder()