server/clientinterface/PacketSorter.java

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

//---IMPORTS---
import uk.ac.ukc.iscream.util.*;
import uk.ac.ukc.iscream.componentmanager.*;
import uk.ac.ukc.iscream.core.*;
import java.util.*;

/**
 * Receives data from the incoming CORBA servant, places
 * it in a Queue, and then arranges distribution to the
 * DataHandlers.
 * Has extra functionality to send data to DataHandlers
 * on a per-host basis - ie. the Client can request which
 * hosts it would like to listen for.
 *
 * @author  $Author: tdb1 $
 * @version $Id: PacketSorter.java,v 1.13 2001/03/10 04:03:04 tdb1 Exp $
 */
class PacketSorter extends Thread {

//---FINAL ATTRIBUTES---

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

//---CONSTRUCTORS---

    /**
     * Creates a new PacketSorter.
     *
     * @param queueMonitorInterval The interval at which to monitor the Queue
     */
    public PacketSorter(int queueMonitorInterval) {
        // set the Thread name
        setName("clientinterface.PacketSorter");
        
        _queue = new Queue();
        // startup a monitor on this queue, every minute
        String queueName = _name + " PacketSorterQueue";
        _queue.startMonitor(queueMonitorInterval*1000, queueName);
        _hostMap = new HashMap();
        _allHostDataList = new LinkedList();
        _allHostsList = new LinkedList();
        _logger.write(toString(), Logger.SYSINIT, "created");
    }
    
//---PUBLIC METHODS---

    /**
     * Method to start the PacketSorter running. This method will
     * loop forever processing and sending data.
     */
    public void run() {
        XMLPacketMaker xmlPacketMaker = new XMLPacketMaker();
        int qID = _queue.getQueue();
        while(true) {
            // attempt to get some data from the Queue
            String xml = "";
            try {
                xml = (String) _queue.get(qID);
            }
            catch(InvalidQueueException e) {
                _logger.write(toString(), Logger.ERROR, "Queue failure: "+e);
            }
            
            XMLPacket packet = null;
            
            try {
                packet = xmlPacketMaker.createXMLPacket(xml);
            } catch(InvalidXMLException e) {
                _logger.write(toString(), Logger.ERROR, "Invalid XML: "+e);
                // skip the rest of this loop iteration
                continue;
            }
            
            String packetType = packet.getParam("packet.attributes.type");
            // check if we need to send it regardless
            if(packetType.equals("data") || packetType.equals("heartbeat")) {
                String host = packet.getParam("packet.attributes.machine_name");
                
                // look in the hostMap to see if anyone wants this data
                synchronized(this) {
                    if(_hostMap.containsKey(host)) {
                        LinkedList list = (LinkedList) _hostMap.get(host);
                        Iterator i = list.iterator();
                        // push the data to the listening Handler's queue
                        while(i.hasNext()) {
                            ((Queue) i.next()).add(xml);
                        }
                    }
                }
                
                // any handler in this list wants all packets, so send
                // it on to them regardless
                synchronized(this) {
                    Iterator i = _allHostDataList.iterator();
                    while(i.hasNext()) {
                        ((Queue) i.next()).add(xml);
                    }
                }
            }
            else {
                // always send this packet to all hosts, because it's
                // "extra" data, not host data
                synchronized(this) {
                    Iterator i = _allHostsList.iterator();
                    while(i.hasNext()) {
                        ((Queue) i.next()).add(xml);
                    }
                }
            }
        }
    }
    
    /**
     * Register a DataHandler in the system. This method
     * actually takes a reference to a Queue, which should be
     * a Queue that the DataHandler is making use of.
     * It also takes a hostList, this being a semi-colon
     * seperated list of hosts that the Client the DataHandler
     * is serving has requested. If this list is simply an empty
     * String, it is assumed the Client wants to listen to all
     * host information.
     *
     * @param dhQueue a Queue being used by the DataHandler that is registering
     * @param hostList a semi-colon seperated list of hosts
     */
    public void register(Queue dhQueue, String hostList) {
        // check to see if we want all hosts
        if(hostList.equals("")) {
            synchronized(this) {
                _allHostDataList.add(dhQueue);
            }
            _logger.write(toString(), Logger.SYSMSG, "registered DataHandler for all hosts");
        }
        else {
            // go through the list of hosts
            StringTokenizer st = new StringTokenizer(hostList, ";");
            while(st.hasMoreTokens()) {
                String host = st.nextToken();
                synchronized(this) {
                    // see if we already have a list in the map for this host
                    if(_hostMap.containsKey(host)) {
                        // we do, so add to it
                        List list = (List) _hostMap.get(host);
                        list.add(dhQueue);
                    }
                    else {
                        // we don't, so create a list and put it in the map
                        LinkedList list = new LinkedList();
                        list.add(dhQueue);
                        _hostMap.put(host, list);
                    }
                }
            }
            _logger.write(toString(), Logger.SYSMSG, "registered DataHandler for hosts: "+hostList);
        }
        // always add host to our complete host list
        synchronized(this) {
            _allHostsList.add(dhQueue);
        }
    }
    
