taskserver.java

nesC写的heed算法

					outStream.writeObject(moteClientInfo);
				}
				catch (Exception e)
				{
					outStream.writeObject(null);
					e.printStackTrace();
				}
				outStream.flush();
				return true;
			}
		case ADD_MOTE:
			{
				TASKMoteClientInfo moteClientInfo = null;
				try
				{
					moteClientInfo = (TASKMoteClientInfo)inStream.readObject();
				}
				catch (Exception e)
				{
					e.printStackTrace();
				}
				PreparedStatement ps = null;
				try
				{
					ps = dbConn.prepareStatement("INSERT INTO task_mote_info values (?, ?, ?, ?, NULL, ?, ?)");
					ps.setInt(1, moteClientInfo.moteId);
					ps.setDouble(2, moteClientInfo.xCoord);
					ps.setDouble(3, moteClientInfo.yCoord);
					ps.setDouble(4, moteClientInfo.zCoord);
					ps.setBytes(5, moteClientInfo.data);
					ps.setString(6, moteClientInfo.clientInfoName);
					ps.executeUpdate();
					ps.close();
					outStream.writeInt(TASKError.SUCCESS);
				}
				catch (SQLException e)
				{
					if (ps != null)
					{
						try
						{
							ps.close();
						}
						catch (Exception e1)
						{
							e1.printStackTrace();
						}
					}
					outStream.writeInt(TASKError.FAIL);
					e.printStackTrace();
				}
				outStream.flush();
				return true;
			}
		case DELETE_MOTE:
			{
				int moteId = inStream.readInt();
				PreparedStatement ps = null;
				try
				{
					ps = dbConn.prepareStatement("DELETE FROM task_mote_info WHERE mote_id = ?");
					ps.setInt(1, moteId);
					ps.executeUpdate();
					ps.close();
					outStream.writeInt(TASKError.SUCCESS);
				}
				catch (SQLException e)
				{
					if (ps != null)
					{
						try
						{
							ps.close();
						}
						catch (Exception e1)
						{
						}
					}
					outStream.writeInt(TASKError.FAIL);
					e.printStackTrace();
				}
				outStream.flush();
				return true;
			}
		case DELETE_MOTES:
			{
				String clientInfoName = null;
				try
				{
					clientInfoName = (String)inStream.readObject();
				}
				catch (Exception e)
				{
					e.printStackTrace();
				}
				PreparedStatement ps = null;
				try
				{
					ps = dbConn.prepareStatement("DELETE FROM task_mote_info WHERE clientinfo_name = ?");
					ps.setString(1, clientInfoName);
					ps.executeUpdate();
					ps.close();
					outStream.writeInt(TASKError.SUCCESS);
				}
				catch (SQLException e)
				{
					if (ps != null)
					{
						try
						{
							ps.close();
						}
						catch (Exception e1)
						{
						}
					}
					outStream.writeInt(TASKError.FAIL);
					e.printStackTrace();
				}
				outStream.flush();
				return true;
			}
		case GET_ALLMOTECLIENTINFO:
			{
				try
				{
					String clientinfoName = (String)inStream.readObject();
					ResultSet rs = dbStmt.executeQuery("SELECT mote_id, clientinfo_name, x_coord, y_coord, z_coord, moteinfo, clientinfo_name FROM task_mote_info where clientinfo_name = '" + clientinfoName + "'");
					Vector moteClientInfos = new Vector();
					while (rs.next())
					{
						TASKMoteClientInfo moteClientInfo = new TASKMoteClientInfo(rs.getInt(1), rs.getDouble(3), rs.getDouble(4), rs.getDouble(5), rs.getBytes(6), rs.getString(2));
						moteClientInfos.add(moteClientInfo);
					}
					rs.close();
					outStream.writeObject(moteClientInfos);
				}
				catch (Exception e)
				{
					outStream.writeObject(null);
					e.printStackTrace();
				}
				outStream.flush();
				return true;
			}
		case GET_SERVERCONFIGINFO:
			{
				TASKServerConfigInfo serverConfigInfo;
				serverConfigInfo = new TASKServerConfigInfo(TinyDBMain.groupid,
											urlPSQL, dbUser, dbPwd,
											TinyDBMain.sfHost,
											TinyDBMain.sfPort,
											TinyDBMain.sfCommPort);
				outStream.writeObject(serverConfigInfo);
				outStream.flush();
				return true;
			}
		case STOP_QUERY:
			{
				short whichQuery = inStream.readShort();
				int queryId;
				TinyDBQuery q;

				if (whichQuery == SENSOR_QUERY)
				    {
					if (sensorTinyDBQuery == null || sensorQuery == null)
					{
				    outStream.writeInt(TASKError.SUCCESS);
					outStream.flush();
					return true;
					}
					q = sensorTinyDBQuery;
					queryId = sensorQuery.getQueryId();
				    }
				else
				    {
					if (healthTinyDBQuery == null || healthQuery == null)
					{
				    outStream.writeInt(TASKError.SUCCESS);
					outStream.flush();
					return true;
					}
					q = healthTinyDBQuery;
					queryId = healthQuery.getQueryId();
				    }
				boolean ok = stopRunningQuery(q,queryId);
				if (ok) {
				    outStream.writeInt(TASKError.SUCCESS);
				} else {
				    outStream.writeInt(TASKError.FAIL);
				}
				outStream.flush();
				return true;
			}
		case RUN_COMMAND:
			{
				TASKCommand command = null;
				try
				{
					command = (TASKCommand)inStream.readObject();
				}
				catch (Exception e)
				{
					e.printStackTrace();
				}
				TASKCommandInfo commandInfo = null;
				boolean found = false;
				for (Iterator it = commandInfos.iterator(); it.hasNext(); )
				{
					commandInfo = (TASKCommandInfo)it.next();
					if (commandInfo.getCommandName().equalsIgnoreCase(command.getCommandName()))
					{
						found = true;
						break;
					}
				}
				int error;
				if (!found)
				{
					error = TASKError.INVALID_COMMAND;
				}
				else
				{
					Message cmdMessage = command.getTinyOSMessage(commandInfo);
					if (cmdMessage != null)
					    error = sendTinyOSMessage(cmdMessage);
					else
					    System.out.println("Invalid command.");
				}
				String cmdStr = command.toString(commandInfo);
				PreparedStatement ps = null;
				try
				{
					ps = dbConn.prepareStatement("INSERT INTO task_command_log VALUES (?, now(), ?)");
					ps.setInt(1, nextCommandId());
					ps.setString(2, cmdStr);
					ps.executeUpdate();
					ps.close();
					outStream.writeInt(TASKError.SUCCESS);
				}
				catch (SQLException e)
				{
					if (ps != null)
					{
						try
						{
							ps.close();
						}
						catch (Exception e1)
						{
						}
					}
					outStream.writeInt(TASKError.FAIL);
					e.printStackTrace();
				}
				outStream.flush();
				return true;
			}
		case RUN_CALIBRATION:
			{
				if (calibQueryInProgress)
				{
					System.out.println("calibration query in progress.");
					outStream.writeInt(TASKError.SUCCESS);
					outStream.flush();
					return true;
				}
				else
					calibQueryInProgress = true;
				int error = TASKError.SUCCESS;
				PreparedStatement ps = null;
				if (calibTinyDBQuery == null)
				{
					String queryStr = "SELECT nodeid, prcalib SAMPLE PERIOD 2000";
					try 
					{
						calibTinyDBQuery = SensorQueryer.translateQuery(queryStr, CALIBRATION_TINYDB_QID);
					}
					catch (Exception e)
					{
						e.printStackTrace();
						error = TASKError.INVALID_QUERY;
						outStream.writeInt(error);
						outStream.flush();
						calibQueryInProgress = false;
						return true;
					}
					try
					{
						ps = dbConn.prepareStatement("INSERT INTO task_query_log VALUES (?, ?, ?, ?, ?)");
						ps.setInt(1, CALIBRATION_QUERY_ID);
						ps.setShort(2, (short)CALIBRATION_TINYDB_QID);
						ps.setString(3, queryStr);
						ps.setString(4, "calibration");
						ps.setString(5, "task_mote_info");
						ps.executeUpdate();
						ps.close();
					}
					catch (SQLException e)
					{
						// ignore exception here because the calibration
						// query might have already been logged
						try
						{
							ps.close();
						}
						catch (Exception e1)
						{
						}
					}
					try
					{
						ps = dbConn.prepareStatement("INSERT INTO task_query_time_log (query_id, start_time) VALUES (?, now())");
						ps.setInt(1, CALIBRATION_QUERY_ID);
						ps.executeUpdate();
						ps.close();
					}
					catch (SQLException e)
					{
						try
						{
							dbConn.setAutoCommit(true);
							ps.close();
						}
						catch (Exception e1)
						{
						}
						outStream.writeInt(TASKError.FAIL);
						e.printStackTrace();
						outStream.flush();
						calibQueryInProgress = false;
						return true;
					}
				}
				else
				{
					try
					{
						ps = dbConn.prepareStatement("UPDATE task_query_time_log SET stop_time = now() WHERE query_id = ? AND stop_time IS NULL");
						ps.setInt(1, CALIBRATION_QUERY_ID);
						ps.executeUpdate();
						ps.close();
						ps = dbConn.prepareStatement("INSERT INTO task_query_time_log (query_id, start_time) VALUES (?, now())");
						ps.setInt(1, CALIBRATION_QUERY_ID);
						ps.executeUpdate();
						ps.close();
					}
					catch (SQLException e)
					{
						try
						{
							ps.close();
						}
						catch (Exception e1)
						{
						}
						outStream.writeInt(TASKError.FAIL);
						e.printStackTrace();
						outStream.flush();
						calibQueryInProgress = false;
						return true;
					}
				}
				calibNodes = new Vector();
				try
				{
					ResultSet rs = dbStmt.executeQuery("SELECT distinct mote_id FROM task_mote_info WHERE calib IS NULL and mote_id > 0");
					while (rs.next())
					{
						calibNodes.add(new Integer(rs.getInt(1)));
					}
					rs.close();
				}
				catch (Exception e)
				{
					e.printStackTrace();
					error = TASKError.FAIL;
					outStream.writeInt(error);
					outStream.flush();
					calibQueryInProgress = false;
					return true;
				}
				if (calibNodes.isEmpty()) {
				    System.out.println("Already have calibration data for all nodes.");
				    try {
					outStream.writeInt(TASKError.SUCCESS);
					outStream.flush();
				    } catch (IOException e) {
					//oh well...
				    }
				    calibQueryInProgress = false;
				    return true;
				}
				try
				{
					int cnt = 0;
					while (calibQueryInProgress)
					{
						if (cnt == 15)
						{
							System.out.println("calibration query timed out.");
							error = TASKError.FAIL;
							outStream.writeInt(error);
							outStream.flush();
							calibQueryInProgress = false;
							stopRunningQuery(calibTinyDBQuery, CALIBRATION_QUERY_ID);
							return true;
						}
						System.out.println("sending calibration query.");
						TinyDBMain.injectQuery(calibTinyDBQuery, this);
						Thread.currentThread().sleep(4000);
						cnt++;
					}
				}
				catch (Exception e)
				{
					e.printStackTrace();
					error = TASKError.INVALID_QUERY;
					outStream.writeInt(error);
					outStream.flush();
					calibQueryInProgress = false;
					return true;
				}
				System.out.println("calibration query injected.");
				outStream.writeInt(TASKError.SUCCESS);
				outStream.flush();
				return true;
			}
		}
		return true;
    }

	private int sendTinyOSMessage(Message msg)
	{
		int error = TASKError.SUCCESS;
		try
		{
			System.out.print(msg);
			System.out.println("");
			TinyDBMain.mif.send(TASKCommand.BROADCAST_ID, msg);
		}
		catch (Exception e)
		{
			System.out.println("Error sending TinyOS message.");
			e.printStackTrace();
			error = TASKError.TINYOS_MESSAGE_SEND_FAILED;
		}
		return error;
	}

    /** ResultListener method called when a query returns are result
	TASKServer maps these onto its own set of result listener
	registered through the socket interface.
    */
    public void addResult(QueryResult qr) 
	{
		String tableName = null;
		int queryId = -1;
		if (sensorTinyDBQuery != null &&
			sensorTinyDBQuery.getId() == qr.qid())
		{
			tableName = sensorQuery.getTableName();
			queryId = sensorQuery.getQueryId();
		}
		else if (healthQuery != null &&
				 healthTinyDBQuery.getId() == qr.qid())
		{
			tableName = healthQuery.getTableName();
			queryId = healthQuery.getQueryId();
		}
		else if (qr.qid() == CALIBRATION_TINYDB_QID)
		{
			handleCalibrationResult(qr);
			return;
		}
		if (tableName == null)
			return;
		String insertStmt = DBLogger.insertStmt(qr, tableName);
		System.out.println("insert statement: " + insertStmt);
		try
		{
			if (dbConn == null)
			{
				// try to reconnect to database
				dbConn = DriverManager.getConnection(urlPSQL, dbUser, dbPwd);
				dbStmt = dbConn.createStatement();
			}
			dbStmt.executeUpdate(insertStmt);
		}
		catch (Exception e)
		{
			try
			{
				if (dbConn != null)
					dbConn.close();
			}
			catch (Exception e1)
			{
			}
			dbConn = null;
			e.printStackTrace();
		}
		Vector ls = (Vector)listeners.get(new Integer(queryId));
		if (ls != null)
		{
			Iterator it = ls.iterator();
			while (it.hasNext()) 
			{
				ObjectOutputStream outStream = (ObjectOutputStream)it.next();
				try 
				{
					// XXX repeatedly sending fieldinfos is very wasteful!
					Vector fieldInfos = new Vector();
					Vector fieldValues = qr.getFieldValueObjs();
					TinyDBQuery q = qr.getQuery();
					for (int i = 0; i < fieldValues.size(); i++)
					{
						QueryField qf = q.getField(i);
						fieldInfos.add(new TASKFieldInfo(qf));
					}
					fieldValues.add(new Integer(qr.epochNo()));
 					fieldInfos.add(new TASKFieldInfo("epoch", TASKTypes.UINT16));

					TASKResult result = new TASKResult(queryId, fieldValues, fieldInfos);
					outStream.writeObject(result);
					outStream.flush();
					System.out.println("TASKResult sent "+qr.epochNo());
				}
				catch (IOException e) 
				{
					System.out.println("Removing listener.");
					it.remove(); //delete this listener, since it died
					e.printStackTrace();
				}
			}