queue.java

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

            retval=removeInternal();
            /*null result means we timed out*/
            if(retval == null) throw new TimeoutException("timeout=" + timeout + "ms");

            /*if we reached an end marker we are going to close the queue*/
//            if(retval == endMarker) {
//                close(false);
//                throw new QueueClosedException();
//            }
            /*at this point we actually did receive a value from the queue, return it*/
            return retval;
        }
    }


    /**
     * removes a specific object from the queue.
     * the object is matched up using the Object.equals method.
     * @param   obj the actual object to be removed from the queue
     */
    public void removeElement(Object obj) throws QueueClosedException {
        Element el, tmp_el;

        if(obj == null) {
            if(log.isErrorEnabled()) log.error("argument must not be null");
            return;
        }

        synchronized(mutex) {
            if(closed) /*check to see if the queue is closed*/
                throw new QueueClosedException();

            el=head;

            /*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;
                el.obj=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();
                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.obj=null;
                    el.next=el.next.next;  // point to the el past the next one. can be null.
                    tmp_el.next=null;
                    tmp_el.obj=null;
                    decrementSize();
                    break;
                }
                el=el.next;
            }
        }
    }


    /**
     * 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;

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

            if(closed)
                throw new QueueClosedException();

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

        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;

        synchronized(mutex) {
            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("timeout=" + timeout + "ms");

            if(retval == endMarker) {
                close(false);
                throw new QueueClosedException();
            }
            return retval;
        }
    }

    /** Removes all elements from the queue. This method can succeed even when the queue is closed */
    public void clear() {
        synchronized(mutex) {
            head=tail=null;
            size=0;
            num_markers=0;
            mutex.notifyAll();
        }
    }


    /**
     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) {
        synchronized(mutex) {
            if(flush_entries && size > 0) {
                try {
                    add(endMarker); // add an end-of-entries marker to the end of the queue
                    num_markers++;
                }
                catch(QueueClosedException closed_ex) {
                }
                return;
            }
            closed=true;
            mutex.notifyAll();
        }
    }

    /** Waits until the queue has been closed. Returns immediately if already closed
     * @param timeout Number of milliseconds to wait. A value <= 0 means to wait forever
     */
    public void waitUntilClosed(long timeout) {
        synchronized(mutex) {
            if(closed)
                return;
            try {
                mutex.wait(timeout);
            }
            catch(InterruptedException e) {
            }
        }
    }


    /**
     * resets the queue.
     * This operation removes all the objects in the queue and marks the queue open
     */
    public void reset() {
        synchronized(mutex) {
           num_markers=0;
           if(!closed)
              close(false);
            size=0;
            head=null;
            tail=null;
            closed=false;
            mutex.notifyAll();
        }
    }

    /**
     * Returns all the elements of the queue
     * @return A copy of the queue
     */
    public LinkedList values() {
        LinkedList retval=new LinkedList();
        synchronized(mutex) {
            Element el=head;
            while(el != null) {
                retval.add(el.obj);
                el=el.next;
            }
        }
        return retval;
    }


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

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




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


    private final void addInternal(Object obj) {
        /*create a new linked list element*/
        Element el=new Element(obj);
        /*check the first element*/
        if(head == null) {
            /*the object added is the first element*/
            /*set the head to be this object*/
            head=el;
            /*set the tail to be this object*/
            tail=head;
            /*set the size to be one, since the queue was empty*/
            size=1;
        }
        else {
            /*add the object to the end of the linked list*/
            tail.next=el;
            /*set the tail to point to the last element*/
            tail=el;
            /*increase the size*/
            size++;
        }
    }

    /**
     * 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;
        Object obj;

        /*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();
        if(head != null && head.obj == endMarker) {
            closed=true;
            mutex.notifyAll();
        }

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


    /** Doesn't need to be synchronized; is always called from synchronized methods */
    final private void decrementSize() {
        size--;
        if(size < 0)
            size=0;
    }


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

}