1
\$\begingroup\$

We need to send messages with highest priority first so we use a PriorityQueue for our purpose.

PriorityQueue<MessageData> queue = new PriorityQueue<MessageData>();

However, we also want our queue to behave like a sorted set as well. Therefore, we adapt the PriorityQueue to ignore insertions which repeat existing members.

import java.util.Comparator;
import java.util.PriorityQueue;

public class PrioritySet<E> extends PriorityQueue<E> {

    private static final long serialVersionUID = 34658778L;

    public PrioritySet() {
        super();
    }

    public PrioritySet(int initialCapacity, Comparator<? super E> comparator) {
        super(initialCapacity, comparator);
    }

    @Override
    public boolean offer(E e) {
        boolean isAdded = false;
        if(!super.contains(e)) {
            isAdded = super.offer(e);
        }
        return isAdded;
    }
}

Now our app specific implementation of the data structure.

import java.util.Comparator;

public class MessagePrioritySet extends PrioritySet<MessageData> {

    private static final long serialVersionUID = 34658779L;

    private int minPriorityNumber;

    public MessagePrioritySet() {
        super();
    }

    public MessagePrioritySet(int initialCapacity, Comparator<MessageData> comparator) {
        super(initialCapacity, comparator);
    }

    public synchronized int getMinPriorityNumber() {
        return minPriorityNumber;
    }

    public synchronized void setMinPriorityNumber(int minPriorityNumber) {
        this.minPriorityNumber = minPriorityNumber;
    }

    @Override
    public synchronized boolean offer(MessageData notification) {
        boolean isAdded = super.offer(notification);
        if (notification.getPriority() < minPriorityNumber)
            minPriorityNumber = notification.getPriority();
        return isAdded;
    }

    public synchronized void reportSent(MessageData notification) {
        MessageData nextMessageData = peek();
        if (nextMessageData == null)
            minPriorityNumber = 0;
        else if (nextMessageData.getPriority() > notification.getPriority())
            minPriorityNumber = nextMessageData.getPriority();
    }
}

Here, we want the data structure to be aware of the minimum priority value of the messages so we declare an instance variable for that. The priority of the incoming message is checked and if this priority is lower than the stored value, the value stored is updated. The use of the class is required to report any sent messages. If no other member of the data structure has a priority as low as the one being removed, then the next element's priority becomes the stored priority.

Two threads share the implemented queue. One thread fetches data from the database and inserts them into the queue. The other reads the queue and sends the highest priority message with the lowest priority number. Because the queue sets the minimum priority value to 0 and the thread which fetches data from the database reads rows with priority value lower than or equal to the minimum value stored in the queue if the stored minimum value is not zero, we can be pretty sure that while the current messages in the queue are being sent, only the new messages which are more important than those already in the queue will be added to the queue.

private void startMptSender() {
    sleepInterval = 1000;
    final MessagePrioritySet messagePrioritySet = new MessagePrioritySet();

    Runnable mptReader = new Runnable() {

        @Override
        public void run() {
            while (true) {
                List<MessageData> messageDataList;

                if (messagePrioritySet.getMinPriorityNumber() == 0)
                    messageDataList = messageDao.readSMSMpt();
                else
                    messageDataList = messageDao.readSMSMpt(messagePrioritySet.getMinPriorityNumber());

                for (MessageData messageData : messageDataList) {
                    messagePrioritySet.offer(messageData);
                }
                try {
                    Thread.sleep(sleepInterval);
                } catch (InterruptedException ie) {

                }
            }
        }
    };

    executor.execute(mptReader);

    Runnable mptPusher = new Runnable() {

        @Override
        public void run() {
            while (status) {
                if (messagePrioritySet.size() > 0) {

                    while (messagePrioritySet.size() != 0) {
                        MessageData noti = messagePrioritySet.remove();
                        mptSender.sendSms(noti);
                        messageDao.markNotificationAsRead(noti.getSyskey());
                        messagePrioritySet.reportSent(noti);
                        try {
                            Thread.sleep(sleepInterval);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                } else {
                    try {
                        Thread.sleep(sleepInterval);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    };

    executor.execute(mptPusher);
}

}

We would like to make sure that our thinking is correct and would like anyone to point out weaknesses in our design.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.