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.
#	User	Rev	Content
1	tdb	1.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	tdb	1.4	* 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	tdb	1.1	*
18	tdb	1.2	* @author $Author: tdb1 $
19	tdb	1.14	* @version $Id: PacketSorter.java,v 1.13 2001/03/10 04:03:04 tdb1 Exp $
20	tdb	1.1	*/
21			class PacketSorter extends Thread {
22
23			//---FINAL ATTRIBUTES---
24
25			/**
26			* The current CVS revision of this class
27			*/
28	tdb	1.14	public final String REVISION = "$Revision: 1.13 $";
29	tdb	1.1
30			//---STATIC METHODS---
31
32			//---CONSTRUCTORS---
33
34			/**
35			* Creates a new PacketSorter.
36	tdb	1.6	*
37			* @param queueMonitorInterval The interval at which to monitor the Queue
38	tdb	1.1	*/
39	tdb	1.6	public PacketSorter(int queueMonitorInterval) {
40	tdb	1.14	// set the Thread name
41			setName("clientinterface.PacketSorter");
42
43	tdb	1.1	_queue = new Queue();
44	tdb	1.5	// startup a monitor on this queue, every minute
45	tdb	1.6	String queueName = _name + " PacketSorterQueue";
46			_queue.startMonitor(queueMonitorInterval*1000, queueName);
47	tdb	1.11	_hostMap = new HashMap();
48			_allHostDataList = new LinkedList();
49			_allHostsList = new LinkedList();
50	tdb	1.1	_logger.write(toString(), Logger.SYSINIT, "created");
51			}
52
53			//---PUBLIC METHODS---
54
55	tdb	1.4	/**
56			* Method to start the PacketSorter running. This method will
57			* loop forever processing and sending data.
58			*/
59	tdb	1.1	public void run() {
60	tdb	1.13	XMLPacketMaker xmlPacketMaker = new XMLPacketMaker();
61	tdb	1.1	int qID = _queue.getQueue();
62			while(true) {
63	tdb	1.4	// attempt to get some data from the Queue
64	tdb	1.1	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	tdb	1.10	XMLPacket packet = null;
73
74			try {
75	tdb	1.13	packet = xmlPacketMaker.createXMLPacket(xml);
76	tdb	1.10	} catch(InvalidXMLException e) {
77			_logger.write(toString(), Logger.ERROR, "Invalid XML: "+e);
78			// skip the rest of this loop iteration
79			continue;
80			}
81	tdb	1.1
82	tdb	1.8	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	tdb	1.11	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	tdb	1.8	}
97			}
98
99			// any handler in this list wants all packets, so send
100			// it on to them regardless
101	tdb	1.11	synchronized(this) {
102			Iterator i = _allHostDataList.iterator();
103			while(i.hasNext()) {
104			((Queue) i.next()).add(xml);
105			}
106			}
107	tdb	1.1	}
108	tdb	1.8	else {
109	tdb	1.10	// always send this packet to all hosts, because it's
110			// "extra" data, not host data
111	tdb	1.11	synchronized(this) {
112			Iterator i = _allHostsList.iterator();
113			while(i.hasNext()) {
114			((Queue) i.next()).add(xml);
115			}
116			}
117	tdb	1.2	}
118	tdb	1.1	}
119			}
120
121	tdb	1.4	/**
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	tdb	1.12	public void register(Queue dhQueue, String hostList) {
135	tdb	1.4	// check to see if we want all hosts
136	tdb	1.2	if(hostList.equals("")) {
137	tdb	1.12	synchronized(this) {
138			_allHostDataList.add(dhQueue);
139			}
140	tdb	1.2	_logger.write(toString(), Logger.SYSMSG, "registered DataHandler for all hosts");
141			}
142			else {
143	tdb	1.4	// go through the list of hosts
144	tdb	1.2	StringTokenizer st = new StringTokenizer(hostList, ";");
145			while(st.hasMoreTokens()) {
146			String host = st.nextToken();
147	tdb	1.12	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	tdb	1.2	}
161	tdb	1.1	}
162	tdb	1.2	_logger.write(toString(), Logger.SYSMSG, "registered DataHandler for hosts: "+hostList);
163	tdb	1.1	}
164	tdb	1.10	// always add host to our complete host list
165	tdb	1.12	synchronized(this) {
166			_allHostsList.add(dhQueue);
167			}
168	tdb	1.1	}
169
170	tdb	1.4	/**
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	tdb	1.12	public void deregister(Queue dhQueue, String hostList) {
182	tdb	1.4	// go through the list of hosts
183	tdb	1.2	if(hostList.equals("")) {
184	tdb	1.12	synchronized(this) {
185			_allHostDataList.remove(dhQueue);
186			}
187	tdb	1.2	_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	tdb	1.12	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	tdb	1.2	}
204	tdb	1.1	}
205			}
206	tdb	1.2	_logger.write(toString(), Logger.SYSMSG, "deregistered DataHandler for hosts: "+hostList);
207	tdb	1.1	}
208	tdb	1.10	// always remove host from our complete host list
209	tdb	1.12	synchronized(this) {
210			_allHostsList.remove(dhQueue);
211			}
212	tdb	1.1	}
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	tdb	1.4
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	tdb	1.1	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	tdb	1.4	/**
265			* A reference to the Queue we're using.
266			*/
267	tdb	1.1	private Queue _queue;
268
269	tdb	1.4	/**
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	tdb	1.11	private HashMap _hostMap;
274	tdb	1.4
275			/**
276			* A list specifically for a Queue's associated with DataHandlers
277			* that want all host information.
278	tdb	1.10	*/
279	tdb	1.11	private LinkedList _allHostDataList;
280	tdb	1.10
281			/**
282			* A list of all hosts.
283	tdb	1.4	*/
284	tdb	1.11	private LinkedList _allHostsList;
285	tdb	1.1
286			//---STATIC ATTRIBUTES---
287
288			}