Queue with timeout that removes not retrieved objects

Question

I have thread which writes something to socket and then starts waiting for response. This thread needs to retrieve response in a given time, 100 milliseconds for example. My idea is to create reader thread which will read from socket and store response in a blocking queue. First thread just waits when reader finishes work and put object in the queue. If reader does it in a time everything OK, if reader does not I want to throw exception and it's also OK. The problem is that when I come after TimeOutException, object that wasn't taken from queue because of timeout is still there, and may be even just prepared to be inserted into the queue. And I want only next object. How should I skip this message? Or may be there is better approach to this problem?

Also, connection is keep alive, so socket timeout is not suitable.

As far as I known, a socket timeout is just an event that tells you a read is blocking too long. You can ignore this timeout event and continue to read: the socket connection has not changed. Have you tried this? — vanOekel
"This thread needs to retrieve response in a given time, 100 milliseconds for example. " What should the program do when read timeout? — Weibo Li

Raffaele Rossi Raffaele Rossi · Accepted Answer · 2014-01-17T13:25:28

I would subclass BlockingQueue to override poll() and offer() methods, which simply invokes super.poll() and super.offer(). It will add the timestamp when offering new items and discards any item that has been in the queue too long when polling. Something like:

public class TimedItemBlockingQueue<T> extends ArrayBlockingQueue<TimedItem<T>> {
    private class TimedItem<T> {
        T data;
        long timestamp;
        TimedItem<T>(T data) {
            this.data = data;
            this.timeStamp = <get_tiemstamp>
    }

    T poll() {
        TimedItem<T> item = super.poll();
        if (isValid(item.timestamp)) {
            return item.data;
        }
        /* Plus extra code to handle exceptions and other cases */
    }

    boolean offer(T data) {
        TimedItem<T> item = new TimedItem<T>(data);
        return super.offer(item);
}

You might need to override all poll() methods.

In this case the first thread would just invoke poll() without even knowing whether there is such a queue in the middle or how many objects have "expired"

Queue with timeout that removes not retrieved objects

2 Answers