    /**
     * Deregister a DataHandler. The DataHandler should give a reference
     * to the Queue it's using, and the *same* hostList it gave when it
     * register. It is imperative that the hostList is the same, otherwise
     * there will be all sorts of problems with lists getting out of sync.
     *
     * NB: Possible future addition would be recording of hostList's.
     *
     * @param dhQueue a Queue being used by the DataHandler that is deregistering
     * @param hostList a semi-colon seperated list of hosts
     */
    public void deregister(Queue dhQueue, String hostList) {
        // go through the list of hosts
        if(hostList.equals("")) {
            synchronized(this) {
                _allHostDataList.remove(dhQueue);
            }
            _logger.write(toString(), Logger.SYSMSG, "deregistered DataHandler for all hosts");
        }
        else {
            StringTokenizer st = new StringTokenizer(hostList, ";");
            while(st.hasMoreTokens()) {
                String host = st.nextToken();
                synchronized(this) {
                    // this should in reality always be true, but best check
                    if(_hostMap.containsKey(host)) {
                        // get the list and remove the host in question
                        LinkedList list = (LinkedList) _hostMap.get(host);
                        list.remove(dhQueue);
                        // if the list is now empty, we might as well remove it
                        if(list.size()==0) {
                            _hostMap.remove(host);
                        }
                    }
                }
            }
            _logger.write(toString(), Logger.SYSMSG, "deregistered DataHandler for hosts: "+hostList);
        }
        // always remove host from our complete host list
        synchronized(this) {
            _allHostsList.remove(dhQueue);
        }
    }
    
    /**
     * 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.NameFormat 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---
    
    /**
     * Accessor to return a reference to the Queue object. This
     * is needed so the ClientInterfaceServant can get add data
     * easily.
     *
     * @return a reference to our Queue object.
     */
    public Queue getQueue() {
        return _queue;
    }

//---ATTRIBUTES---

    /**
     * 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 = ClientInterfaceMain.NAME;

    /**
     * This holds a reference to the
     * system logger that is being used.
     */
    private Logger _logger = ReferenceManager.getInstance().getLogger();
    
    /**
     * A reference to the Queue we're using.
     */
    private Queue _queue;
    
    /**
     * A HashMap to store lists of Queue's (in the DataHandlers)
     * in a way that can be easily accessed when data comes in.
     */
    private HashMap _hostMap;
    
    /**
     * A list specifically for a Queue's associated with DataHandlers
     * that want all host information.
     */
    private LinkedList _allHostDataList;
    
