server/Queue/Queue.java

//---PACKAGE DECLARATION---
//package uk.ac.ukc.iscream.util;

//---IMPORTS---
import java.util.LinkedList;
import java.util.NoSuchElementException;
//import uk.ac.ukc.iscream.util.*;

/**
 * A Queue class designed to operate in a multi-threaded environment, with
 * added support for multiple "consumer" threads. Also offers blocking on
 * the get() methods, which ensures the consumer waits until the queue
 * actually contains some elements.
 *
 * @author  $Author: tdb1 $
 * @version $Id: Queue.java,v 1.9 2001/02/12 02:21:45 tdb1 Exp $
 */
public class Queue {

//---FINAL ATTRIBUTES---

    /**
     * The current CVS revision of this class
     */
    public static final String REVISION = "$Revision: 1.9 $";
    
//---STATIC METHODS---

//---CONSTRUCTORS---  

//---PUBLIC METHODS---
    
    /**
     * This method adds a given object to every queue. It will notify
     * any waiting consumers (on an empty queue) during this process.
     *
     * @param o An Object to be added to the queues.
     */
    public void add(Object o) {
        for(int i=0; i < _lists.size(); i++) {
            // skip over any gaps left in the list
            if(_lists.get(i) != null) {
                int s = ((LinkedList) _lists.get(i)).size();
                synchronized(this) {
                    // add() does the same thing, but this ensures behaviour
                    ((LinkedList) _lists.get(i)).addLast(o);
                }
                // if the queue was empty before the add it is possible
                // that a consumer is waiting... so we notify them
                if (s == 0) {
                    synchronized(((LinkedList) _lists.get(i))) {
                        ((LinkedList) _lists.get(i)).notifyAll();
                    }
                }
            }
        }
        // we keep track of the total additions for the status() method
        _count++;
    }
    
    /**
     * This method returns an object from the front of a given queue.
     * It will block until data exists in the queue if required.
     *
     * @param The queue to retrieve data from.
     * @return The object from the front of the queue.
     * @throws InvalidQueueException if the queue does not exist.
     */
    public Object get(int queue) throws InvalidQueueException {
        // make sure queue exists
        if (queue >= _lists.size() || _lists.get(queue) == null) {
            throw new InvalidQueueException("Requested queue "+queue+" does not exist");
        }
        // block if the queue is empty
        if (((LinkedList) _lists.get(queue)).size() == 0) {
            synchronized(((LinkedList) _lists.get(queue))) {
                try { ((LinkedList) _lists.get(queue)).wait(); } catch(Exception e) {}
            }
        }
        // get an item, it should never be null due to the blocking above
        Object o = null;
        synchronized(this) {
            try {
                o = ((LinkedList) _lists.get(queue)).removeFirst();
            }
            catch (NoSuchElementException e) {
                // This should never happen !
            }
        }
        return o;
    }
    
    /**
     * This method releases a get() method that's currently
     * waiting on an empty queue. This was designed for
     * shutdown() type methods that may have problems closing
     * if the thread of control is waiting on a queue.
     *
     * @param queue the queue to release.
     */
    public void releaseQueue(int queue) {
        synchronized(((LinkedList) _lists.get(queue))) {
                ((LinkedList) _lists.get(queue)).notifyAll();
        }
    }

    /**
     * This method erases the contents of a given queue. This
     * method should be used with care. It can only empty one
     * internal queue, not all of them. This must be called
     * multiple times to empty all queues.
     *
     * @param queue the queue to empty.
     */
    public void clearQueue(int queue) {
        synchronized(this) {
            ((LinkedList) _lists.get(queue)).clear();
        }
    }
    
    /**
     * This method returns an XML textual status of the queues.
     * It is merely for observation, and would most likely be
     * used by a larger "monitoring" component. Information
     * returned includes the current size of each queue, and
     * the total items passed through.
     *
     * @return A String message containing status information in XML format
     */
    public String xmlStatus() {
        String status = "<queue ";
        for(int i=0; i < _lists.size(); i++) {
            // check for null entries
            if(_lists.get(i) != null) {
                status += "queue"+i+"=\""+((LinkedList) _lists.get(i)).size()+"\" ";
            }
            else {
                status += "queue"+i+"=\"null\" ";
            }
        }
        status += "total=\""+_count+"\"></queue>";
        return status;
    }
    
