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.6 2001/02/21 19:11:28 tdb1 Exp $
 */
class PacketSorter extends Thread {

//---FINAL ATTRIBUTES---

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

//---CONSTRUCTORS---

    /**
     * Creates a new PacketSorter.
     *
     * @param queueMonitorInterval The interval at which to monitor the Queue
     */
    public PacketSorter(int queueMonitorInterval) {
        _queue = new Queue();
        // startup a monitor on this queue, every minute
        String queueName = _name + " PacketSorterQueue";
        _queue.startMonitor(queueMonitorInterval*1000, queueName);
        _hostMap = Collections.synchronizedMap(new HashMap());
        _allHostsList = Collections.synchronizedList(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() {
        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);
            }
            
            // find out which host this packet is actually for
            XMLPacketMaker xmlPacketMaker = new XMLPacketMaker(xml);
            XMLPacket packet = xmlPacketMaker.createXMLPacket();
            String host = packet.getParam("packet.attributes.machine_name");
            
            // look in the hostMap to see if anyone wants this data
            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
            Iterator j = _allHostsList.iterator();
            while(j.hasNext()) {
                ((Queue) j.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("")) {
            _allHostsList.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();
                // see if we already have a list in the map for this host
                if(_hostMap.containsKey(host)) {
                    // we do, so add to it
                    LinkedList list = (LinkedList) _hostMap.get(host);
                    list.add(dhQueue);
                }
                else {
                    // we don't, so create a list and put it in the map
                    List list = Collections.synchronizedList(new LinkedList());
                    list.add(dhQueue);
                    _hostMap.put(host, list);
                }
            }
            _logger.write(toString(), Logger.SYSMSG, "registered DataHandler for hosts: "+hostList);
        }
    }
    