    /**
     * A list of all hosts.
     */
    private LinkedList _allHostsList;
    
//---STATIC ATTRIBUTES---

}
Revision:	1.14
Committed:	Tue Mar 13 02:19:44 2001 UTC (23 years, 2 months ago) by tdb
Branch:	MAIN
Changes since 1.13:	+5 -2 lines
Log Message:	Given all the classes that extend Thread a name using Thread.setName(). It is only representative as far as it will tell us which class the Thread is, but this will go some way to aiding debugging. If time permitted, more effort could be taken to name each thread according to what it was dealing with.
#	Content
1	//---PACKAGE DECLARATION---
2	package uk.ac.ukc.iscream.clientinterface;
3
4	//---IMPORTS---
5	import uk.ac.ukc.iscream.util.*;
6	import uk.ac.ukc.iscream.componentmanager.*;
7	import uk.ac.ukc.iscream.core.*;
8	import java.util.*;
9
10	/**
11	* Receives data from the incoming CORBA servant, places
12	* it in a Queue, and then arranges distribution to the
13	* DataHandlers.
14	* Has extra functionality to send data to DataHandlers
15	* on a per-host basis - ie. the Client can request which
16	* hosts it would like to listen for.
17	*
18	* @author $Author: tdb1 $
19	* @version $Id: PacketSorter.java,v 1.13 2001/03/10 04:03:04 tdb1 Exp $
20	*/
21	class PacketSorter extends Thread {
22
23	//---FINAL ATTRIBUTES---
24
25	/**
26	* The current CVS revision of this class
27	*/
28	public final String REVISION = "$Revision: 1.13 $";
29
30	//---STATIC METHODS---
31
32	//---CONSTRUCTORS---
33
34	/**
35	* Creates a new PacketSorter.
36	*
37	* @param queueMonitorInterval The interval at which to monitor the Queue
38	*/
39	public PacketSorter(int queueMonitorInterval) {
40	// set the Thread name
41	setName("clientinterface.PacketSorter");
42
43	_queue = new Queue();
44	// startup a monitor on this queue, every minute
45	String queueName = _name + " PacketSorterQueue";
46	_queue.startMonitor(queueMonitorInterval*1000, queueName);
47	_hostMap = new HashMap();
48	_allHostDataList = new LinkedList();
49	_allHostsList = new LinkedList();
50	_logger.write(toString(), Logger.SYSINIT, "created");
51	}
52
53	//---PUBLIC METHODS---
54
55	/**
56	* Method to start the PacketSorter running. This method will
57	* loop forever processing and sending data.
58	*/
59	public void run() {
60	XMLPacketMaker xmlPacketMaker = new XMLPacketMaker();
61	int qID = _queue.getQueue();
62	while(true) {
63	// attempt to get some data from the Queue
64	String xml = "";
65	try {
66	xml = (String) _queue.get(qID);
67	}
68	catch(InvalidQueueException e) {
69	_logger.write(toString(), Logger.ERROR, "Queue failure: "+e);
70	}
71
72	XMLPacket packet = null;
73
74	try {
75	packet = xmlPacketMaker.createXMLPacket(xml);
76	} catch(InvalidXMLException e) {
77	_logger.write(toString(), Logger.ERROR, "Invalid XML: "+e);
78	// skip the rest of this loop iteration
79	continue;
80	}
81
82	String packetType = packet.getParam("packet.attributes.type");
83	// check if we need to send it regardless
84	if(packetType.equals("data") \|\| packetType.equals("heartbeat")) {
85	String host = packet.getParam("packet.attributes.machine_name");
86
87	// look in the hostMap to see if anyone wants this data
88	synchronized(this) {
89	if(_hostMap.containsKey(host)) {
90	LinkedList list = (LinkedList) _hostMap.get(host);
91	Iterator i = list.iterator();
92	// push the data to the listening Handler's queue
93	while(i.hasNext()) {
94	((Queue) i.next()).add(xml);
95	}
96	}
97	}
98
99	// any handler in this list wants all packets, so send
100	// it on to them regardless
101	synchronized(this) {
102	Iterator i = _allHostDataList.iterator();
103	while(i.hasNext()) {
104	((Queue) i.next()).add(xml);
105	}
106	}
107	}
108	else {
109	// always send this packet to all hosts, because it's
110	// "extra" data, not host data
111	synchronized(this) {
112	Iterator i = _allHostsList.iterator();
113	while(i.hasNext()) {
114	((Queue) i.next()).add(xml);
115	}
116	}
117	}
118	}
119	}
120
121	/**
122	* Register a DataHandler in the system. This method
123	* actually takes a reference to a Queue, which should be
124	* a Queue that the DataHandler is making use of.
125	* It also takes a hostList, this being a semi-colon
126	* seperated list of hosts that the Client the DataHandler
127	* is serving has requested. If this list is simply an empty
128	* String, it is assumed the Client wants to listen to all
129	* host information.
130	*
131	* @param dhQueue a Queue being used by the DataHandler that is registering
132	* @param hostList a semi-colon seperated list of hosts
133	*/
134	public void register(Queue dhQueue, String hostList) {
135	// check to see if we want all hosts
136	if(hostList.equals("")) {
137	synchronized(this) {
138	_allHostDataList.add(dhQueue);
139	}
140	_logger.write(toString(), Logger.SYSMSG, "registered DataHandler for all hosts");
141	}
142	else {
143	// go through the list of hosts
144	StringTokenizer st = new StringTokenizer(hostList, ";");
145	while(st.hasMoreTokens()) {
146	String host = st.nextToken();
147	synchronized(this) {
148	// see if we already have a list in the map for this host
149	if(_hostMap.containsKey(host)) {
150	// we do, so add to it
151	List list = (List) _hostMap.get(host);
152	list.add(dhQueue);
153	}
154	else {
155	// we don't, so create a list and put it in the map
156	LinkedList list = new LinkedList();
157	list.add(dhQueue);
158	_hostMap.put(host, list);
159	}
160	}
161	}
162	_logger.write(toString(), Logger.SYSMSG, "registered DataHandler for hosts: "+hostList);
163	}
164	// always add host to our complete host list
165	synchronized(this) {
166	_allHostsList.add(dhQueue);
167	}
168	}
169
170	/**
171	* Deregister a DataHandler. The DataHandler should give a reference
172	* to the Queue it's using, and the same hostList it gave when it
173	* register. It is imperative that the hostList is the same, otherwise
174	* there will be all sorts of problems with lists getting out of sync.
175	*
176	* NB: Possible future addition would be recording of hostList's.
177	*
178	* @param dhQueue a Queue being used by the DataHandler that is deregistering
179	* @param hostList a semi-colon seperated list of hosts
180	*/
181	public void deregister(Queue dhQueue, String hostList) {
182	// go through the list of hosts
183	if(hostList.equals("")) {
184	synchronized(this) {
185	_allHostDataList.remove(dhQueue);
186	}
187	_logger.write(toString(), Logger.SYSMSG, "deregistered DataHandler for all hosts");
188	}
189	else {
190	StringTokenizer st = new StringTokenizer(hostList, ";");
191	while(st.hasMoreTokens()) {
192	String host = st.nextToken();
193	synchronized(this) {
194	// this should in reality always be true, but best check
195	if(_hostMap.containsKey(host)) {
196	// get the list and remove the host in question
197	LinkedList list = (LinkedList) _hostMap.get(host);
198	list.remove(dhQueue);
199	// if the list is now empty, we might as well remove it
200	if(list.size()==0) {
201	_hostMap.remove(host);
202	}
203	}
204	}
205	}
206	_logger.write(toString(), Logger.SYSMSG, "deregistered DataHandler for hosts: "+hostList);
207	}
208	// always remove host from our complete host list
209	synchronized(this) {
210	_allHostsList.remove(dhQueue);
211	}
212	}
213
214	/**
215	* Overrides the {@link java.lang.Object#toString() Object.toString()}
216	* method to provide clean logging (every class should have this).
217	*
218	* This uses the uk.ac.ukc.iscream.util.NameFormat class
219	* to format the toString()
220	*
221	* @return the name of this class and its CVS revision
222	*/
223	public String toString() {
224	return FormatName.getName(
225	_name,
226	getClass().getName(),
227	REVISION);
228	}
229
230	//---PRIVATE METHODS---
231
232	//---ACCESSOR/MUTATOR METHODS---
233
234	/**
235	* Accessor to return a reference to the Queue object. This
236	* is needed so the ClientInterfaceServant can get add data
237	* easily.
238	*
239	* @return a reference to our Queue object.
240	*/
241	public Queue getQueue() {
242	return _queue;
243	}
244
245	//---ATTRIBUTES---
246
247	/**
248	* This is the friendly identifier of the
249	* component this class is running in.
250	* eg, a Filter may be called "filter1",
251	* If this class does not have an owning
252	* component, a name from the configuration
253	* can be placed here. This name could also
254	* be changed to null for utility classes.
255	*/
256	private String _name = ClientInterfaceMain.NAME;
257
258	/**
259	* This holds a reference to the
260	* system logger that is being used.
261	*/
262	private Logger _logger = ReferenceManager.getInstance().getLogger();
263
264	/**
265	* A reference to the Queue we're using.
266	*/
267	private Queue _queue;
268
269	/**
270	* A HashMap to store lists of Queue's (in the DataHandlers)
271	* in a way that can be easily accessed when data comes in.
272	*/
273	private HashMap _hostMap;
274
275	/**
276	* A list specifically for a Queue's associated with DataHandlers
277	* that want all host information.
278	*/
279	private LinkedList _allHostDataList;
280
281	/**
282	* A list of all hosts.
283	*/
284	private LinkedList _allHostsList;
285
286	//---STATIC ATTRIBUTES---
287
288	}