taskclient.java

来自「nesC写的heed算法」· Java 代码 · 共 1,198 行 · 第 1/2 页

			outStream.writeShort(TASKServer.GET_ALLMOTECLIENTINFO);
			outStream.writeObject(clientinfoName);
			outStream.flush();
			moteClientInfos = (Vector)inStream.readObject();
			outStream.close();
			inStream.close();
			conn.close();
		}
		catch (Exception e)
		{
			try
			{
				if (conn != null)
					conn.close();
				if (outStream != null)
					outStream.close();
				if (inStream != null)
					inStream.close();
			}
			catch (Exception ex)
			{
			}
			e.printStackTrace();
		}
		return moteClientInfos;
	};
	/**
	 * Get health related information about a mote.
	 *
	 * @param	moteId	mote id.
	 */
	public TASKMoteInfo getMoteInfo(int moteId) 
	{
		// XXX to be completed later
		return null;
	};
	/**
	 * Get health reated information about all motes.
	 * Call this method to figure out which motes have disappeared
	 * from the network, or have low batteries, etc.
	 */
	public TASKMoteInfo[] getAllMoteInfo() 
	{
		// XXX to be completed later
		return null;
	};
	/**
	 * Get all attribute information.  At TASKClient start time,
	 * it pre-fetches all the attribute information from the TASKServer
	 * and saves them in attributeInfos.
	 *
	 * @return an array of attribute information.
	 */
	public Vector getAttributes() 
	{
		return attributeInfos;
	};
	/**
	 * Returns information about the named attribute.
	 *
	 * @param	name	attribute name.
	 * @return	information about the attribute.
	 */
	public TASKAttributeInfo getAttribute(String name) 
	{
		for (int i = 0; i < attributeInfos.size(); i++)
		{
			TASKAttributeInfo attrInfo = (TASKAttributeInfo)attributeInfos.elementAt(i);
			if (attrInfo.name.equalsIgnoreCase(name))
				return attrInfo;
		}
		return null;
	};
	/**
	 * Get all command information.  At TASKClient start time,
	 * it pre-fetches all the command information from the TASKServer
	 * and saves them in commandInfos.
	 *
	 * @return an array of command information.
	 */
	public Vector getCommands() 
	{
		return commandInfos;
	};
	/**
	 * Returns information about the named command.
	 *
	 * @param	name 	command name
	 * @return	command information
	 */
	public TASKCommandInfo getCommand(String name) 
	{
		for (int i = 0; i < commandInfos.size(); i++)
		{
			TASKCommandInfo cmdInfo = (TASKCommandInfo)commandInfos.elementAt(i);
			if (cmdInfo.getCommandName().equalsIgnoreCase(name))
				return cmdInfo;
		}
		return null;
	};
	/**
	 * Get information about all supported aggregates.  At TASKClient start time
	 * it pre-fetches all the aggregate information from the TASKServer
	 * and saves them in aggregateInfos.
	 *
	 * @return an array of aggregate information.
	 */
	public Vector getAggregates() 
	{
		return aggregateInfos;
	};
	/**
	 * Returns information about the named aggregate.
	 *
	 * @param	name	aggregate name.
	 * @return	aggregate information.
	 */
	public TASKAggInfo getAggregate(String name) 
	{
		for (int i = 0; i < aggregateInfos.size(); i++)
		{
			TASKAggInfo aggInfo = (TASKAggInfo)aggregateInfos.elementAt(i);
			if (aggInfo.getName().equalsIgnoreCase(name))
				return aggInfo;
		}
		return null;
	};
	/**
	 * Get basic configuration information about TASKServer.
	 *
	 * @return configuration information about TASKServer.
	 */
	public TASKServerConfigInfo getServerConfigInfo() 
	{
		Socket conn = null;
		ObjectOutputStream outStream = null;
		ObjectInputStream inStream = null;
		TASKServerConfigInfo serverConfigInfo = null;
		try
		{
			conn = getConnection();
			outStream = new ObjectOutputStream(conn.getOutputStream());
			inStream = new ObjectInputStream(conn.getInputStream());
			outStream.writeShort(TASKServer.GET_SERVERCONFIGINFO);
			outStream.flush();
			serverConfigInfo  = (TASKServerConfigInfo)inStream.readObject();
			outStream.close();
			inStream.close();
			conn.close();
		}
		catch (Exception e)
		{
			try
			{
				if (conn != null)
					conn.close();
				if (outStream != null)
					outStream.close();
				if (inStream != null)
					inStream.close();
			}
			catch (Exception ex)
			{
			}
			e.printStackTrace();
		}
		return serverConfigInfo;
	};
	/**
	 * Submit sensor query to the sensor network via the TASKServer.
	 *
	 * @param query	the query to be submitted
	 * @return an error code defined in TASKError
	 */
	public int submitSensorQuery(TASKQuery query) 
	{
		return sendQuery(query, TASKServer.SENSOR_QUERY);
	};
	/**
	 * Stop the current sensor query.
	 * @return true if successful, false otherwise
	 */
	public int stopSensorQuery() 
	{
		return stopQuery(TASKServer.SENSOR_QUERY);
	};
	/**
	 * Submit health monitoring query to the sensor network via the TASKServer.
	 * For now, we will restrict that the health query's sample period must
	 * be multiples of the sensor query's sample period.
	 *
	 * @param query	The query to be submitted
	 * @return an error code defined in TASKError
	 */
	public int submitHealthQuery(TASKQuery query)
	{
		return sendQuery(query, TASKServer.HEALTH_QUERY);
	};
	/**
	 * Stop the current heath query.
	 * @return true if successful, false otherwise
	 */
	public int stopHealthQuery() 
	{
		return stopQuery(TASKServer.HEALTH_QUERY);
	};
	/**
	 * Add a listener which will be called upon arrival
	 * of any result from the sensor query.
	 * Note that all listeners will be called in the same thread.
	 * Therefore they should not perform heavy computation.
	 *
	 * @param listener	a result listener.
	 */
	public int addSensorResultListener(TASKResultListener listener) 
	{
		return addListener(TASKServer.SENSOR_QUERY, listener);
	};
	/**
	 * Add a listener which will be called upon arrival
	 * of any result from the network health monitoring query.
	 * Note that all listeners will be called in the same thread.
	 * Therefore they should not perform heavy computation.
	 *
	 * @param listern	a result listener
	 */
	public int addHealthResultListener(TASKResultListener listener) 
	{
		return addListener(TASKServer.HEALTH_QUERY, listener);
	};
	/**
	 * Submit a command to the sensor network via the TASKServer.
	 * For now, we are not handling return results of commands.
	 *
	 * @param	command	a TASK command
	 * @return	an error code defined in TASKError
	 */
	public int submitCommand(TASKCommand command)
	{
		Socket conn = null;
		ObjectOutputStream outStream = null;
		ObjectInputStream inStream = null;
		TASKServerConfigInfo serverConfigInfo = null;
		int error = TASKError.SOCKET_IO_EXCEPTION;
		try
		{
			conn = getConnection();
			outStream = new ObjectOutputStream(conn.getOutputStream());
			inStream = new ObjectInputStream(conn.getInputStream());
			outStream.writeShort(TASKServer.RUN_COMMAND);
			outStream.writeObject(command);
			outStream.flush();
			error  = inStream.readInt();
			outStream.close();
			inStream.close();
			conn.close();
		}
		catch (Exception e)
		{
			try
			{
				if (conn != null)
					conn.close();
				if (outStream != null)
					outStream.close();
				if (inStream != null)
					inStream.close();
			}
			catch (Exception ex)
			{
			}
			e.printStackTrace();
		}
		return error;
	}

	/**
	 * run the calibration query until sensor calibration coefficients
	 * are collected from every mote in the network.
	 *
	 * @return	an error code defined in TASKError
	 */
	public int collectCalibration()
	{
		Socket conn = null;
		ObjectOutputStream outStream = null;
		ObjectInputStream inStream = null;
		TASKServerConfigInfo serverConfigInfo = null;
		int error = TASKError.SOCKET_IO_EXCEPTION;
		try
		{
			conn = getConnection();
			outStream = new ObjectOutputStream(conn.getOutputStream());
			inStream = new ObjectInputStream(conn.getInputStream());
			outStream.writeShort(TASKServer.RUN_CALIBRATION);
			outStream.flush();
			error  = inStream.readInt();
			outStream.close();
			inStream.close();
			conn.close();
		}
		catch (Exception e)
		{
			try
			{
				if (conn != null)
					conn.close();
				if (outStream != null)
					outStream.close();
				if (inStream != null)
					inStream.close();
			}
			catch (Exception ex)
			{
			}
			e.printStackTrace();
		}
		return error;
	}

    /** 
	 * Return a socket allowing communication with the server
	 *
	 * @throws IOException If the server is unavailable
     */
    Socket getConnection() throws IOException 
	{
		try 
		{
			return new Socket(InetAddress.getByName(serverIP), serverPort);
		} 
		catch(UnknownHostException e) 
		{
	    	throw new IOException("Invalid TASK Server at " + serverIP + ":" + serverPort);
		}
    }

    /** Send a query to the server
	@param query The text of the query
	@param whichQuery is it a sensor query or health query?
	@return an error code defined in TASKError
    */
    private int sendQuery(TASKQuery query, short whichQuery) 
	{
		Socket conn = null;
		ObjectOutputStream outStream = null;
		ObjectInputStream inStream = null;
		TASKServerConfigInfo serverConfigInfo = null;
		int error = TASKError.SOCKET_IO_EXCEPTION;
		try
		{
			conn = getConnection();
			outStream = new ObjectOutputStream(conn.getOutputStream());
			inStream = new ObjectInputStream(conn.getInputStream());
			outStream.writeShort(TASKServer.RUN_QUERY);
			outStream.writeObject(query);
			outStream.writeShort(whichQuery);
			outStream.flush();
			error  = inStream.readInt();
			outStream.close();
			inStream.close();
			conn.close();
		}
		catch (Exception e)
		{
			try
			{
				if (conn != null)
					conn.close();
				if (outStream != null)
					outStream.close();
				if (inStream != null)
					inStream.close();
			}
			catch (Exception ex)
			{
			}
			e.printStackTrace();
		}
		return error;
    }

	private int stopQuery(short whichQuery)
	{
		Socket conn = null;
		ObjectOutputStream outStream = null;
		ObjectInputStream inStream = null;
		TASKServerConfigInfo serverConfigInfo = null;
		int error = TASKError.SOCKET_IO_EXCEPTION;
		try
		{
			conn = getConnection();
			outStream = new ObjectOutputStream(conn.getOutputStream());
			inStream = new ObjectInputStream(conn.getInputStream());
			outStream.writeShort(TASKServer.STOP_QUERY);
			outStream.writeShort(whichQuery);
			outStream.flush();
			error  = inStream.readInt();
			outStream.close();
			inStream.close();
			conn.close();
		}
		catch (Exception e)
		{
			try
			{
				if (conn != null)
					conn.close();
				if (outStream != null)
					outStream.close();
				if (inStream != null)
					inStream.close();
			}
			catch (Exception ex)
			{
			}
			e.printStackTrace();
		}
		return error;
	}

	private int addListener(short whichQuery, TASKResultListener listener)
	{
		Socket conn = null;
		ObjectOutputStream outStream = null;
		ObjectInputStream inStream = null;
		ListenerHandler handler;
		int error = TASKError.SOCKET_IO_EXCEPTION;

		if (whichQuery == TASKServer.SENSOR_QUERY)
			handler = sensorListenerHandler;
		else
			handler = healthListenerHandler;
		handler.addListener(listener);
		if (handler.getSocket() != null)
		{
			return TASKError.SUCCESS;
		}

		// add listener in the server
		try
		{
			conn = getConnection();
			System.out.println("got listener socket.");
			outStream = new ObjectOutputStream(conn.getOutputStream());
			System.out.println("got listener output stream.");
			inStream = new ObjectInputStream(conn.getInputStream());
			System.out.println("got listener input stream.");
			outStream.writeShort(TASKServer.ADD_LISTENER);
			outStream.writeShort(whichQuery);
			outStream.flush();
			System.out.println("server add listener request sent.");
			error  = inStream.readInt();
			if (error == TASKError.SUCCESS)
				handler.setSocket(conn, inStream);
			else
			{
				System.out.println("server add listener request failed.");
				outStream.close();
				inStream.close();
				conn.close();
			}
		}
		catch (Exception e)
		{
			try
			{
				if (conn != null)
					conn.close();
				if (outStream != null)
					outStream.close();
				if (inStream != null)
					inStream.close();
			}
			catch (Exception ex)
			{
			}
			e.printStackTrace();
		}
		return error;
	}

    /** A test for server connection
	Requires two parameters:
	- query to run
	- hostname of the server 
	Runs the query, registers as a listener for it, and prints out results
    public static void main(String[] argv) {
	if (argv.length != 2)
	    System.out.println("Invalid argument list\n usage:\n TASKServerConn query host");
	else {
	    try {
		String host = argv[1];
		String query = argv[0];
		Socket results;
		TASKServerConn serv = new TASKServerConn(host, TASKServer.getServerPort());
		boolean ok;
		String result;
		
		ok = serv.sendQuery(query, 0);
		if (ok) {
		    System.out.println("Sent query successfully!");
		    results = serv.registerListener(0);
		    BufferedReader br = new BufferedReader(new InputStreamReader(results.getInputStream()));
		    while (true) {
			while ((result = br.readLine()) != null) {
			    System.out.println("Read result: " + result);
			}
		    }
		} else {
		    System.out.println("Send query failed.");
		}
	    
	    } catch (IOException e) {
		System.out.println("Error: " + e);
	    }
	}
    }    
	*/

	private Vector	attributeInfos;	// pre-fetched attribute info's
	private Vector	commandInfos;	// pre-fetched command info's 
	private Vector	aggregateInfos;	// pre-fetched aggregate info's
	private String	serverIP;	// server ip address
	private int		serverPort; // server port number
	private ListenerHandler	sensorListenerHandler;
	private ListenerHandler	healthListenerHandler;
};

