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 (24 years, 7 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
#	Content
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	* @author $Author: ajm4 $
17	* @version $Id: PeerHandler.java,v 1.1 2001/05/04 02:04:35 ajm4 Exp $
18	*/
19	class PeerHandler extends Handler {
20
21	//---FINAL / FINAL STATIC ATTRIBUTES---
22
23	/**
24	* The current CVS revision of this class
25	*/
26	public static final String REVISION = "$Revision: 1.1 $";
27
28	//---STATIC METHODS---
29
30	//---CONSTRUCTORS---
31
32	public PeerHandler(Socket socket) throws IOException {
33	super(socket);
34	// setup reader & writer
35	_objectOut = new ObjectOutputStream(_socket.getOutputStream());
36	_objectIn = new ObjectInputStream(_socket.getInputStream());
37	_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
46	while(_running) {
47	// VERY basic protocol, only allow sending of agents ;-)
48	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	}
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	_objectOut.writeObject("SENDAGENT");
80	_objectOut.flush();
81
82	// send agent bytecode
83
84	// TODO - very dodgy and crude at the moment.
85	AgentClassLoader cl = new AgentClassLoader(ClassLoader.getSystemClassLoader());
86
87	// 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
101	// send agent state (serialized object)
102
103	AgentStation.getInstance().removeAgent(agent);
104	_objectOut.writeObject(agent);
105	_objectOut.flush();
106
107	_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	// TODO - very crude
150	// receive agent bytecode
151	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
166
167	// receive agent state (serialized object)
168	Agent agent = (Agent) _objectIn.readObject();
169	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	private ObjectInputStream _objectIn;
221
222	/**
223	* Used for the output stream of this socket
224	*/
225	private ObjectOutputStream _objectOut;
226
227	//---STATIC ATTRIBUTES---
228
229	}