server/filter/TCPReaderInit.java

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

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

/**
 * This provides Host heartbeat functionality
 *
 * @author  $Author: ajm4 $
 * @version $Id: TCPReaderInit.java,v 1.10 2000/12/13 13:36:46 ajm4 Exp $
 */
class TCPReaderInit extends Thread {

//---FINAL ATTRIBUTES---

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

//---CONSTRUCTORS---

    public TCPReaderInit(Socket socket, Queue queue) throws IOException {
        _socket = socket;
        _queue = queue;
        _socketIn = new BufferedReader(new InputStreamReader(_socket.getInputStream()));
        _socketOut = new PrintWriter(_socket.getOutputStream());
        _logger.write(toString(), Logger.SYSINIT, "created");
    }
    
//---PUBLIC METHODS---

    public void run() {
        try {
            //variables
            String filelist = "";
            String lastModified = "";
            String inBound = "";
            
            inBound = _socketIn.readLine();
            if(!inBound.equals("HEARTBEAT")) {
                _socketOut.println("ERROR");
                _socketOut.flush();
                throw new IOException("protocol error - expecting:HEARTBEAT got:" + inBound);
            } else {
                _socketOut.println("OK");
                _socketOut.flush();
            }
            
            inBound = _socketIn.readLine();
            if(!inBound.equals("CONFIG")) {
                _socketOut.println("ERROR");
                _socketOut.flush();
                throw new IOException("protocol error - expecting:CONFIG got:" + inBound);
            } else {
                _socketOut.println("OK");
                _socketOut.flush();
            }
            
            inBound = _socketIn.readLine();
            filelist = inBound;
            _socketOut.println("OK");
            _socketOut.flush();
            
            inBound = _socketIn.readLine();
            lastModified = inBound;
            
            boolean newConfig = _configManager.isModified(filelist, Long.parseLong(lastModified));
            
            if(newConfig) {
                _socketOut.println("ERROR");
            }
            else {
                _socketOut.println("OK");
            }
            _socketOut.flush();
            
            inBound = _socketIn.readLine();
            if(!inBound.equals("ENDHEARTBEAT")) {
                _socketOut.println("ERROR");
                _socketOut.flush();
                throw new IOException("protocol error - expecting:ENDHEARTBEAT got:" + inBound);
            } else {
                _socketOut.println("OK");
                _socketOut.flush();
            }

            String date = new Long(System.currentTimeMillis()).toString();
            String hostname = _socket.getInetAddress().getHostName();
            String ipadd = _socket.getInetAddress().getHostAddress();
            String xml = "<packet type=\"heartbeat\" machine_name=\""+hostname+"\" date=\""+date+"\" ip=\""+ipadd+"\"></packet>";
            
            _queue.add(xml);
            
        } catch (Exception e) {
            _logger.write(toString(), Logger.ERROR, "ERROR: " + e.getMessage());
        }
        
        _socketOut.flush();
        // Disconnect streams & socket
        try {
            _socketIn.close();
            _socketOut.close();
            _socket.close();
        } catch (IOException e) {
            _logger.write(toString(), Logger.ERROR, "exception on socket close");
        }
        _logger.write(toString(), Logger.SYSMSG, "finished");
    }
    
    /**
     * 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---

    /**
     * A reference to the configuration manager
     */
    ConfigurationManager _configManager = ReferenceManager.getInstance().getCM();
    
    /**
     * 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 socket we are talking on
     */
    Socket _socket;
    
    /** 
     * The input from the socket
     */
    BufferedReader _socketIn;
    
    /**
     * The output from the socket
     */
    PrintWriter _socketOut;
    