    /**
     * 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("")) {
            _allHostsList.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();
                // 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);
        }
    }
    
    /**
     * 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 Map _hostMap;
    
    /**
     * A list specifically for a Queue's associated with DataHandlers
     * that want all host information.
     */
    private List _allHostsList;
    
//---STATIC ATTRIBUTES---

}
Revision:	1.7
Committed:	Mon Feb 26 23:41:35 2001 UTC (23 years, 3 months ago) by tdb
Branch:	MAIN
Changes since 1.6:	+7 -7 lines
Log Message:	Oh, some nasty concurrency stuff going on in here. If I've understood the API correctly what I've done should make the Lists's and Map's thread safe, without worrying about synchronize() bits everywhere.
#	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.7	* @version $Id: PacketSorter.java,v 1.6 2001/02/21 19:11:28 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.7	public final String REVISION = "$Revision: 1.6 $";
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.1	_queue = new Queue();
41	tdb	1.5	// startup a monitor on this queue, every minute
42	tdb	1.6	String queueName = _name + " PacketSorterQueue";
43			_queue.startMonitor(queueMonitorInterval*1000, queueName);
44	tdb	1.7	_hostMap = Collections.synchronizedMap(new HashMap());
45			_allHostsList = Collections.synchronizedList(new LinkedList());
46	tdb	1.1	_logger.write(toString(), Logger.SYSINIT, "created");
47			}
48
49			//---PUBLIC METHODS---
50
51	tdb	1.4	/**
52			* Method to start the PacketSorter running. This method will
53			* loop forever processing and sending data.
54			*/
55	tdb	1.1	public void run() {
56			int qID = _queue.getQueue();
57			while(true) {
58	tdb	1.4	// attempt to get some data from the Queue
59	tdb	1.1	String xml = "";
60			try {
61			xml = (String) _queue.get(qID);
62			}
63			catch(InvalidQueueException e) {
64			_logger.write(toString(), Logger.ERROR, "Queue failure: "+e);
65			}
66
67	tdb	1.4	// find out which host this packet is actually for
68	tdb	1.3	XMLPacketMaker xmlPacketMaker = new XMLPacketMaker(xml);
69			XMLPacket packet = xmlPacketMaker.createXMLPacket();
70	tdb	1.1	String host = packet.getParam("packet.attributes.machine_name");
71
72	tdb	1.4	// look in the hostMap to see if anyone wants this data
73	tdb	1.3	if(_hostMap.containsKey(host)) {
74			LinkedList list = (LinkedList) _hostMap.get(host);
75			Iterator i = list.iterator();
76	tdb	1.4	// push the data to the listening Handler's queue
77	tdb	1.3	while(i.hasNext()) {
78			((Queue) i.next()).add(xml);
79			}
80	tdb	1.1	}
81	tdb	1.2
82	tdb	1.4	// any handler in this list wants all packets, so send
83			// it on to them regardless
84	tdb	1.2	Iterator j = _allHostsList.iterator();
85			while(j.hasNext()) {
86			((Queue) j.next()).add(xml);
87			}
88	tdb	1.1	}
89			}
90
91	tdb	1.4	/**
92			* Register a DataHandler in the system. This method
93			* actually takes a reference to a Queue, which should be
94			* a Queue that the DataHandler is making use of.
95			* It also takes a hostList, this being a semi-colon
96			* seperated list of hosts that the Client the DataHandler
97			* is serving has requested. If this list is simply an empty
98			* String, it is assumed the Client wants to listen to all
99			* host information.
100			*
101			* @param dhQueue a Queue being used by the DataHandler that is registering
102			* @param hostList a semi-colon seperated list of hosts
103			*/
104	tdb	1.1	public void register(Queue dhQueue, String hostList) {
105	tdb	1.4	// check to see if we want all hosts
106	tdb	1.2	if(hostList.equals("")) {
107			_allHostsList.add(dhQueue);
108			_logger.write(toString(), Logger.SYSMSG, "registered DataHandler for all hosts");
109			}
110			else {
111	tdb	1.4	// go through the list of hosts
112	tdb	1.2	StringTokenizer st = new StringTokenizer(hostList, ";");
113			while(st.hasMoreTokens()) {
114			String host = st.nextToken();
115	tdb	1.4	// see if we already have a list in the map for this host
116	tdb	1.2	if(_hostMap.containsKey(host)) {
117	tdb	1.4	// we do, so add to it
118	tdb	1.2	LinkedList list = (LinkedList) _hostMap.get(host);
119			list.add(dhQueue);
120			}
121			else {
122	tdb	1.4	// we don't, so create a list and put it in the map
123	tdb	1.7	List list = Collections.synchronizedList(new LinkedList());
124	tdb	1.2	list.add(dhQueue);
125			_hostMap.put(host, list);
126			}
127	tdb	1.1	}
128	tdb	1.2	_logger.write(toString(), Logger.SYSMSG, "registered DataHandler for hosts: "+hostList);
129	tdb	1.1	}
130			}
131
132	tdb	1.4	/**
133			* Deregister a DataHandler. The DataHandler should give a reference
134			* to the Queue it's using, and the same hostList it gave when it
135			* register. It is imperative that the hostList is the same, otherwise
136			* there will be all sorts of problems with lists getting out of sync.
137			*
138			* NB: Possible future addition would be recording of hostList's.
139			*
140			* @param dhQueue a Queue being used by the DataHandler that is deregistering
141			* @param hostList a semi-colon seperated list of hosts
142			*/
143	tdb	1.1	public void deregister(Queue dhQueue, String hostList) {
144	tdb	1.4	// go through the list of hosts
145	tdb	1.2	if(hostList.equals("")) {
146			_allHostsList.remove(dhQueue);
147			_logger.write(toString(), Logger.SYSMSG, "deregistered DataHandler for all hosts");
148			}
149			else {
150			StringTokenizer st = new StringTokenizer(hostList, ";");
151			while(st.hasMoreTokens()) {
152			String host = st.nextToken();
153	tdb	1.4	// this should in reality always be true, but best check
154	tdb	1.2	if(_hostMap.containsKey(host)) {
155	tdb	1.4	// get the list and remove the host in question
156	tdb	1.2	LinkedList list = (LinkedList) _hostMap.get(host);
157			list.remove(dhQueue);
158	tdb	1.4	// if the list is now empty, we might as well remove it
159	tdb	1.2	if(list.size()==0) {
160			_hostMap.remove(host);
161			}
162	tdb	1.1	}
163			}
164	tdb	1.2	_logger.write(toString(), Logger.SYSMSG, "deregistered DataHandler for hosts: "+hostList);
165	tdb	1.1	}
166			}
167
168			/**
169			* Overrides the {@link java.lang.Object#toString() Object.toString()}
170			* method to provide clean logging (every class should have this).
171			*
172			* This uses the uk.ac.ukc.iscream.util.NameFormat class
173			* to format the toString()
174			*
175			* @return the name of this class and its CVS revision
176			*/
177			public String toString() {
178			return FormatName.getName(
179			_name,
180			getClass().getName(),
181			REVISION);
182			}
183
184			//---PRIVATE METHODS---
185
186			//---ACCESSOR/MUTATOR METHODS---
187	tdb	1.4
188			/**
189			* Accessor to return a reference to the Queue object. This
190			* is needed so the ClientInterfaceServant can get add data
191			* easily.
192			*
193			* @return a reference to our Queue object.
194			*/
195	tdb	1.1	public Queue getQueue() {
196			return _queue;
197			}
198
199			//---ATTRIBUTES---
200
201			/**
202			* This is the friendly identifier of the
203			* component this class is running in.
204			* eg, a Filter may be called "filter1",
205			* If this class does not have an owning
206			* component, a name from the configuration
207			* can be placed here. This name could also
208			* be changed to null for utility classes.
209			*/
210			private String _name = ClientInterfaceMain.NAME;
211
212			/**
213			* This holds a reference to the
214			* system logger that is being used.
215			*/
216			private Logger _logger = ReferenceManager.getInstance().getLogger();
217
218	tdb	1.4	/**
219			* A reference to the Queue we're using.
220			*/
221	tdb	1.1	private Queue _queue;
222
223	tdb	1.4	/**
224			* A HashMap to store lists of Queue's (in the DataHandlers)
225			* in a way that can be easily accessed when data comes in.
226			*/
227	tdb	1.7	private Map _hostMap;
228	tdb	1.4
229			/**
230			* A list specifically for a Queue's associated with DataHandlers
231			* that want all host information.
232			*/
233	tdb	1.7	private List _allHostsList;
234	tdb	1.1
235			//---STATIC ATTRIBUTES---
236
237			}