ThreadSafeQueue.java
/*
** Module : ThreadSafeQueue.java
** Abstract : a multi-thread safe object queue
**
** Copyright (c) 2005-2017, Golden Code Development Corporation.
**
** -#- -I- --Date-- --JPRM-- ------------------Description-------------------
** 001 SIY 20050113 @19510 Created initial implementation of the queue.
** LinkedList is used as an internal storage.
** 002 SIY 20050339 @20557 Added capability to terminate queue.
** 003 SIY 20051001 @22919 Added counter for waiting threads. Fixed
** formatting.
** 004 GES 20060501 @25883 Added synchronization for accessing the
** count, made the dequeuing loop until an
** element is available or termination occurs
** and cleaned up formatting and javadoc
** problems. The dequeuing change was needed
** because it was possible to be interrupted
** and attempt to get an element from the list
** while the list was empty.
** 005 ECF 20071106 @35895 Implemented as a generic class.
** 006 GES 20090816 @43641 Added isTerminated() method.
** 007 IAS 20160805 @43641 Replaced List with BlockingQueue
*/
/*
** This program is free software: you can redistribute it and/or modify
** it under the terms of the GNU Affero General Public License as
** published by the Free Software Foundation, either version 3 of the
** License, or (at your option) any later version.
**
** This program is distributed in the hope that it will be useful,
** but WITHOUT ANY WARRANTY; without even the implied warranty of
** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
** GNU Affero General Public License for more details.
**
** You may find a copy of the GNU Affero GPL version 3 at the following
** location: https://www.gnu.org/licenses/agpl-3.0.en.html
**
** Additional terms under GNU Affero GPL version 3 section 7:
**
** Under Section 7 of the GNU Affero GPL version 3, the following additional
** terms apply to the works covered under the License. These additional terms
** are non-permissive additional terms allowed under Section 7 of the GNU
** Affero GPL version 3 and may not be removed by you.
**
** 0. Attribution Requirement.
**
** You must preserve all legal notices or author attributions in the covered
** work or Appropriate Legal Notices displayed by works containing the covered
** work. You may not remove from the covered work any author or developer
** credit already included within the covered work.
**
** 1. No License To Use Trademarks.
**
** This license does not grant any license or rights to use the trademarks
** Golden Code, FWD, any Golden Code or FWD logo, or any other trademarks
** of Golden Code Development Corporation. You are not authorized to use the
** name Golden Code, FWD, or the names of any author or contributor, for
** publicity purposes without written authorization.
**
** 2. No Misrepresentation of Affiliation.
**
** You may not represent yourself as Golden Code Development Corporation or FWD.
**
** You may not represent yourself for publicity purposes as associated with
** Golden Code Development Corporation, FWD, or any author or contributor to
** the covered work, without written authorization.
**
** 3. No Misrepresentation of Source or Origin.
**
** You may not represent the covered work as solely your work. All modified
** versions of the covered work must be marked in a reasonable way to make it
** clear that the modified work is not originating from Golden Code Development
** Corporation or FWD. All modified versions must contain the notices of
** attribution required in this license.
*/
package com.goldencode.p2j.util;
import java.util.*;
import java.util.concurrent.*;
/**
* This class implements a thread safe queue of the objects. Any number of
* threads can access the queue to store/retrieve objects to/from the queue.
* If the queue is empty then the thread which tries to retrieve an object
* from the queue is suspended until another thread stores an object into
* the queue.
*
* @param <E>
* Type of element object to be queued.
*/
public class ThreadSafeQueue<E>
{
/** Termination token. */
private static final Poison POISON = new Poison();
/** Internal storage for the objects. */
private final BlockingQueue<Object> queue = new LinkedBlockingQueue<>();
/** Flag indicating if all access to the queue should be ended. */
private volatile boolean terminated = false;
/**
* Construct an empty queue.
*/
public ThreadSafeQueue()
{
}
/**
* Put an object into the queue and notify one waiting thread.
*
* @param obj
* An object to place into queue.
*
* @throws InterruptedException
* When the queue is terminated.
*/
public void enqueue(E obj)
throws InterruptedException
{
if (terminated)
{
throw new InterruptedException("The queue has been terminated.");
}
queue.add(obj);
}
/**
* Get the next object from the queue. Wait indefinitely if the queue is
* empty.
*
* @return The next object from the queue.
*
* @throws InterruptedException
* When the queue is terminated or if interrupted while waiting.
*/
public E dequeue()
throws InterruptedException
{
// Retrieve the next object when it will be available
// At this point the InterruptedException can be thrown
Object e = queue.take();
// re-insert the poison object so other waiters are unblocked
// this may add a small overhead with terminated queue but
// checking the 'terminated' flag before 'take'
// adds (smaller) overhead to each dequeue operation.
if (e == POISON)
{
queue.add(POISON);
// As the POISON was retrieved the 'terminated' flag is already set
// and the InterruptedException will be thrown below
}
if (terminated)
{
throw new InterruptedException("The queue has been terminated.");
}
// this cast is in fact safe as the only way for put the non-E object is via terminate
return (E)e;
}
/**
* Terminate queue processing and release all waiting threads. All
* subsequent calls to this instance will generate an exception.
*/
public void terminate()
{
terminated = true;
queue.add(POISON);
}
/**
* Report if the queue processing has been terminated.
*
* @return <code>true</code> if the queue can no longer be used.
*/
public boolean isTerminated()
{
return terminated;
}
/**
* Class for the termination token
*/
private static class Poison
{
}
}