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: tdb1 $
 * @version $Id: TemplateClass.java,v 1.11 2001/03/22 21:50:41 tdb1 Exp $
 */
class PeerHandler extends Handler {

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

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

//---CONSTRUCTORS---

    public PeerHandler(Socket socket) throws IOException {
        super(socket);
        // setup reader & writer
        _socketIn = new BufferedReader(new InputStreamReader(_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 ;-)
                getInBoundMessage("SENDAGENT");
                receiveAgent();
            }
            
        } 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
            
            PrintWriter socketOut = new PrintWriter(_socket.getOutputStream(), true);
            socketOut.println("SENDAGENT");
            socketOut.close();
            
            // send agent bytecode
            
            // TODO
            
            // send agent state (serialized object)
            ObjectOutputStream objectOutStream = new ObjectOutputStream(_socket.getOutputStream());
            AgentStation.getInstance().removeAgent(agent);
            objectOutStream.writeObject(agent);
            objectOutStream.flush();
            objectOutStream.close();
            
            _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 {
            // receive agent bytecode
            
            // TODO
        
            // receive agent state (serialized object)
            ObjectInputStream objectInStream = new ObjectInputStream(_socket.getInputStream());
            Agent agent = (Agent) objectInStream.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();
            
            objectInStream.close();
            _logger.write(toString(), Logger.SYSMSG, "agent received");
            
        } catch (ClassNotFoundException e) {
            _logger.write(toString(), Logger.ERROR, "ERROR - possible error transferring agent bytecode - " + e);
        }
    }
    
    /**
     * Get the next inbound string from the socket and expect it
     * to be the given text.
     *
     * @param expected the expected text
     *
     * @throws IOException on I/O error or protocol error
     */
     private String getInBoundMessage(String expected) throws IOException {
        // grab the input
        String inBound = getInBoundMessage();
        // check if it's what we're expecting
        if(!inBound.equals(expected)) {
            throw new IOException("protocol error - expected:" + expected + " got:" + inBound);
        }
        // it should be ok then
        return inBound;
    }
    
    /**
     * Get the next inbound string from the socket.
     *
     * @throws IOException on I/O error or protocol error
     */
    private String getInBoundMessage() throws IOException {
        // grab the input
        String inBound = _socketIn.readLine();
        // check for null's, likely disconnection
        if(inBound == null) {
            throw new IOException("got null from peer, maybe it died");
        }
        // it's a valid message it seems
        return inBound;
    }

//---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 BufferedReader _socketIn;
    
