server/filter/UDPReader.java

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

//---IMPORTS---
import java.io.*;
import java.net.*;
import java.util.*;
import uk.ac.ukc.iscream.core.*;
import uk.ac.ukc.iscream.filter.*;
import uk.ac.ukc.iscream.util.*;

/**
 * This class contains the main method to be run by
 * the filter children.  It harvests UDP traffic.
 *
 * @author  $Author: ajm4 $
 * @version $Id: UDPReader.java,v 1.11 2000/12/13 21:11:26 ajm4 Exp $
 */
public class UDPReader extends Thread{

//---FINAL ATTRIBUTES---

    /**
     * The current CVS revision of this class
     */
    public final String REVISION = "$Revision: 1.11 $";

    /**
     * The maximum size of a packet
     */
    private final int packetSizeLimit = 8192;   
    
//---STATIC METHODS---

//---CONSTRUCTORS---

    /**
     * It is normal to use this constructor in preference
     * to any other in this class.
     */
    public UDPReader(int port, Queue queue){
        _port = port;
        _queue = queue;
    }

//---PUBLIC METHODS---

    public void run() {

        DatagramSocket socket = null;
        try {
            socket = new DatagramSocket(_port);
        }
        catch (BindException e){
            _logger.write(this.toString(), Logger.FATAL, "Could not start the UDPReader thread on port "+_port+" as this port was already in use.");
            return;
        }
        catch (Exception e){
            _logger.write(this.toString(), Logger.FATAL, "Could not start the UDPReader thread on port "+_port+".");
            return;
        }
        
        _logger.write(this.toString(), Logger.SYSMSG, "UDPReader thread ready and listening for UDP packets on port "+_port);

        byte[] buf;
        
        boolean running = true;
        while (running){
            try {

                    // receive request and put it in the Queue               
                buf = new byte[packetSizeLimit];
                DatagramPacket packet = new DatagramPacket(buf, buf.length);
                socket.receive(packet);
                String xml = new String(packet.getData());
                _queue.add(xml);
            }
            catch (IOException e) {
                _logger.write(this.toString(), Logger.WARNING, "This UDPReader thread has been shut down as an exception occured: "+e);
                return;
            }
        }
        socket.close();
    }

    /**
     * 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---

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

    /**
     * This holds a reference to the
     * system logger that is being used.
     */
    private Logger _logger = ReferenceManager.getInstance().getLogger();
    
    /**
     * The port that this reader is using
     */
    int _port;
    
