queue2.java

来自「JGRoups源码」· Java 代码 · 共 688 行 · 第 1/2 页

            /*the queue is empty*/
            if(el == null) return;

            /*check to see if the head element is the one to be removed*/
            if(el.obj.equals(obj)) {
                /*the head element matched we will remove it*/
                head=el.next;
                el.next=null;
                /*check if we only had one object left
                 *at this time the queue becomes empty
                 *this will set the tail=head=null
                 */
                if(size == 1)
                    tail=head;  // null
                decrementSize();
                removed=true;

                /*and end the operation, it was successful*/
                return;
            }

            /*look through the other elements*/
            while(el.next != null) {
                if(el.next.obj.equals(obj)) {
                    tmp_el=el.next;
                    if(tmp_el == tail) // if it is the last element, move tail one to the left (bela Sept 20 2002)
                        tail=el;
                    el.next=el.next.next;  // point to the el past the next one. can be null.
                    tmp_el.next=null;
                    decrementSize();
                    removed=true;
                    break;
                }
                el=el.next;
            }
        }
        catch(InterruptedException e) {

        }
        finally {
            if(removed)
                remove_condvar.broadcast();
            mutex.release();
        }
    }


    /**
     * returns the first object on the queue, without removing it.
     * If the queue is empty this object blocks until the first queue object has
     * been added
     * @return the first object on the queue
     */
    public Object peek() throws QueueClosedException {
        Object retval=null;

        try {
            mutex.acquire();
            while(size == 0) {
                if(closed)
                    throw new QueueClosedException();
                try {
                    add_condvar.await();
                }
                catch(InterruptedException ex) {
                }
            }

            if(closed)
                throw new QueueClosedException();

            retval=(head != null)? head.obj : null;

            // @remove:
            if(retval == null) {
                // print some diagnostics
                if(log.isErrorEnabled()) log.error("retval is null: head=" + head + ", tail=" + tail + ", size()=" + size() +
                        ", num_markers=" + num_markers + ", closed()=" + closed());
            }
        }
        catch(InterruptedException e) {

        }
        finally {
            mutex.release();
        }

        if(retval == endMarker) {
            close(false); // mark queue as closed
            throw new QueueClosedException();
        }

        return retval;
    }


    /**
     * returns the first object on the queue, without removing it.
     * If the queue is empty this object blocks until the first queue object has
     * been added or the operation times out
     * @param timeout how long in milli seconds will this operation wait for an object to be added to the queue
     *        before it times out
     * @return the first object on the queue
     */

    public Object peek(long timeout) throws QueueClosedException, TimeoutException {
        Object retval=null;

        try {
            mutex.acquire();
            if(size == 0) {
                if(closed)
                    throw new QueueClosedException();
                try {
                    mutex.wait(timeout);
                }
                catch(InterruptedException ex) {
                }
            }
            if(closed)
                throw new QueueClosedException();

            retval=head != null? head.obj : null;

            if(retval == null) throw new TimeoutException();

            if(retval == endMarker) {
                close(false);
                throw new QueueClosedException();
            }
        }
        catch(InterruptedException e) {

        }
        finally {
            mutex.release();
        }
        return retval;
    }


    /**
     Marks the queues as closed. When an <code>add</code> or <code>remove</code> operation is
     attempted on a closed queue, an exception is thrown.
     @param flush_entries When true, a end-of-entries marker is added to the end of the queue.
     Entries may be added and removed, but when the end-of-entries marker
     is encountered, the queue is marked as closed. This allows to flush
     pending messages before closing the queue.
     */
    public void close(boolean flush_entries) {
        if(flush_entries) {
            try {
                add(endMarker); // add an end-of-entries marker to the end of the queue
                num_markers++;
            }
            catch(QueueClosedException closed) {
            }
            return;
        }

        try {
            mutex.acquire();
            closed=true;
            try {
                add_condvar.broadcast();
                remove_condvar.broadcast();
            }
            catch(Exception e) {
                if(log.isErrorEnabled()) log.error("exception=" + e);
            }
        }
        catch(InterruptedException e) {

        }
        finally {
            mutex.release();
        }
    }


    /**
     * resets the queue.
     * This operation removes all the objects in the queue and marks the queue open
     */
    public void reset() {
        num_markers=0;
        if(!closed)
            close(false);

        try {
            mutex.acquire();
            size=0;
            head=null;
            tail=null;
            closed=false;
        }
        catch(InterruptedException e) {

        }
        finally {
            mutex.release();
        }
    }


    /**
     * returns the number of objects that are currently in the queue
     */
    public int size() {
        return size - num_markers;
    }

    /**
     * prints the size of the queue
     */
    public String toString() {
        return "Queue2 (" + size() + ") messages";
    }

    /**
     * Dumps internal state @remove
     */
    public String debug() {
        return toString() + ", head=" + head + ", tail=" + tail + ", closed()=" + closed() + ", contents=" + getContents();
    }


    /**
     * returns a vector with all the objects currently in the queue
     */
    public Vector getContents() {
        Vector retval=new Vector();
        Element el;

        try {
            mutex.acquire();
            el=head;
            while(el != null) {
                retval.addElement(el.obj);
                el=el.next;
            }
        }
        catch(InterruptedException e) {

        }
        finally {
            mutex.release();
        }
        return retval;
    }


    /**
     * Blocks until the queue has no elements left. If the queue is empty, the call will return
     * immediately
     * @param timeout Call returns if timeout has elapsed (number of milliseconds). 0 means to wait forever
     * @throws QueueClosedException Thrown if queue has been closed
     * @throws TimeoutException Thrown if timeout has elapsed
     */
    public void waitUntilEmpty(long timeout) throws QueueClosedException, TimeoutException {
        long time_to_wait=timeout >=0? timeout : 0; // eliminate negative values

        try {
            mutex.acquire();

            if(timeout == 0) {
                while(size > 0 && closed == false) {
                    remove_condvar.await();
                }
            }
            else {
                long start_time=System.currentTimeMillis();

                while(time_to_wait > 0 && size > 0 && closed == false) {
                    try {
                        remove_condvar.timedwait(time_to_wait);
                    }
                    catch(InterruptedException ex) {

                    }
                    time_to_wait=timeout - (System.currentTimeMillis() - start_time);
                }

                if(size > 0)
                    throw new TimeoutException("queue has " + size + " elements");
            }

            if(closed)
                throw new QueueClosedException();
        }
        catch(InterruptedException e) {

        }
        finally {
            mutex.release();
        }
    }


    /* ------------------------------------- Private Methods ----------------------------------- */


    /**
     * Removes the first element. Returns null if no elements in queue.
     * Always called with mutex locked (we don't have to lock mutex ourselves)
     */
    private Object removeInternal() {
        Element retval;

        /*if the head is null, the queue is empty*/
        if(head == null)
            return null;

        retval=head;       // head must be non-null now

        head=head.next;
        if(head == null)
            tail=null;

        decrementSize();

        remove_condvar.broadcast(); // todo: correct ?

        if(head != null && head.obj == endMarker) {
            closed=true;
        }

        retval.next=null;
        return retval.obj;
    }


    void decrementSize() {
        size--;
        if(size < 0)
            size=0;
    }


    /* ---------------------------------- End of Private Methods -------------------------------- */

}