experimental/agents/PeerHandler.java

//---PACKAGE DECLARATION---

//---IMPORTS---
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.PrintWriter;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.IOException;
import java.net.Socket;

/**
 * Handles communication between this agent station and a peer 
 * agent stations.
 *
 * @author  $Author: ajm4 $
 * @version $Id: PeerHandler.java,v 1.1 2001/05/04 02:04:35 ajm4 Exp $
 */
class PeerHandler extends Handler {

//---FINAL / FINAL STATIC ATTRIBUTES---

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

//---CONSTRUCTORS---

    public PeerHandler(Socket socket) throws IOException {
        super(socket);
        // setup reader & writer
        _objectOut = new ObjectOutputStream(_socket.getOutputStream()); 
        _objectIn = new ObjectInputStream(_socket.getInputStream()); 
        _logger.write(toString(), Logger.SYSINIT, "created");
    }

//---PUBLIC METHODS---

    public void run() {
        _logger.write(toString(), Logger.SYSINIT, "started");
        try {    
            
            while(_running) {
                // VERY basic protocol, only allow sending of agents ;-)
                try {
                    String message = (String) _objectIn.readObject();
                    if (message.equals("SENDAGENT")) {
                        receiveAgent();
                    } else {
                        _logger.write(toString(), Logger.ERROR, "protocol error expected: SENDAGENT got: " + message);
                    }
                } catch (ClassNotFoundException e) {
                    _logger.write(toString(), Logger.ERROR, "object protocol error expected: a String got: a " + e.getMessage());
                }
            }
            
        } catch (IOException e) {
            _logger.write(toString(), Logger.ERROR, "I/O ERROR - " + e);
        }
        
        shutdownSocket();
        AgentStation.getInstance().removePeer(this);
        _logger.write(toString(), Logger.SYSMSG, "finished");
    }

    public boolean sendAgent(Agent agent) {
        boolean result = false;
        _logger.write(toString(), Logger.SYSMSG, "starting agent transfer - outbound");
        try {
            // VERY basic protocol, only allow sending of agents ;-)
            
            // INCOMPLETE - infact it needs to create a second socket as I don't believe multiple
            // stream types (ie, objectstream and printwriter) can be used together
            // needs investigating
            
            _objectOut.writeObject("SENDAGENT");
            _objectOut.flush();
                        
            // send agent bytecode
            
            // TODO - very dodgy and crude at the moment.
            AgentClassLoader cl = new AgentClassLoader(ClassLoader.getSystemClassLoader());
            
            // send the super class
            byte[] byteCode = cl.getBytecode(agent.getClass().getSuperclass().getName());
            _objectOut.writeInt(byteCode.length);
            _objectOut.flush();
            _objectOut.write(byteCode);
            _objectOut.flush();
            
            // send the agent
            byteCode = cl.getBytecode(agent.getClass().getName());
            _objectOut.writeInt(byteCode.length);
            _objectOut.flush();
            _objectOut.write(byteCode);
            _objectOut.flush();
            
            // send agent state (serialized object)
            
            AgentStation.getInstance().removeAgent(agent);
            _objectOut.writeObject(agent);
            _objectOut.flush();
                        
            _logger.write(toString(), Logger.SYSMSG, "agent sent");
            result = true;
        } catch (IOException e) {
            _logger.write(toString(), Logger.ERROR, "I/O ERROR - " + e);
        }
        return result;
    }

    /**
     * Used to shutdown this thread
     */
    public void shutdown() {
        _running = false;
    }

    /**
     * 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.FormatName 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/PROTECTED METHODS---

    /**
     * Carry out functions to receive an Agent
     * NOTE - duplicated code in DirectAgentHandler, solution? -
     *
     * @throws IOException on I/O error
     */
    private void receiveAgent() throws IOException {
        _logger.write(toString(), Logger.SYSMSG, "starting agent transfer - inbound");
        try {
            // TODO - very crude
            // receive agent bytecode
            AgentClassLoader cl = new AgentClassLoader(ClassLoader.getSystemClassLoader());
            
            // first the super class
            int givenLength = _objectIn.readInt();
            byte[] byteCode = new byte[givenLength];
            _objectIn.readFully(byteCode);
            cl.defineAgentClass(byteCode);
            
            // then the agent
            
            givenLength = _objectIn.readInt();
            byteCode = new byte[givenLength];
            _objectIn.readFully(byteCode);
            cl.defineAgentClass(byteCode);
            
        
            // receive agent state (serialized object)
            Agent agent = (Agent) _objectIn.readObject();
            AgentStation.getInstance().addAgent(agent);
                    
            _logger.write(toString(), Logger.DEBUG, "Agent state received");
            _logger.write(toString(), Logger.SYSMSG, "Agent name - " + agent.getName());
            _logger.write(toString(), Logger.DEBUG, agent.getName() + "'s transmission count - " + agent.getTransmissionCount());
            
            _logger.write(toString(), Logger.SYSMSG, "Welcoming agent");
            agent.onArrival(AgentStation.getInstance());
    
            _logger.write(toString(), Logger.SYSMSG, "Starting agent thread");
            
            Thread thread = new Thread(agent);
            thread.start();
            
            _logger.write(toString(), Logger.SYSMSG, "agent received");
            
        } catch (ClassNotFoundException e) {
            _logger.write(toString(), Logger.ERROR, "ERROR - possible error transferring agent bytecode - " + e);
        }
    }
    
//---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 = "PeerHandler";