    /**
     * The Queue object
     */
    Queue _queue;    

//---STATIC ATTRIBUTES---

}
Revision:	1.12
Committed:	Fri Jan 12 00:45:25 2001 UTC (23 years, 4 months ago) by tdb
Branch:	MAIN
Changes since 1.11:	+9 -10 lines
Log Message:	A structural change to the Filter. The old setup looked like this; UDPReader ------> FilterThread (thread) --> (parent) TCPReader ------> FilterThread (thread) --> (parent) FilterServant --> FilterThread (thread) --> (parent) Seeing this from a threaded point of view, each time a packet came in (through whatever means - UDP, TCP or CORBA), a FilterThread instance was created to deal with it. If the link to the parent was slow this resulting in a build-up of FilterThreads all waiting to talk to the parent - and there is only one actual parent object, with a synchronised thread, so they have to queue up anyway. As a result of this, the following change has been made. UDPReader -------\ TCPReader ----------> Queue (single) >-- FilterThread --> (parent) FilterServant ---/ In this setup, each of the three objects that generate packets only see the single instance of a Queue. They all add their data to this Queue, and then carry on with the task of listening. The FilterThread (having it's role changed slightly) now acts as a consumer of the Queue, in that it grabs data from the Queue and deals with passing it on to the parent. This setup should be more efficient in the long run, especially under a high load situation. The only problem could be the Queue growing to an unlimited size, but this is a Queue design issue.
#	Content
1	//---PACKAGE DECLARATION---
2	package uk.ac.ukc.iscream.filter;
3
4	//---IMPORTS---
5	import java.io.*;
6	import java.net.*;
7	import java.util.*;
8	import uk.ac.ukc.iscream.core.*;
9	import uk.ac.ukc.iscream.filter.*;
10	import uk.ac.ukc.iscream.util.*;
11
12	/**
13	* This class contains the main method to be run by
14	* the filter children. It harvests UDP traffic.
15	*
16	* @author $Author: ajm4 $
17	* @version $Id: UDPReader.java,v 1.11 2000/12/13 21:11:26 ajm4 Exp $
18	*/
19	public class UDPReader extends Thread{
20
21	//---FINAL ATTRIBUTES---
22
23	/**
24	* The current CVS revision of this class
25	*/
26	public final String REVISION = "$Revision: 1.11 $";
27
28	/**
29	* The maximum size of a packet
30	*/
31	private final int packetSizeLimit = 8192;
32
33	//---STATIC METHODS---
34
35	//---CONSTRUCTORS---
36
37	/**
38	* It is normal to use this constructor in preference
39	* to any other in this class.
40	*/
41	public UDPReader(int port, Queue queue){
42	_port = port;
43	_queue = queue;
44	}
45
46	//---PUBLIC METHODS---
47
48	public void run() {
49
50	DatagramSocket socket = null;
51	try {
52	socket = new DatagramSocket(_port);
53	}
54	catch (BindException e){
55	_logger.write(this.toString(), Logger.FATAL, "Could not start the UDPReader thread on port "+_port+" as this port was already in use.");
56	return;
57	}
58	catch (Exception e){
59	_logger.write(this.toString(), Logger.FATAL, "Could not start the UDPReader thread on port "+_port+".");
60	return;
61	}
62
63	_logger.write(this.toString(), Logger.SYSMSG, "UDPReader thread ready and listening for UDP packets on port "+_port);
64
65	byte[] buf;
66
67	boolean running = true;
68	while (running){
69	try {
70
71	// receive request and put it in the Queue
72	buf = new byte[packetSizeLimit];
73	DatagramPacket packet = new DatagramPacket(buf, buf.length);
74	socket.receive(packet);
75	String xml = new String(packet.getData());
76	_queue.add(xml);
77	}
78	catch (IOException e) {
79	_logger.write(this.toString(), Logger.WARNING, "This UDPReader thread has been shut down as an exception occured: "+e);
80	return;
81	}
82	}
83	socket.close();
84	}
85
86	/**
87	* Overrides the {@link java.lang.Object#toString() Object.toString()}
88	* method to provide clean logging (every class should have this).
89	*
90	* This uses the uk.ac.ukc.iscream.util.NameFormat class
91	* to format the toString()
92	*
93	* @return the name of this class and its CVS revision
94	*/
95	public String toString() {
96	return FormatName.getName(
97	_name,
98	getClass().getName(),
99	REVISION);
100	}
101
102	//---PRIVATE METHODS---
103
104	//---ACCESSOR/MUTATOR METHODS---
105
106	//---ATTRIBUTES---
107
108	/**
109	* This is the friendly identifier of the
110	* component this class is running in.
111	* eg, a Filter may be called "filter1",
112	* If this class does not have an owning
113	* component, a name from the configuration
114	* can be placed here. This name could also
115	* be changed to null for utility classes.
116	*/
117	private String _name = FilterMain.NAME;
118
119	/**
120	* This holds a reference to the
121	* system logger that is being used.
122	*/
123	private Logger _logger = ReferenceManager.getInstance().getLogger();
124
125	/**
126	* The port that this reader is using
127	*/
128	int _port;
129
130	/**
131	* The Queue object
132	*/
133	Queue _queue;
134
135	//---STATIC ATTRIBUTES---
136
137	}