/**
 * ListenerHandler: internal class responsible for listening on a socket
 * for results then call all registered listeners on the socket once
 * a result is received.
 */
class ListenerHandler implements Runnable
{
	public ListenerHandler()
	{
		this.inStream = null;
		this.socket = null;
		this.resultListeners = new Vector();
	}

	public void run()
	{
		TASKResult result;
		try
		{
			while ((result = (TASKResult)inStream.readObject()) != null)
			{
				for (Iterator it = resultListeners.iterator(); it.hasNext(); )
				{
					TASKResultListener listener = (TASKResultListener)it.next();
					listener.addResult(result);
				}
			}
			// end of stream is reached, clean up
			inStream.close();
			socket.close();
		}
		catch (Exception e)
		{
			e.printStackTrace();
		}
		socket = null;
		inStream = null;
	}

	public void addListener(TASKResultListener listener)
	{
		resultListeners.add(listener);
	}

	public void setSocket(Socket sock, ObjectInputStream inStream)
	{
		Thread t = new Thread(this);
		this.socket = sock;
		try
		{
			this.inStream = inStream;
		}
		catch (Exception e)
		{
			e.printStackTrace();
		}
		t.start();
	}

	public Socket getSocket()
	{
		return socket;
	}

	private Socket				socket;
	private ObjectInputStream	inStream;
	private Vector				resultListeners;
}