    /**
     * A reference to our Queue
     */
    Queue _queue;
//---STATIC ATTRIBUTES---

}
Revision:	1.11
Committed:	Fri Jan 12 00:45:25 2001 UTC (23 years, 4 months ago) by tdb
Branch:	MAIN
Changes since 1.10:	+7 -8 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.
#	User	Rev	Content
1	tdb	1.1	//---PACKAGE DECLARATION---
2	tdb	1.5	package uk.ac.ukc.iscream.filter;
3	tdb	1.1
4			//---IMPORTS---
5			import uk.ac.ukc.iscream.core.*;
6			import uk.ac.ukc.iscream.filter.*;
7			import java.net.Socket;
8			import java.io.InputStream;
9			import java.io.OutputStream;
10			import java.io.IOException;
11			import java.io.*;
12	ajm	1.9	import uk.ac.ukc.iscream.util.*;
13	tdb	1.1
14			/**
15	ajm	1.10	* This provides Host heartbeat functionality
16	tdb	1.1	*
17	ajm	1.10	* @author $Author: ajm4 $
18	tdb	1.11	* @version $Id: TCPReaderInit.java,v 1.10 2000/12/13 13:36:46 ajm4 Exp $
19	tdb	1.1	*/
20			class TCPReaderInit extends Thread {
21
22			//---FINAL ATTRIBUTES---
23
24			/**
25			* The current CVS revision of this class
26			*/
27	tdb	1.11	public final String REVISION = "$Revision: 1.10 $";
28	tdb	1.1
29			//---STATIC METHODS---
30
31			//---CONSTRUCTORS---
32
33	tdb	1.11	public TCPReaderInit(Socket socket, Queue queue) throws IOException {
34	tdb	1.1	_socket = socket;
35	tdb	1.11	_queue = queue;
36	tdb	1.1	_socketIn = new BufferedReader(new InputStreamReader(_socket.getInputStream()));
37			_socketOut = new PrintWriter(_socket.getOutputStream());
38	tdb	1.8	_logger.write(toString(), Logger.SYSINIT, "created");
39	tdb	1.1	}
40
41			//---PUBLIC METHODS---
42
43			public void run() {
44			try {
45	tdb	1.3	//variables
46			String filelist = "";
47			String lastModified = "";
48			String inBound = "";
49	tdb	1.2
50	tdb	1.3	inBound = _socketIn.readLine();
51			if(!inBound.equals("HEARTBEAT")) {
52			_socketOut.println("ERROR");
53			_socketOut.flush();
54			throw new IOException("protocol error - expecting:HEARTBEAT got:" + inBound);
55			} else {
56			_socketOut.println("OK");
57			_socketOut.flush();
58			}
59
60			inBound = _socketIn.readLine();
61			if(!inBound.equals("CONFIG")) {
62			_socketOut.println("ERROR");
63			_socketOut.flush();
64			throw new IOException("protocol error - expecting:CONFIG got:" + inBound);
65			} else {
66			_socketOut.println("OK");
67			_socketOut.flush();
68			}
69
70			inBound = _socketIn.readLine();
71			filelist = inBound;
72			_socketOut.println("OK");
73			_socketOut.flush();
74
75			inBound = _socketIn.readLine();
76			lastModified = inBound;
77
78			boolean newConfig = _configManager.isModified(filelist, Long.parseLong(lastModified));
79
80			if(newConfig) {
81			_socketOut.println("ERROR");
82			}
83			else {
84			_socketOut.println("OK");
85			}
86			_socketOut.flush();
87
88			inBound = _socketIn.readLine();
89			if(!inBound.equals("ENDHEARTBEAT")) {
90			_socketOut.println("ERROR");
91			_socketOut.flush();
92			throw new IOException("protocol error - expecting:ENDHEARTBEAT got:" + inBound);
93			} else {
94			_socketOut.println("OK");
95			_socketOut.flush();
96			}
97	tdb	1.4
98	tdb	1.6	String date = new Long(System.currentTimeMillis()).toString();
99			String hostname = _socket.getInetAddress().getHostName();
100			String ipadd = _socket.getInetAddress().getHostAddress();
101			String xml = "<packet type=\"heartbeat\" machine_name=\""+hostname+"\" date=\""+date+"\" ip=\""+ipadd+"\"></packet>";
102
103	tdb	1.11	_queue.add(xml);
104	tdb	1.1
105			} catch (Exception e) {
106	tdb	1.8	_logger.write(toString(), Logger.ERROR, "ERROR: " + e.getMessage());
107	tdb	1.1	}
108
109			_socketOut.flush();
110			// Disconnect streams & socket
111			try {
112			_socketIn.close();
113			_socketOut.close();
114			_socket.close();
115			} catch (IOException e) {
116	tdb	1.8	_logger.write(toString(), Logger.ERROR, "exception on socket close");
117	tdb	1.1	}
118	tdb	1.8	_logger.write(toString(), Logger.SYSMSG, "finished");
119	tdb	1.1	}
120
121			/**
122			* Overrides the {@link java.lang.Object#toString() Object.toString()}
123			* method to provide clean logging (every class should have this).
124			*
125	ajm	1.10	* This uses the uk.ac.ukc.iscream.util.NameFormat class
126			* to format the toString()
127			*
128	tdb	1.1	* @return the name of this class and its CVS revision
129			*/
130			public String toString() {
131	ajm	1.10	return FormatName.getName(
132			_name,
133			getClass().getName(),
134			REVISION);
135	tdb	1.1	}
136
137			//---PRIVATE METHODS---
138
139			//---ACCESSOR/MUTATOR METHODS---
140
141			//---ATTRIBUTES---
142
143	ajm	1.10	/**
144			* A reference to the configuration manager
145			*/
146	tdb	1.8	ConfigurationManager _configManager = ReferenceManager.getInstance().getCM();
147	ajm	1.10
148			/**
149			* This is the friendly identifier of the
150			* component this class is running in.
151			* eg, a Filter may be called "filter1",
152			* If this class does not have an owning
153			* component, a name from the configuration
154			* can be placed here. This name could also
155			* be changed to null for utility classes.
156			*/
157			private String _name = FilterMain.NAME;
158
159			/**
160			* This holds a reference to the
161			* system logger that is being used.
162			*/
163			private Logger _logger = ReferenceManager.getInstance().getLogger();
164
165			/**
166			* The socket we are talking on
167			*/
168	tdb	1.1	Socket _socket;
169	ajm	1.10
170			/**
171			* The input from the socket
172			*/
173	tdb	1.1	BufferedReader _socketIn;
174	ajm	1.10
175			/**
176			* The output from the socket
177			*/
178	tdb	1.1	PrintWriter _socketOut;
179	ajm	1.10
180			/**
181	tdb	1.11	* A reference to our Queue
182	ajm	1.10	*/
183	tdb	1.11	Queue _queue;
184	tdb	1.1	//---STATIC ATTRIBUTES---
185
186			}