    /**
     * Returns the size of a given queue. A consumer can use
     * this to see how big their queue is at any given time.
     * they should use their queue number as the parameter.
     *
     * @param queue The queue number to query.
     * @return the current size of the queue.
     * @throws InvalidQueueException if the queue does not exist.
     */
    public int queueSize(int queue) throws InvalidQueueException {
        if (queue >= _lists.size() || _lists.get(queue) == null) {
            throw new InvalidQueueException("Requested queue "+queue+" does not exist");
        }
        return ((LinkedList) _lists.get(queue)).size();
    }
    
    /**
     * Returns the total numer of elements to have passed
     * through this queue (ie. a counter on the add method).
     *
     * @return the element-ometer.
     */
    public int elementCount() {
        return _count;
    }
    
    /**
     * This method assigns a queue to a consumer. The idea behind
     * this is to ensure that only 1 consumer can be associated with
     * a given queue, otherwise the whole "blocking" thing fails
     * miserably. Queues are created upon requested.
     *
     * It is IMPORTANT that removeQueue() is used when the queue is
     * no longer required.
     *
     * @return An integer to be passed to the get() method.
     */
    public int getQueue() {
        int pos = -1;
        for(int i=0; i < _lists.size(); i++) {
            if(_lists.get(i) == null) {
                // found a gap, re-use it
                pos = i;
                _lists.set(i, new LinkedList());
            }
        }
        if(pos == -1) {
            //we didn't find a gap, add at end
            pos = _lists.size();
            _lists.add(pos, new LinkedList());
        }
        return pos;
    }
    
    /**
     * This method sets a entry to null in the list. This ensures
     * that it will no longer be added to after it is no longer
     * required be a consumer.
     *
     * @param queue The integer identifier for the queue, given by getQueue().
     */
    public void removeQueue(int queue) {
        _lists.set(queue, null);
    }
    
    /**
     * Start a monitor on our own Queue. This will log XML
     * statistics about our Queue to a given Queue (could be
     * the one being monitored).
     *
     * @param interval The long interval, in milliseconds, at which to take samples
     * @param destQueue The queue to monitor to
     * @param name A name to identify this Queue with
     * @return whether we succeeded
     */
    public boolean startMonitor(long interval, Queue destQueue, String name) {
        if(_queueMon == null) {
            // start a monitor
            _queueMon = new QueueMonitor(this, destQueue, interval, name);
            _queueMon.start();
            return true;
        }
        else {
            // already have a monitor running
            return false;
        }
    }
    
    /**
     * Start a monitor on our own Queue. This will log XML
     * statistics about our Queue to this Queue.
     *
     * @param interval The long interval, in milliseconds, at which to take samples
     * @param name A name to identify this Queue with
     * @return whether we succeeded
     */
    public boolean startMonitor(long interval, String name) {
        return startMonitor(interval, this, name);
    }
    
    /**
     * Stop a monitor on our Queue if we have on running.
     *
     * @return whether we succeeded
     */
    public boolean stopMonitor() {
        if(_queueMon != null) {
            // stop a monitor
            _queueMon.shutdown();
            _queueMon = null;
            return true;
        }
        else {
            // no monitor running
            return false;
        }
    }
    
    /**
     * Overrides the {@link java.lang.Object#toString() Object.toString()}
     * method to provide clean logging (every class should have this).
     *
     * This uses the uk.ac.ukc.iscream.util.FormatName class
     * to format the toString()
     *
     * @return the name of this class and its CVS revision.
     */
    public String toString() {
        return FormatName.getName(
            _name,
            getClass().getName(),
            REVISION);
    }

//---PRIVATE METHODS---

//---ACCESSOR/MUTATOR METHODS---

//---ATTRIBUTES---
    
    /**
     * The LinkedLists of queues.
     */
    private LinkedList _lists = new LinkedList();
    
