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, 7 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
#	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			* @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			}