Commons Collections example source code file (BlockingBuffer.java)

This example Commons Collections source code file (BlockingBuffer.java) is included in the DevDaily.com "Java Source Code Warehouse" project. The intent of this project is to help you "Learn Java by Example" ^TM.

Java - Commons Collections tags/keywords

blockingbuffer, blockingbuffer, buffer, bufferunderflowexception, bufferunderflowexception, caused, interruptedexception, interruptedexception, io, object, printwriter, printwriter, stringwriter, synchronizedbuffer, timeout, util

The Commons Collections BlockingBuffer.java source code

/*
 *  Licensed to the Apache Software Foundation (ASF) under one or more
 *  contributor license agreements.  See the NOTICE file distributed with
 *  this work for additional information regarding copyright ownership.
 *  The ASF licenses this file to You under the Apache License, Version 2.0
 *  (the "License"); you may not use this file except in compliance with
 *  the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */
package org.apache.commons.collections.buffer;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Collection;

import org.apache.commons.collections.Buffer;
import org.apache.commons.collections.BufferUnderflowException;

/**
 * Decorates another <code>Buffer to make {@link #get()} and
 * {@link #remove()} block when the <code>Buffer is empty.
 * <p>
 * If either <code>get or remove is called on an empty
 * <code>Buffer, the calling thread waits for notification that
 * an <code>add or addAll operation has completed.
 * <p>
 * When one or more entries are added to an empty <code>Buffer,
 * all threads blocked in <code>get or remove are notified.
 * There is no guarantee that concurrent blocked <code>get or
 * <code>remove requests will be "unblocked" and receive data in the
 * order that they arrive.
 * <p>
 * This class is Serializable from Commons Collections 3.1.
 * This class contains an extra field in 3.2, however the serialization
 * specification will handle this gracefully.
 *
 * @author Stephen Colebourne
 * @author Janek Bogucki
 * @author Phil Steitz
 * @author James Carman
 * @version $Revision: 646777 $ $Date: 2008-04-10 13:33:15 +0100 (Thu, 10 Apr 2008) $
 * @since Commons Collections 3.0
 */
public class BlockingBuffer extends SynchronizedBuffer {

    /** Serialization version. */
    private static final long serialVersionUID = 1719328905017860541L;
    /** The timeout value in milliseconds. */
    private final long timeout;

    /**
     * Factory method to create a blocking buffer.
     *
     * @param buffer the buffer to decorate, must not be null
     * @return a new blocking Buffer
     * @throws IllegalArgumentException if buffer is null
     */
    public static Buffer decorate(Buffer buffer) {
        return new BlockingBuffer(buffer);
    }

    /**
     * Factory method to create a blocking buffer with a timeout value.
     *
     * @param buffer  the buffer to decorate, must not be null
     * @param timeoutMillis  the timeout value in milliseconds, zero or less for no timeout
     * @return a new blocking buffer
     * @throws IllegalArgumentException if the buffer is null
     * @since Commons Collections 3.2
     */
    public static Buffer decorate(Buffer buffer, long timeoutMillis) {
        return new BlockingBuffer(buffer, timeoutMillis);
    }

    //-----------------------------------------------------------------------    
    /**
     * Constructor that wraps (not copies).
     *
     * @param buffer the buffer to decorate, must not be null
     * @throws IllegalArgumentException if the buffer is null
     */
    protected BlockingBuffer(Buffer buffer) {
        super(buffer);
        this.timeout = 0;
    }

    /**
     * Constructor that wraps (not copies).
     *
     * @param buffer  the buffer to decorate, must not be null
     * @param timeoutMillis  the timeout value in milliseconds, zero or less for no timeout
     * @throws IllegalArgumentException if the buffer is null
     * @since Commons Collections 3.2
     */
    protected BlockingBuffer(Buffer buffer, long timeoutMillis) {
        super(buffer);
        this.timeout = (timeoutMillis < 0 ? 0 : timeoutMillis);
    }