    /**
     * A counter so we know how many data items have been
     * passed through, for statistical purposes.
     */
    private int _count = 0;
    
    /**
     * A reference to our QueueMonitor, if we have one.
     */
    private QueueMonitor _queueMon = null;
    
    /**
     * This is the friendly identifier of the
     * component this class is running in.
     * eg, a Filter may be called "filter1",
     * If this class does not have an owning
     * component,  a name from the configuration
     * can be placed here.  This name could also
     * be changed to null for utility classes.
     */
    private String _name = null;

//---STATIC ATTRIBUTES---

}
Revision:	1.7
Committed:	Mon Feb 12 02:27:56 2001 UTC (23 years, 9 months ago) by tdb
Branch:	MAIN
Changes since 1.6:	+73 -17 lines
Log Message:	Brought in line with changes in main tree. Now supports Queue monitoring in XML format, to another Queue (or the same Queue).
#	User	Rev	Content
1	tdb	1.2	//---PACKAGE DECLARATION---
2	tdb	1.7	//package uk.ac.ukc.iscream.util;
3	tdb	1.2
4			//---IMPORTS---
5	tdb	1.1	import java.util.LinkedList;
6			import java.util.NoSuchElementException;
7	tdb	1.7	//import uk.ac.ukc.iscream.util.*;
8	tdb	1.1
9	tdb	1.2	/**
10			* A Queue class designed to operate in a multi-threaded environment, with
11			* added support for multiple "consumer" threads. Also offers blocking on
12			* the get() methods, which ensures the consumer waits until the queue
13			* actually contains some elements.
14			*
15	tdb	1.3	* @author $Author: tdb1 $
16	tdb	1.7	* @version $Id: Queue.java,v 1.9 2001/02/12 02:21:45 tdb1 Exp $
17	tdb	1.2	*/
18	tdb	1.4	public class Queue {
19	tdb	1.2
20			//---FINAL ATTRIBUTES---
21
22			/**
23			* The current CVS revision of this class
24			*/
25	tdb	1.7	public static final String REVISION = "$Revision: 1.9 $";
26	tdb	1.2
27			//---STATIC METHODS---
28
29	tdb	1.3	//---CONSTRUCTORS---
30
31	tdb	1.2	//---PUBLIC METHODS---
32
33			/**
34			* This method adds a given object to every queue. It will notify
35			* any waiting consumers (on an empty queue) during this process.
36			*
37			* @param o An Object to be added to the queues.
38			*/
39			public void add(Object o) {
40	tdb	1.3	for(int i=0; i < _lists.size(); i++) {
41			// skip over any gaps left in the list
42			if(_lists.get(i) != null) {
43			int s = ((LinkedList) _lists.get(i)).size();
44			synchronized(this) {
45			// add() does the same thing, but this ensures behaviour
46			((LinkedList) _lists.get(i)).addLast(o);
47			}
48			// if the queue was empty before the add it is possible
49			// that a consumer is waiting... so we notify them
50			if (s == 0) {
51			synchronized(((LinkedList) _lists.get(i))) {
52			((LinkedList) _lists.get(i)).notifyAll();
53			}
54	tdb	1.2	}
55			}
56	tdb	1.1	}
57	tdb	1.2	// we keep track of the total additions for the status() method
58	tdb	1.1	_count++;
59			}
60
61	tdb	1.2	/**
62			* This method returns an object from the front of a given queue.
63			* It will block until data exists in the queue if required.
64			*
65	tdb	1.6	* @param The queue to retrieve data from.
66	tdb	1.2	* @return The object from the front of the queue.
67	tdb	1.3	* @throws InvalidQueueException if the queue does not exist.
68	tdb	1.2	*/
69	tdb	1.3	public Object get(int queue) throws InvalidQueueException {
70			// make sure queue exists
71			if (queue >= _lists.size() \|\| _lists.get(queue) == null) {
72			throw new InvalidQueueException("Requested queue "+queue+" does not exist");
73			}
74	tdb	1.2	// block if the queue is empty
75	tdb	1.3	if (((LinkedList) _lists.get(queue)).size() == 0) {
76			synchronized(((LinkedList) _lists.get(queue))) {
77			try { ((LinkedList) _lists.get(queue)).wait(); } catch(Exception e) {}
78	tdb	1.2	}
79	tdb	1.1	}
80	tdb	1.2	// get an item, it should never be null due to the blocking above
81	tdb	1.1	Object o = null;
82	tdb	1.2	synchronized(this) {
83			try {
84	tdb	1.3	o = ((LinkedList) _lists.get(queue)).removeFirst();
85	tdb	1.2	}
86			catch (NoSuchElementException e) {
87			// This should never happen !
88			}
89	tdb	1.1	}
90			return o;
91			}
92	tdb	1.5
93			/**
94			* This method releases a get() method that's currently
95			* waiting on an empty queue. This was designed for
96			* shutdown() type methods that may have problems closing
97			* if the thread of control is waiting on a queue.
98			*
99	tdb	1.6	* @param queue the queue to release.
100	tdb	1.5	*/
101			public void releaseQueue(int queue) {
102			synchronized(((LinkedList) _lists.get(queue))) {
103			((LinkedList) _lists.get(queue)).notifyAll();
104			}
105			}
106	tdb	1.6
107			/**
108			* This method erases the contents of a given queue. This
109			* method should be used with care. It can only empty one
110			* internal queue, not all of them. This must be called
111			* multiple times to empty all queues.
112			*
113			* @param queue the queue to empty.
114			*/
115			public void clearQueue(int queue) {
116			synchronized(this) {
117			((LinkedList) _lists.get(queue)).clear();
118			}
119			}
120	tdb	1.5
121	tdb	1.2	/**
122	tdb	1.7	* This method returns an XML textual status of the queues.
123			* It is merely for observation, and would most likely be
124			* used by a larger "monitoring" component. Information
125			* returned includes the current size of each queue, and
126			* the total items passed through.
127	tdb	1.2	*
128	tdb	1.7	* @return A String message containing status information in XML format
129	tdb	1.2	*/
130	tdb	1.7	public String xmlStatus() {
131			String status = "<queue ";
132	tdb	1.3	for(int i=0; i < _lists.size(); i++) {
133			// check for null entries
134			if(_lists.get(i) != null) {
135	tdb	1.7	status += "queue"+i+"=\""+((LinkedList) _lists.get(i)).size()+"\" ";
136	tdb	1.3	}
137			else {
138	tdb	1.7	status += "queue"+i+"=\"null\" ";
139	tdb	1.3	}
140	tdb	1.2	}
141	tdb	1.7	status += "total=\""+_count+"\"></queue>";
142	tdb	1.1	return status;
143			}
144
145	tdb	1.2	/**
146	tdb	1.4	* Returns the size of a given queue. A consumer can use
147			* this to see how big their queue is at any given time.
148			* they should use their queue number as the parameter.
149			*
150			* @param queue The queue number to query.
151			* @return the current size of the queue.
152	tdb	1.6	* @throws InvalidQueueException if the queue does not exist.
153	tdb	1.4	*/
154			public int queueSize(int queue) throws InvalidQueueException {
155	tdb	1.5	if (queue >= _lists.size() \|\| _lists.get(queue) == null) {
156	tdb	1.4	throw new InvalidQueueException("Requested queue "+queue+" does not exist");
157			}
158	tdb	1.5	return ((LinkedList) _lists.get(queue)).size();
159	tdb	1.4	}
160
161			/**
162			* Returns the total numer of elements to have passed
163			* through this queue (ie. a counter on the add method).
164			*
165	tdb	1.6	* @return the element-ometer.
166	tdb	1.4	*/
167			public int elementCount() {
168	tdb	1.5	return _count;
169	tdb	1.4	}
170
171			/**
172	tdb	1.2	* This method assigns a queue to a consumer. The idea behind
173			* this is to ensure that only 1 consumer can be associated with
174			* a given queue, otherwise the whole "blocking" thing fails
175	tdb	1.3	* miserably. Queues are created upon requested.
176			*
177			* It is IMPORTANT that removeQueue() is used when the queue is
178			* no longer required.
179	tdb	1.2	*
180			* @return An integer to be passed to the get() method.
181			*/
182	tdb	1.3	public int getQueue() {
183			int pos = -1;
184			for(int i=0; i < _lists.size(); i++) {
185			if(_lists.get(i) == null) {
186			// found a gap, re-use it
187			pos = i;
188			_lists.set(i, new LinkedList());
189			}
190	tdb	1.2	}
191	tdb	1.3	if(pos == -1) {
192			//we didn't find a gap, add at end
193			pos = _lists.size();
194			_lists.add(pos, new LinkedList());
195	tdb	1.2	}
196	tdb	1.3	return pos;
197	tdb	1.2	}
198	tdb	1.3
199			/**
200			* This method sets a entry to null in the list. This ensures
201			* that it will no longer be added to after it is no longer
202			* required be a consumer.
203			*
204			* @param queue The integer identifier for the queue, given by getQueue().
205			*/
206			public void removeQueue(int queue) {
207			_lists.set(queue, null);
208			}
209
210	tdb	1.2	/**
211	tdb	1.7	* Start a monitor on our own Queue. This will log XML
212			* statistics about our Queue to a given Queue (could be
213			* the one being monitored).
214			*
215			* @param interval The long interval, in milliseconds, at which to take samples
216			* @param destQueue The queue to monitor to
217			* @param name A name to identify this Queue with
218			* @return whether we succeeded
219			*/
220			public boolean startMonitor(long interval, Queue destQueue, String name) {
221			if(_queueMon == null) {
222			// start a monitor
223			_queueMon = new QueueMonitor(this, destQueue, interval, name);
224			_queueMon.start();
225			return true;
226			}
227			else {
228			// already have a monitor running
229			return false;
230			}
231			}
232
233			/**
234			* Start a monitor on our own Queue. This will log XML
235			* statistics about our Queue to this Queue.
236			*
237			* @param interval The long interval, in milliseconds, at which to take samples
238			* @param name A name to identify this Queue with
239			* @return whether we succeeded
240			*/
241			public boolean startMonitor(long interval, String name) {
242			return startMonitor(interval, this, name);
243			}
244
245			/**
246			* Stop a monitor on our Queue if we have on running.
247			*
248			* @return whether we succeeded
249			*/
250			public boolean stopMonitor() {
251			if(_queueMon != null) {
252			// stop a monitor
253			_queueMon.shutdown();
254			_queueMon = null;
255			return true;
256			}
257			else {
258			// no monitor running
259			return false;
260			}
261			}
262
263			/**
264	tdb	1.2	* Overrides the {@link java.lang.Object#toString() Object.toString()}
265			* method to provide clean logging (every class should have this).
266			*
267			* This uses the uk.ac.ukc.iscream.util.FormatName class
268			* to format the toString()
269			*
270	tdb	1.6	* @return the name of this class and its CVS revision.
271	tdb	1.2	*/
272	tdb	1.4	public String toString() {
273	tdb	1.2	return FormatName.getName(
274			_name,
275			getClass().getName(),
276			REVISION);
277	tdb	1.4	}
278	tdb	1.2
279			//---PRIVATE METHODS---
280
281			//---ACCESSOR/MUTATOR METHODS---
282
283			//---ATTRIBUTES---
284
285			/**
286	tdb	1.3	* The LinkedLists of queues.
287	tdb	1.2	*/
288	tdb	1.3	private LinkedList _lists = new LinkedList();
289	tdb	1.2
290			/**
291			* A counter so we know how many data items have been
292			* passed through, for statistical purposes.
293			*/
294			private int _count = 0;
295	tdb	1.7
296			/**
297			* A reference to our QueueMonitor, if we have one.
298			*/
299			private QueueMonitor _queueMon = null;
300	tdb	1.2
301			/**
302			* This is the friendly identifier of the
303			* component this class is running in.
304			* eg, a Filter may be called "filter1",
305			* If this class does not have an owning
306			* component, a name from the configuration
307			* can be placed here. This name could also
308			* be changed to null for utility classes.
309			*/
310	tdb	1.4	private String _name = null;
311	tdb	1.2
312			//---STATIC ATTRIBUTES---
313
314			}