dfdbkb.java

java实现的P2P多agent中间件

				lWhere.add(tmp+".sownership='"+serviceOwner+"'");
			}
			lWhere.add(tmp+".descrid=dfagentdescr.id");
			i++;
			
			// Service languages
			Iterator iterS = service.getAllLanguages();
			int j = 0;
			while(iterS.hasNext()){
				String tmp1 = SERVICELANGUAGE+j;
				lAs.add(", servicelanguage "+tmp1);
				lWhere.add(tmp1+".language='"+(String)iterS.next()+"'");
				lWhere.add(tmp1+".serviceid="+tmp+".id");
				j++;
			}
			// Service ontologies
			iterS = service.getAllOntologies();
			j = 0;
			while(iterS.hasNext()){
				String tmp1 = SERVICEONTOLOGY+j;
				lAs.add(", serviceontology "+tmp1);
				lWhere.add(tmp1+".ontology='"+(String)iterS.next()+"'");
				lWhere.add(tmp1+".serviceid="+tmp+".id");
				j++;
			}
			// Service protocols
			iterS = service.getAllProtocols();
			j = 0;
			while(iterS.hasNext()){
				String tmp1 = SERVICEPROTOCOL+j;
				lAs.add(", serviceprotocol "+tmp1);
				lWhere.add(tmp1+".protocol='"+(String)iterS.next()+"'");
				lWhere.add(tmp1+".serviceid="+tmp+".id");
				j++;
			}
			// Service properties
			iterS = service.getAllProperties();
			j = 0;
			while(iterS.hasNext()){
				String tmp1 = SERVICEPROPERTY+j;
				lAs.add(", serviceproperty "+tmp1);
				Property prop = (Property) iterS.next();	
				
				if (prop.getName() != null)
					lWhere.add(tmp1+".propkey='"+prop.getName()+"'");
				
				Object value = prop.getValue();
				if (value != null) {
					if (needSerialization(value)) {
						String hashStr = getHashValue(prop.getValue());
						lWhere.add(tmp1+".propvalhash='"+ hashStr +"'");
					}
					else {
						lWhere.add(tmp1+".propval_str='"+ value +"'");
					}
				}
				lWhere.add(tmp1+".serviceid="+tmp+".id");  
				j++;
			}			
		}
		
		// Concatenate all the aliases
		iter = lAs.iterator();
		while (iter.hasNext()) {
			select.append((String) iter.next());
		}
		// Concatenate all WHERE
		if (lWhere.size() > 0) {
			select.append(" WHERE ");
		}
		iter = lWhere.iterator();
		i = 0;
		while (iter.hasNext()) {
			if(i > 0) {
				select.append(" and ");
			}
			select.append((String) iter.next());
			++i;
		}
		return select.toString();
	}
	
	////////////////////////////////////////
	// DB cleaning methods
	////////////////////////////////////////
	
	/**
	 Removes DF registrations and subscriptions whose lease time 
	 has expired.
	 This method is called at startup and each MAX_REGISTER_WITHOUT_CLEAN
	 registrations.
	 */
	private void clean(){
		cleanExpiredRegistrations();
		cleanExpiredSubscriptions();
	}
	
	/**
	 * Removes DF registrations whose lease time has expired.
	 */
	private void cleanExpiredRegistrations(){
		
		ResultSet rs = null;
		long currTime = System.currentTimeMillis();
		try{
			PreparedStatements pss = getPreparedStatements();
			pss.stm_selExpiredDescr.setString(1, String.valueOf(currTime));
			rs = pss.stm_selExpiredDescr.executeQuery();
			
			while(rs.next()){
				remove(rs.getString("aid"));
			}
		}
		catch(SQLException se){
			if(logger.isLoggable(Logger.WARNING))
				logger.log(Logger.WARNING, "Error cleaning expired DF registrations", se);
			
		} finally {
			closeResultSet(rs);
		}
	}
	
	/**
	 Removes DF subscriptions whose lease time has expired.
	 */
	private void cleanExpiredSubscriptions() {
		//FIXME: To be implemented
	}
	
	private StringACLCodec codec = new StringACLCodec();
	
	protected void subscribeSingle(Object dfd, SubscriptionResponder.Subscription s) throws SQLException, NotUnderstoodException{
		ACLMessage aclM = s.getMessage();
		String msgStr = aclM.toString();
		String convID = aclM.getConversationId();
		registerSubscription(convID, msgStr);
	}
	
	/**
	 * Add a subscription to the database
	 * @param convID conversation id (used as primary key)
	 * @param aclM ACL message for the subscription
	 */
	private void registerSubscription(String convID, String aclM) throws SQLException {
		Connection conn = getConnectionWrapper().getConnection();
		try {
			PreparedStatements pss = getPreparedStatements();
			String base64Str = new String(Base64.encodeBase64(aclM.getBytes("US-ASCII")), "US-ASCII"); 
			// --> convert string to Base64 encoding
			pss.stm_insSubscription.setString(1, convID);
			pss.stm_insSubscription.setString(2, base64Str);
			pss.stm_insSubscription.execute();
			conn.commit();			
		} 
		catch (SQLException sqle) {
			// Rollback the transaction
			try {
				conn.rollback();
			} catch (SQLException se) {
				logger.log(Logger.SEVERE,"Rollback for incomplete subscription failed.", se);
			}
			// Re-throw the exception
			throw sqle;
		}
		catch (Exception e) {
			logger.log(Logger.SEVERE, "Error encoding subscription message in Base64.", e);
			throw new SQLException("Error encoding subscription message in Base64. "+e.getMessage());
		}
	}
	
	/**
	 * Return all known subscriptions at the DF
	 * @return <code>Enumeration</code> with instances of the class 
	 * <code> jade.proto.SubscriptionResponder&Subscription</code>
	 */
	public Enumeration getSubscriptions(){
		Vector subscriptions = new Vector();
		StringACLCodec codec = new StringACLCodec();
		ResultSet rs = null;
		
		try {
			rs = getPreparedStatements().stm_selSubscriptions.executeQuery();
			while (rs.next()) {
				String base64Str = rs.getString("aclm");
				String aclmStr = new String(Base64.decodeBase64(base64Str.getBytes("US-ASCII")), "US-ASCII");
				ACLMessage aclm = codec.decode(aclmStr.getBytes(), ACLCodec.DEFAULT_CHARSET);
				subscriptions.add(sr.createSubscription(aclm));
			}
			
		} catch (Exception e) {
			if(logger.isLoggable(Logger.SEVERE))
				logger.log(Logger.SEVERE, "Error retrieving subscriptions from the database", e);
			
		} finally {
			closeResultSet(rs);
		}
		return subscriptions.elements();
	}
	
	
	protected void unsubscribeSingle(SubscriptionResponder.Subscription sub) throws SQLException {
		ACLMessage aclM = sub.getMessage();
		String convID = aclM.getConversationId();
		boolean deleted = deregisterSubscription(convID);
		if(!deleted)
			if(logger.isLoggable(Logger.WARNING))
				logger.log(Logger.WARNING,"No subscription to delete.");
	}
	
	
	/**
	 * Removes a registration from the database.
	 * @param convID id for the subscription
	 * @return <code>true</code> if an entry has been found and removed  
	 * - otherwise <code>false</code>
	 */
	private boolean deregisterSubscription(String convID) throws SQLException {

		Connection conn = getConnectionWrapper().getConnection();
		try {
			PreparedStatements pss = getPreparedStatements();
			pss.stm_delSubscription.setString(1, convID);
			int rowCount = pss.stm_delSubscription.executeUpdate();
			conn.commit();
			return (rowCount != 0);
		} catch (SQLException sqle) {
			// Rollback the transaction
			try {
				conn.rollback();
			} catch (SQLException se) {
				logger.log(Logger.SEVERE,"Rollback for incomplete un-subscription failed.", se);
			}
			// Re-throw the exception
			throw sqle;
		}
	}
	
	
	////////////////////////////////////////////////
	// Helper methods
	////////////////////////////////////////////////
	
	
	/**
	 * Closes an open result set and logs an appropriate error message
	 * when it fails
	 */
	private void closeResultSet(ResultSet rs) {
		try {
			if (rs != null) {
				rs.close();
				rs = null;
			}
		} catch (SQLException e) {
			// result set has already been closed. 
			//(depends party on the database driver)
		}
	}
	
	/**
	 * Closes an open SQL statement and logs an appropriate error message
	 * when it fails
	 */
	private void closeStatement(Statement s) {
		try {
			if (s != null) {
				s.close();
				s = null;
			}
		} catch (Exception e) {
			if(logger.isLoggable(Logger.WARNING))
				logger.log(Logger.WARNING,"Closing SQL statement failed.");
		}
	}
	
	/**
	 * Serializes any serializable object to a Base64 encoded string
	 * @param obj the object to serialize
	 * @throws IOException An error during the serialization
	 */
	private String serializeObj(Object obj) throws IOException {
		if (obj == null)
			return null;
		
		ByteArrayOutputStream baos = new ByteArrayOutputStream();
		ObjectOutputStream oos = new ObjectOutputStream(baos);
		oos.writeObject(obj);
		oos.close();
		byte[] data = baos.toByteArray(); 
		return new String(Base64.encodeBase64(data), "US-ASCII");
	}
	
	/**
	 * Deserializes any serializable object from a string
	 * @param str string which represents a serialized object
	 * @throws IOException An error during the serialization
	 * @throws ClassNotFoundException The deserialized java class is unknown
	 */
	private Object deserializeObj(String str) throws IOException, ClassNotFoundException {
		if (str == null)
			return null;
		byte[] data = Base64.decodeBase64(str.getBytes("US-ASCII"));
		ByteArrayInputStream bais = new ByteArrayInputStream(data);
		ObjectInputStream ois = new ObjectInputStream(bais);
		return ois.readObject();
	}
	
	/**
	 * Returns an MD5 hash value for an object
	 * @param obj Object to use for the hash value calcualtion
	 * @return an MD5 value encoded in ISO-8859-1
	 * @throws Exception The hash value couldn't be generated
	 */
	protected String getHashValue(Object obj) throws Exception {
		final String HASH_ALGORITHM = "MD5";
		
		if (obj == null)
			return "null";
		
		try {
			ByteArrayOutputStream baos = new ByteArrayOutputStream();
			ObjectOutputStream oos = new ObjectOutputStream(baos);
			oos.writeObject(obj);
			oos.close();
			byte[] data = baos.toByteArray();
			MessageDigest md = MessageDigest.getInstance(HASH_ALGORITHM);
			byte[] digest = md.digest(data);
			return new String(Base64.encodeBase64(digest), "US-ASCII");
			
		} catch (Exception e) {
			throw new Exception("Couldn't create " + HASH_ALGORITHM + " hash for given object.", e);
		}
	}
	
	/**
	 * This method must be used when inserting text values into a DB. It doubles all ' and " characters
	 * If the passed parameter is null, then an empty string is returned.
	 **/
	protected String prepDBStr(String s) {
		if (s == null){
			return "";
		}
		String result = replace(s,"'","''");
		result = replace(result,"\"","\"\"" );
		
		return result;
	}
	
	/**
	 * Replaces all occurences of a <code>pattern</code> in <code>str</code>
	 * with <code>replaceWith</code>
	 * @param str source string
	 * @param pattern pattern string to search for
	 * @param replaceWith new string
	 */
	protected String replace(String str, String pattern, String replaceWith) {
		int s = 0;
		int e = 0;
		StringBuffer result = new StringBuffer();
		
		while ((e = str.indexOf(pattern, s)) >= 0) {
			result.append(str.substring(s, e));
			if(replaceWith != null)
				result.append(replaceWith);
			s = e+pattern.length();
		}
		result.append(str.substring(s));
		return result.toString();
	}
}