    //-----------------------------------------------------------------------
    public boolean add(Object o) {
        synchronized (lock) {
            boolean result = collection.add(o);
            lock.notifyAll();
            return result;
        }
    }

    public boolean addAll(Collection c) {
        synchronized (lock) {
            boolean result = collection.addAll(c);
            lock.notifyAll();
            return result;
        }
    }

    /**
     * Gets the next value from the buffer, waiting until an object is
     * added if the buffer is empty. This method uses the default timeout
     * set in the constructor.
     *
     * @throws BufferUnderflowException if an interrupt is received
     */
    public Object get() {
        synchronized (lock) {
            while (collection.isEmpty()) {
                try {
                    if (timeout <= 0) {
                        lock.wait();
                    } else {
                        return get(timeout);
                    }
                } catch (InterruptedException e) {
                    PrintWriter out = new PrintWriter(new StringWriter());
                    e.printStackTrace(out);
                    throw new BufferUnderflowException("Caused by InterruptedException: " + out.toString());
                }
            }
            return getBuffer().get();
        }
    }

    /**
     * Gets the next value from the buffer, waiting until an object is
     * added for up to the specified timeout value if the buffer is empty.
     *
     * @param timeout  the timeout value in milliseconds
     * @throws BufferUnderflowException if an interrupt is received
     * @throws BufferUnderflowException if the timeout expires
     * @since Commons Collections 3.2
     */
    public Object get(final long timeout) {
        synchronized (lock) {
            final long expiration = System.currentTimeMillis() + timeout;
            long timeLeft = expiration - System.currentTimeMillis();
            while (timeLeft > 0 && collection.isEmpty()) {
                try {
                    lock.wait(timeLeft);
                    timeLeft = expiration - System.currentTimeMillis();
                } catch(InterruptedException e) {
                    PrintWriter out = new PrintWriter(new StringWriter());
                    e.printStackTrace(out);
                    throw new BufferUnderflowException("Caused by InterruptedException: " + out.toString());
                }
            }
            if (collection.isEmpty()) {
                throw new BufferUnderflowException("Timeout expired");
            }
            return getBuffer().get();
        }
    }

    /**
     * Removes the next value from the buffer, waiting until an object is
     * added if the buffer is empty. This method uses the default timeout
     * set in the constructor.
     *
     * @throws BufferUnderflowException if an interrupt is received
     */
    public Object remove() {
        synchronized (lock) {
            while (collection.isEmpty()) {
                try {
                    if (timeout <= 0) {
                        lock.wait();
                    } else {
                        return remove(timeout);
                    }
                } catch (InterruptedException e) {
                    PrintWriter out = new PrintWriter(new StringWriter());
                    e.printStackTrace(out);
                    throw new BufferUnderflowException("Caused by InterruptedException: " + out.toString());
                }
            }
            return getBuffer().remove();
        }
    }

    /**
     * Removes the next value from the buffer, waiting until an object is
     * added for up to the specified timeout value if the buffer is empty.
     *
     * @param timeout  the timeout value in milliseconds
     * @throws BufferUnderflowException if an interrupt is received
     * @throws BufferUnderflowException if the timeout expires
     * @since Commons Collections 3.2
     */
    public Object remove(final long timeout) {
        synchronized (lock) {
            final long expiration = System.currentTimeMillis() + timeout;
            long timeLeft = expiration - System.currentTimeMillis();
            while (timeLeft > 0 && collection.isEmpty()) {
                try {
                    lock.wait(timeLeft);
                    timeLeft = expiration - System.currentTimeMillis();
                } catch(InterruptedException e) {
                    PrintWriter out = new PrintWriter(new StringWriter());
                    e.printStackTrace(out);
                    throw new BufferUnderflowException("Caused by InterruptedException: " + out.toString());
                }
            }
            if (collection.isEmpty()) {
                throw new BufferUnderflowException("Timeout expired");
            }
            return getBuffer().remove();
        }
    }

}