dfdbkb.java

java实现的P2P多agent中间件

			stmt = getConnectionWrapper().getConnection().createStatement();
			stmt.execute("SELECT COUNT(*) FROM "+name);
			return true;
			
		} catch (SQLException e) {
			// table doesn't exist
			return false;
		} finally {
			if (stmt != null) {
				try{
					stmt.close();
				}
				catch(SQLException see) {
					see.printStackTrace();
				}
			}  
		}
	}
	
	/**
	 * Creates a new DB table
	 * 
	 * @param name name of the table
	 * @param entries array of column and constraint specifications
	 * @throws SQLException If the table cannot be created
	 */
	protected void createTable(String name, String[] entries) {
		if (!tableExists(name)) {
			Statement stmt = null;

			try {
				Connection conn = getConnectionWrapper().getConnection();
				stmt = conn.createStatement();
				
				String sql = "CREATE TABLE " + name + " (";
				for (int i = 0; i < entries.length; i++) {
					sql += entries[i];
					if (i < entries.length - 1)
						sql += ", ";
					else
						sql += ")";
				}
				stmt.executeUpdate(sql);
				conn.commit();
				
			} catch (SQLException e) {
				if(logger.isLoggable(Logger.SEVERE)) 
					logger.log(Logger.SEVERE, "Error creating table '"+name+"'", e);
				
			} finally {
				if (stmt != null) {
					try{
						stmt.close();
					}
					catch(SQLException see) {
						see.printStackTrace();
					}
				}
			}
		}
	}
	
	/**
	 * Adds explicit indices to the database to speed up queries
	 */
	protected void createIndices() {
		Statement stmt = null;
		
		try {
			Connection conn = getConnectionWrapper().getConnection();
			stmt = conn.createStatement();
			stmt.execute("CREATE INDEX dfagentDescrIdx ON dfagentdescr( aid )");
			stmt.execute("CREATE INDEX leaseIdx ON dfagentdescr( lease )");
			stmt.execute("CREATE INDEX agentAddressIdx ON agentaddress( aid )");
			stmt.execute("CREATE INDEX agentResolverIdx ON agentresolver( aid )");
			stmt.execute("CREATE INDEX agentUserdefslotIdx ON agentuserdefslot( aid )");
			stmt.execute("CREATE INDEX serviceLanguageIdx ON servicelanguage( serviceid )"); 
			stmt.execute("CREATE INDEX serviceProtocolIdx ON serviceprotocol( serviceid )"); 
			stmt.execute("CREATE INDEX serviceOntologyIdx ON serviceontology( serviceid )"); 
			stmt.execute("CREATE INDEX servicePropertyIdx ON serviceproperty( serviceid )");
			stmt.execute("CREATE INDEX ontologyIdx ON ontology( descrid )");
			stmt.execute("CREATE INDEX protocolIdx ON ontology( descrid )");
			stmt.execute("CREATE INDEX languageIdx ON ontology( descrid )");
			conn.commit();
			
		} catch (SQLException e) {
			if(logger.isLoggable(Logger.FINE)) 
				logger.log(Logger.FINE, "Indices for DF tables couldn't be created", e);
			
		} finally {
			if (stmt != null) {
				try{
					stmt.close();
				}
				catch(SQLException see) {
					see.printStackTrace();
				}
			}
		}
	}
	
	
	/**
	 * Creates the proper DB tables.
	 */
	protected void createDFTables() {
		String LONGVARCHAR_TYPE = getLongVarCharType();
		
		// Tables for DF registrations      
		createTable(DFAGENTDESCR, new String[] {
				"id VARCHAR(" + MAX_PROP_LENGTH + ")",  
				"aid VARCHAR(" + MAX_PROP_LENGTH + ")",  
				"lease VARCHAR(20)",   
				"PRIMARY KEY( id )"}); 
		
		createTable(AGENTADDRESS, new String[] {
				"id VARCHAR(" + MAX_PROP_LENGTH + ")", 
				"aid VARCHAR(" + MAX_PROP_LENGTH + ")",   
				"address VARCHAR(" + MAX_PROP_LENGTH + ")", 
				"PRIMARY KEY( id )"});
		
		createTable(AGENTRESOLVER, new String[] {
				"id VARCHAR(" + MAX_PROP_LENGTH + ")",
				"aid VARCHAR(" + MAX_PROP_LENGTH + ")",
				"resolveraid VARCHAR(" + MAX_PROP_LENGTH + ")",
				"PRIMARY KEY( id )"});	
		
		createTable(AGENTUSERDEFSLOT, new String[] {
				"id VARCHAR(" + MAX_PROP_LENGTH + ")",
				"aid	VARCHAR(" + MAX_PROP_LENGTH + ")",
				"slotkey	VARCHAR(" + MAX_PROP_LENGTH + ")",
				"slotval	" + LONGVARCHAR_TYPE,
				"PRIMARY KEY( id )"});	 		 	
		
		createTable(ONTOLOGY, new String[] {
				"descrid VARCHAR(" + MAX_PROP_LENGTH + ")",
				"ontology VARCHAR(32)",
				"PRIMARY KEY( descrid, ontology )",
				"FOREIGN KEY( descrid ) REFERENCES dfagentdescr( id )"
		});
		
		createTable(PROTOCOL, new String[] {
				"descrid VARCHAR(" + MAX_PROP_LENGTH + ")",
				"protocol VARCHAR(32)", 
				"PRIMARY KEY( descrid, protocol )", 
				"FOREIGN KEY( descrid ) REFERENCES dfagentdescr( id )"
		});
		
		createTable(LANGUAGE, new String[] {
				"descrid VARCHAR(" + MAX_PROP_LENGTH + ")",
				"language VARCHAR(32)",
				"PRIMARY KEY( descrid, language )",
				"FOREIGN KEY( descrid ) REFERENCES dfagentdescr( id )"
		});
		
		createTable(SERVICE, new String[] {
				"id VARCHAR(" + MAX_PROP_LENGTH + ")",
				"descrid VARCHAR(" + MAX_PROP_LENGTH + ")",
				"sname VARCHAR(" + MAX_PROP_LENGTH + ")",
				"stype VARCHAR(64)",
				"sownership VARCHAR(64)",
				"PRIMARY KEY( id )",
				"FOREIGN KEY( descrid ) REFERENCES dfagentdescr( id )"
		});
		
		createTable(SERVICEPROTOCOL, new String[] {
				"serviceid VARCHAR(" + MAX_PROP_LENGTH + ")",
				"protocol VARCHAR(32)",
				"PRIMARY KEY( serviceid, protocol )",
				"FOREIGN KEY( serviceid ) REFERENCES service( id )"});
		
		createTable(SERVICEONTOLOGY, new String[] {
				"serviceid VARCHAR(" + MAX_PROP_LENGTH + ")",
				"ontology VARCHAR(32)",
				"PRIMARY KEY( serviceid, ontology )",
				"FOREIGN KEY( serviceid ) REFERENCES service( id )"});
		
		createTable(SERVICELANGUAGE, new String[] {
				"serviceid VARCHAR(" + MAX_PROP_LENGTH + ")",
				"language VARCHAR(32)",
				"PRIMARY KEY( serviceid, language )",
				"FOREIGN KEY( serviceid ) REFERENCES service( id )"});
		
		createTable(SERVICEPROPERTY, new String[] {
				"serviceid VARCHAR(" + MAX_PROP_LENGTH + ")",
				"propkey VARCHAR(" + MAX_PROP_LENGTH + ")",
				"propval_obj " + LONGVARCHAR_TYPE,
				"propval_str VARCHAR(" + MAX_PROP_LENGTH + ")",
				"propvalhash VARCHAR(100)",
				"PRIMARY KEY( serviceid, propkey )",
				"FOREIGN KEY( serviceid ) REFERENCES service( id )"});
		
		// Tables for subscriptions
		createTable(SUBSCRIPTION, new String[] { 
				"id	 VARCHAR(" + MAX_PROP_LENGTH + ")"   ,
				"aclm " + LONGVARCHAR_TYPE,
				"PRIMARY KEY( id )"});
		
		createIndices();
		
		//DEBUG
		if(logger.isLoggable(Logger.FINE))
			logger.log(Logger.FINE,"Tables correctly created");
		
	}
	
	/**
	 * Builds an error message for a <code>BatchUpdateException</code>
	 */
	private String getBatchUpdateErroMsg(BatchUpdateException e) {
		StringBuffer msg = new StringBuffer("SQLException: " + e.getMessage() + "\n");
		msg.append("SQLState:  " + e.getSQLState() + "\n");
		msg.append("Message:  " + e.getMessage() + "\n");
		msg.append("Vendor:  " + e.getErrorCode() + "\n");
		msg.append("Update counts: ");
		
		int [] updateCounts = e.getUpdateCounts();
		for (int i = 0; i < updateCounts.length; i++) {
			msg.append(updateCounts[i] + "   ");
		}
		return msg.toString();
	}
	
	/**
	 * Stores the information of an AID of a resolver
	 */
	private void saveResolverAID(AID aid, AID resolverAid) throws SQLException {
		saveAID(resolverAid);
		PreparedStatements pss = getPreparedStatements();
		pss.stm_insAgentResolverAID.setString(1, getGUID());
		pss.stm_insAgentResolverAID.setString(2, aid.getName());
		pss.stm_insAgentResolverAID.setString(3, resolverAid.getName());
		pss.stm_insAgentResolverAID.addBatch(); 
	}
	
	/**
	 * Stores the information of an AID in the database
	 */
	private void saveAID(AID aid) throws SQLException {
		String name = aid.getName();
		PreparedStatements pss = getPreparedStatements();

		// Addresses
		Iterator iter = aid.getAllAddresses();
		if (iter.hasNext()) {
			pss.stm_insAgentAddress.clearBatch();
			while( iter.hasNext()){
				pss.stm_insAgentAddress.setString(1, getGUID());
				pss.stm_insAgentAddress.setString(2, name);
				pss.stm_insAgentAddress.setString(3, (String)iter.next());
				pss.stm_insAgentAddress.addBatch();
			}
			pss.stm_insAgentAddress.executeBatch();
		}
		
		
		// User defined slots
		Properties props = aid.getAllUserDefinedSlot();
		if (props.size() > 0) {
			pss.stm_insAgentUserDefSlot.clearBatch();
			java.util.Iterator pIter = props.entrySet().iterator();
			while (pIter.hasNext()) {
				Map.Entry entry = (Map.Entry)pIter.next();
				pss.stm_insAgentUserDefSlot.setString(1, getGUID());
				pss.stm_insAgentUserDefSlot.setString(2, name);
				pss.stm_insAgentUserDefSlot.setString(3, (String)entry.getKey());
				pss.stm_insAgentUserDefSlot.setString(4, (String)entry.getValue());
				pss.stm_insAgentUserDefSlot.addBatch();
			}
			pss.stm_insAgentUserDefSlot.executeBatch();
		}
		
		// Resolvers
		iter = aid.getAllResolvers();
		if (iter.hasNext()) {
			pss.stm_insAgentResolverAID.clearBatch();
			while(iter.hasNext()){
				AID resolverAID = (AID)iter.next();
				saveResolverAID(aid, resolverAID);
			}  
			pss.stm_insAgentResolverAID.executeBatch();
		}
	}
	
	/**
	 * Returns all resolver AIDs for the given AID
	 * @return <code>Collection</code> of aid strings
	 */
	private Collection getResolverAIDs(String aid) throws SQLException {
		ArrayList res = new ArrayList();
		PreparedStatements pss = getPreparedStatements();

		pss.stm_selAgentResolverAIDs.setString(1, aid);
		ResultSet rs = pss.stm_selAgentResolverAIDs.executeQuery();
		while(rs.next()){
			res.add(rs.getString(1));
		}
		return res;
	}
	
	/**
	 * Stores a collection of services for a specific description Id in the database
	 * @param descrId id of the DFD these services belong to
	 * @param iter iterator for a collection of <code>ServiceDescription</code> instances
	 * @throws SQLException
	 */
	private void saveServices(String descrId, Iterator iter) throws SQLException {
		if (iter.hasNext()) {
			PreparedStatements pss = getPreparedStatements();
			pss.stm_insService.clearBatch();
			pss.stm_insServiceOntology.clearBatch();
			pss.stm_insServiceOntology.clearBatch();
			pss.stm_insServiceLanguage.clearBatch();
			pss.stm_insServiceProperty.clearBatch();
			
			boolean executeProtocolsBatch = false;
			boolean executeOntologiesBatch = false;
			boolean executeLanguagesBatch = false;
			boolean executePropertiesBatch = false;
			
			while(iter.hasNext()){
				ServiceDescription service = (ServiceDescription)iter.next();
				String serviceId = getGUID();
				pss.stm_insService.clearParameters();
				pss.stm_insService.setString(1, serviceId);
				pss.stm_insService.setString(2, descrId);
				pss.stm_insService.setString(3, service.getName());
				pss.stm_insService.setString(4, service.getType());
				pss.stm_insService.setString(5, service.getOwnership());
				pss.stm_insService.addBatch();
				
				// Service - Protocols
				Iterator iterS = service.getAllProtocols();
				while(iterS.hasNext()){
					pss.stm_insServiceProtocol.setString(1, serviceId);
					pss.stm_insServiceProtocol.setString(2, (String)iterS.next());
					pss.stm_insServiceProtocol.addBatch();
					executeProtocolsBatch = true;
				}
				
				// Service - Ontologies
				iterS = service.getAllOntologies();
				while(iterS.hasNext()){
					pss.stm_insServiceOntology.setString(1, serviceId);
					pss.stm_insServiceOntology.setString(2, (String)iterS.next());
					pss.stm_insServiceOntology.addBatch();
					executeOntologiesBatch = true;
				}
				
				// Service - Languages
				iterS = service.getAllLanguages();
				while(iterS.hasNext()){
					pss.stm_insServiceLanguage.setString(1, serviceId);
					pss.stm_insServiceLanguage.setString(2, (String)iterS.next());
					pss.stm_insServiceLanguage.addBatch();
					executeLanguagesBatch = true;
				}
				
				// Service - Properties
				iterS = service.getAllProperties();
				while(iterS.hasNext()){
					
					Property prop = (Property)iterS.next();
					try {
						pss.stm_insServiceProperty.setString(1, serviceId);
						pss.stm_insServiceProperty.setString(2, prop.getName());
						
						// serialize value to a string and calcualte 
						// a hash map for later search operations
						Object value = prop.getValue();
						// store plain String object value directly
						// in 'propval_str' field otherwise store it in