//---STATIC ATTRIBUTES---

}
Revision:	1.1
Committed:	Fri May 4 02:04:35 2001 UTC (23 years, 6 months ago) by ajm
Branch:	MAIN
Log Message:	Loads of new framework but far from complete. The BasicAgent class is as it says, a basic agent, with ability to be run from the command line to fire it into an agentstation. AgentStations now have support for adding and removing agents and support for multiple listeners. Also initial support for peer agent stations. Attempted support at peer -> peer agent transfer, but run into problems. Still no bytecode transfer, but location and method has been figured out. The AgentSystem class is now the bootstrap class for an agent station. It brings up the AgentStation (which is now singleton) and the Logger (also singleton - using standard i-scream logging techniques). It is possible to specify a peer station that the booting agent station should connect to. Initial TODO: agent class loader agent bytecode transfer solve peer -> peer problems - possibly by creating extra socket to send agent, but shouldn't really be needed
#	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: tdb1 $
17	* @version $Id: TemplateClass.java,v 1.11 2001/03/22 21:50:41 tdb1 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.11 $";
27
28	//---STATIC METHODS---
29
30	//---CONSTRUCTORS---
31
32	public PeerHandler(Socket socket) throws IOException {
33	super(socket);
34	// setup reader & writer
35	_socketIn = new BufferedReader(new InputStreamReader(_socket.getInputStream()));
36	_logger.write(toString(), Logger.SYSINIT, "created");
37	}
38
39	//---PUBLIC METHODS---
40
41	public void run() {
42	_logger.write(toString(), Logger.SYSINIT, "started");
43	try {
44
45	while(_running) {
46	// VERY basic protocol, only allow sending of agents ;-)
47	getInBoundMessage("SENDAGENT");
48	receiveAgent();
49	}
50
51	} catch (IOException e) {
52	_logger.write(toString(), Logger.ERROR, "I/O ERROR - " + e);
53	}
54
55	shutdownSocket();
56	AgentStation.getInstance().removePeer(this);
57	_logger.write(toString(), Logger.SYSMSG, "finished");
58	}
59
60	public boolean sendAgent(Agent agent) {
61	boolean result = false;
62	_logger.write(toString(), Logger.SYSMSG, "starting agent transfer - outbound");
63	try {
64	// VERY basic protocol, only allow sending of agents ;-)
65
66	// INCOMPLETE - infact it needs to create a second socket as I don't believe multiple
67	// stream types (ie, objectstream and printwriter) can be used together
68	// needs investigating
69
70	PrintWriter socketOut = new PrintWriter(_socket.getOutputStream(), true);
71	socketOut.println("SENDAGENT");
72	socketOut.close();
73
74	// send agent bytecode
75
76	// TODO
77
78	// send agent state (serialized object)
79	ObjectOutputStream objectOutStream = new ObjectOutputStream(_socket.getOutputStream());
80	AgentStation.getInstance().removeAgent(agent);
81	objectOutStream.writeObject(agent);
82	objectOutStream.flush();
83	objectOutStream.close();
84
85	_logger.write(toString(), Logger.SYSMSG, "agent sent");
86	result = true;
87	} catch (IOException e) {
88	_logger.write(toString(), Logger.ERROR, "I/O ERROR - " + e);
89	}
90	return result;
91	}
92
93	/**
94	* Used to shutdown this thread
95	*/
96	public void shutdown() {
97	_running = false;
98	}
99
100	/**
101	* Overrides the {@link java.lang.Object#toString() Object.toString()}
102	* method to provide clean logging (every class should have this).
103	*
104	* This uses the uk.ac.ukc.iscream.util.FormatName class
105	* to format the toString()
106	*
107	* @return the name of this class and its CVS revision
108	*/
109	public String toString() {
110	return FormatName.getName(
111	_name,
112	getClass().getName(),
113	REVISION);
114	}
115
116	//---PRIVATE/PROTECTED METHODS---
117
118	/**
119	* Carry out functions to receive an Agent
120	* NOTE - duplicated code in DirectAgentHandler, solution? -
121	*
122	* @throws IOException on I/O error
123	*/
124	private void receiveAgent() throws IOException {
125	_logger.write(toString(), Logger.SYSMSG, "starting agent transfer - inbound");
126	try {
127	// receive agent bytecode
128
129	// TODO
130
131	// receive agent state (serialized object)
132	ObjectInputStream objectInStream = new ObjectInputStream(_socket.getInputStream());
133	Agent agent = (Agent) objectInStream.readObject();
134	AgentStation.getInstance().addAgent(agent);
135
136	_logger.write(toString(), Logger.DEBUG, "Agent state received");
137	_logger.write(toString(), Logger.SYSMSG, "Agent name - " + agent.getName());
138	_logger.write(toString(), Logger.DEBUG, agent.getName() + "'s transmission count - " + agent.getTransmissionCount());
139
140	_logger.write(toString(), Logger.SYSMSG, "Welcoming agent");
141	agent.onArrival(AgentStation.getInstance());
142
143	_logger.write(toString(), Logger.SYSMSG, "Starting agent thread");
144
145	Thread thread = new Thread(agent);
146	thread.start();
147
148	objectInStream.close();
149	_logger.write(toString(), Logger.SYSMSG, "agent received");
150
151	} catch (ClassNotFoundException e) {
152	_logger.write(toString(), Logger.ERROR, "ERROR - possible error transferring agent bytecode - " + e);
153	}
154	}
155
156	/**
157	* Get the next inbound string from the socket and expect it
158	* to be the given text.
159	*
160	* @param expected the expected text
161	*
162	* @throws IOException on I/O error or protocol error
163	*/
164	private String getInBoundMessage(String expected) throws IOException {
165	// grab the input
166	String inBound = getInBoundMessage();
167	// check if it's what we're expecting
168	if(!inBound.equals(expected)) {
169	throw new IOException("protocol error - expected:" + expected + " got:" + inBound);
170	}
171	// it should be ok then
172	return inBound;
173	}
174
175	/**
176	* Get the next inbound string from the socket.
177	*
178	* @throws IOException on I/O error or protocol error
179	*/
180	private String getInBoundMessage() throws IOException {
181	// grab the input
182	String inBound = _socketIn.readLine();
183	// check for null's, likely disconnection
184	if(inBound == null) {
185	throw new IOException("got null from peer, maybe it died");
186	}
187	// it's a valid message it seems
188	return inBound;
189	}
190
191	//---ACCESSOR/MUTATOR METHODS---
192
193	//---ATTRIBUTES---
194
195	/**
196	* This is the friendly identifier of the
197	* component this class is running in.
198	* eg, a Filter may be called "filter1",
199	* If this class does not have an owning
200	* component, a name from the configuration
201	* can be placed here. This name could also
202	* be changed to null for utility classes.
203	*/
204	private String _name = "PeerHandler";
205
206	/**
207	* This holds a reference to the
208	* system logger that is being used.
209	*/
210	private Logger _logger = Logger.getInstance();
211
212	/**
213	* A "running" used by the main loop in the run() method to determine if
214	* it should continue running this thread.
215	*/
216	private boolean _running = true;
217
218	/**
219	* Used for the input stream of this socket
220	*/
221	private BufferedReader _socketIn;
222
223	//---STATIC ATTRIBUTES---
224
225	}