    /**
     * This holds a reference to the
     * system logger that is being used.
     */
    private Logger _logger = Logger.getInstance();

    /**
     * A "running" used by the main loop in the run() method to determine if
     * it should continue running this thread.
     */   
    private boolean _running = true;
    
    /**
     * Used for the input stream of this socket
     */
    private ObjectInputStream _objectIn;

    /**
     * Used for the output stream of this socket
     */
    private ObjectOutputStream _objectOut;
        
//---STATIC ATTRIBUTES---

}
Revision:	1.2
Committed:	Sun May 6 19:01:48 2001 UTC (23 years, 6 months ago) by ajm
Branch:	MAIN
CVS Tags:	HEAD
Changes since 1.1:	+62 -58 lines
Log Message:	Now has some VERY crude byte code passing mechanisms in place. Currently it does it a very silly and non standard way, this will change. But hey, it works ;p
#	User	Rev	Content
1	ajm	1.1	//---PACKAGE DECLARATION---
2
3			//---IMPORTS---
4			import java.io.ObjectInputStream;
5			import java.io.ObjectOutputStream;
6			import java.io.PrintWriter;
7			import java.io.BufferedReader;
8			import java.io.InputStreamReader;
9			import java.io.IOException;
10			import java.net.Socket;
11
12			/**
13			* Handles communication between this agent station and a peer
14			* agent stations.
15			*
16	ajm	1.2	* @author $Author: ajm4 $
17			* @version $Id: PeerHandler.java,v 1.1 2001/05/04 02:04:35 ajm4 Exp $
18	ajm	1.1	*/
19			class PeerHandler extends Handler {
20
21			//---FINAL / FINAL STATIC ATTRIBUTES---
22
23			/**
24			* The current CVS revision of this class
25			*/
26	ajm	1.2	public static final String REVISION = "$Revision: 1.1 $";
27	ajm	1.1
28			//---STATIC METHODS---
29
30			//---CONSTRUCTORS---
31
32			public PeerHandler(Socket socket) throws IOException {
33			super(socket);
34			// setup reader & writer
35	ajm	1.2	_objectOut = new ObjectOutputStream(_socket.getOutputStream());
36			_objectIn = new ObjectInputStream(_socket.getInputStream());
37	ajm	1.1	_logger.write(toString(), Logger.SYSINIT, "created");
38			}
39
40			//---PUBLIC METHODS---
41
42			public void run() {
43			_logger.write(toString(), Logger.SYSINIT, "started");
44			try {
45	ajm	1.2
46	ajm	1.1	while(_running) {
47			// VERY basic protocol, only allow sending of agents ;-)
48	ajm	1.2	try {
49			String message = (String) _objectIn.readObject();
50			if (message.equals("SENDAGENT")) {
51			receiveAgent();
52			} else {
53			_logger.write(toString(), Logger.ERROR, "protocol error expected: SENDAGENT got: " + message);
54			}
55			} catch (ClassNotFoundException e) {
56			_logger.write(toString(), Logger.ERROR, "object protocol error expected: a String got: a " + e.getMessage());
57			}
58	ajm	1.1	}
59
60			} catch (IOException e) {
61			_logger.write(toString(), Logger.ERROR, "I/O ERROR - " + e);
62			}
63
64			shutdownSocket();
65			AgentStation.getInstance().removePeer(this);
66			_logger.write(toString(), Logger.SYSMSG, "finished");
67			}
68
69			public boolean sendAgent(Agent agent) {
70			boolean result = false;
71			_logger.write(toString(), Logger.SYSMSG, "starting agent transfer - outbound");
72			try {
73			// VERY basic protocol, only allow sending of agents ;-)
74
75			// INCOMPLETE - infact it needs to create a second socket as I don't believe multiple
76			// stream types (ie, objectstream and printwriter) can be used together
77			// needs investigating
78
79	ajm	1.2	_objectOut.writeObject("SENDAGENT");
80			_objectOut.flush();
81
82			// send agent bytecode
83	ajm	1.1
84	ajm	1.2	// TODO - very dodgy and crude at the moment.
85			AgentClassLoader cl = new AgentClassLoader(ClassLoader.getSystemClassLoader());
86	ajm	1.1
87	ajm	1.2	// send the super class
88			byte[] byteCode = cl.getBytecode(agent.getClass().getSuperclass().getName());
89			_objectOut.writeInt(byteCode.length);
90			_objectOut.flush();
91			_objectOut.write(byteCode);
92			_objectOut.flush();
93
94			// send the agent
95			byteCode = cl.getBytecode(agent.getClass().getName());
96			_objectOut.writeInt(byteCode.length);
97			_objectOut.flush();
98			_objectOut.write(byteCode);
99			_objectOut.flush();
100	ajm	1.1
101			// send agent state (serialized object)
102	ajm	1.2
103	ajm	1.1	AgentStation.getInstance().removeAgent(agent);
104	ajm	1.2	_objectOut.writeObject(agent);
105			_objectOut.flush();
106
107	ajm	1.1	_logger.write(toString(), Logger.SYSMSG, "agent sent");
108			result = true;
109			} catch (IOException e) {
110			_logger.write(toString(), Logger.ERROR, "I/O ERROR - " + e);
111			}
112			return result;
113			}
114
115			/**
116			* Used to shutdown this thread
117			*/
118			public void shutdown() {
119			_running = false;
120			}
121
122			/**
123			* Overrides the {@link java.lang.Object#toString() Object.toString()}
124			* method to provide clean logging (every class should have this).
125			*
126			* This uses the uk.ac.ukc.iscream.util.FormatName class
127			* to format the toString()
128			*
129			* @return the name of this class and its CVS revision
130			*/
131			public String toString() {
132			return FormatName.getName(
133			_name,
134			getClass().getName(),
135			REVISION);
136			}
137
138			//---PRIVATE/PROTECTED METHODS---
139
140			/**
141			* Carry out functions to receive an Agent
142			* NOTE - duplicated code in DirectAgentHandler, solution? -
143			*
144			* @throws IOException on I/O error
145			*/
146			private void receiveAgent() throws IOException {
147			_logger.write(toString(), Logger.SYSMSG, "starting agent transfer - inbound");
148			try {
149	ajm	1.2	// TODO - very crude
150	ajm	1.1	// receive agent bytecode
151	ajm	1.2	AgentClassLoader cl = new AgentClassLoader(ClassLoader.getSystemClassLoader());
152
153			// first the super class
154			int givenLength = _objectIn.readInt();
155			byte[] byteCode = new byte[givenLength];
156			_objectIn.readFully(byteCode);
157			cl.defineAgentClass(byteCode);
158
159			// then the agent
160
161			givenLength = _objectIn.readInt();
162			byteCode = new byte[givenLength];
163			_objectIn.readFully(byteCode);
164			cl.defineAgentClass(byteCode);
165	ajm	1.1
166
167			// receive agent state (serialized object)
168	ajm	1.2	Agent agent = (Agent) _objectIn.readObject();
169	ajm	1.1	AgentStation.getInstance().addAgent(agent);
170
171			_logger.write(toString(), Logger.DEBUG, "Agent state received");
172			_logger.write(toString(), Logger.SYSMSG, "Agent name - " + agent.getName());
173			_logger.write(toString(), Logger.DEBUG, agent.getName() + "'s transmission count - " + agent.getTransmissionCount());
174
175			_logger.write(toString(), Logger.SYSMSG, "Welcoming agent");
176			agent.onArrival(AgentStation.getInstance());
177
178			_logger.write(toString(), Logger.SYSMSG, "Starting agent thread");
179
180			Thread thread = new Thread(agent);
181			thread.start();
182
183			_logger.write(toString(), Logger.SYSMSG, "agent received");
184
185			} catch (ClassNotFoundException e) {
186			_logger.write(toString(), Logger.ERROR, "ERROR - possible error transferring agent bytecode - " + e);
187			}
188			}
189
190			//---ACCESSOR/MUTATOR METHODS---
191
192			//---ATTRIBUTES---
193
194			/**
195			* This is the friendly identifier of the
196			* component this class is running in.
197			* eg, a Filter may be called "filter1",
198			* If this class does not have an owning
199			* component, a name from the configuration
200			* can be placed here. This name could also
201			* be changed to null for utility classes.
202			*/
203			private String _name = "PeerHandler";
204
205			/**
206			* This holds a reference to the
207			* system logger that is being used.
208			*/
209			private Logger _logger = Logger.getInstance();
210
211			/**
212			* A "running" used by the main loop in the run() method to determine if
213			* it should continue running this thread.
214			*/
215			private boolean _running = true;
216
217			/**
218			* Used for the input stream of this socket
219			*/
220	ajm	1.2	private ObjectInputStream _objectIn;
221
222			/**
223			* Used for the output stream of this socket
224			*/
225			private ObjectOutputStream _objectOut;
226
227	ajm	1.1	//---STATIC ATTRIBUTES---
228
229			}