dfdbkb.java

java实现的P2P多agent中间件

						// 'propval_obj' field and fill hash field (this will be used in search phase to allow matching Serializable objects)
						if ( needSerialization(value) ) {
							//System.out.println("DF Handling Object property "+prop.getName()+": value = "+value);
							String valueStr = serializeObj(value);
							pss.stm_insServiceProperty.setString(3, valueStr);
							pss.stm_insServiceProperty.setString(4, null);
							String hashStr = getHashValue(value);
							pss.stm_insServiceProperty.setString(5, hashStr);
						}
						else {
							// set to NULL the serialized representation of the object and its hash
							//System.out.println("DF Handling String property "+prop.getName()+": value = "+value);
							pss.stm_insServiceProperty.setString(3, null);
							pss.stm_insServiceProperty.setString(4, (String) value);
							pss.stm_insServiceProperty.setString(5, null);
						};
						
						pss.stm_insServiceProperty.addBatch();
						executePropertiesBatch = true;            
					} catch (Exception e) {
						if(logger.isLoggable(Logger.SEVERE))
							logger.log(Logger.SEVERE,"Cannot serialize property '" + prop.getName() + 
									"' for service '" + service.getName() + "'", e);
					}
				}
			}
			pss.stm_insService.executeBatch();
			if (executeProtocolsBatch) {
				pss.stm_insServiceProtocol.executeBatch();
			}
			if (executeOntologiesBatch) {
				pss.stm_insServiceOntology.executeBatch();
			}
			if (executeLanguagesBatch) {
				pss.stm_insServiceLanguage.executeBatch();
			}
			if (executePropertiesBatch) {
				pss.stm_insServiceProperty.executeBatch();
			}
		}
	}
	
	private static final boolean needSerialization(Object value) {
		return !((value instanceof String) && ( ((String) value).length() <= MAX_PROP_LENGTH ));		
	}
	
	/**
	 *  Insert a new DFD object.
	 *  @return the previous DFD (if any) corresponding to the same AID
	 */
	protected Object insertSingle(Object name, Object fact) throws SQLException {
		DFAgentDescription dfd = (DFAgentDescription) fact;
		AID agentAID = dfd.getName();
		String agentName = agentAID.getName();
		DFAgentDescription dfdToReturn = null;
		String batchErrMsg = "";
		
		Connection conn = getConnectionWrapper().getConnection();
		PreparedStatements pss = getPreparedStatements();
		try {
			// -- Remove the previous DFD if any
			dfdToReturn = (DFAgentDescription) removeSingle(dfd.getName());

			// -- add new DFD

			// DF Agent Description
			Date leaseTime = dfd.getLeaseTime();
			long lt = (leaseTime != null ? leaseTime.getTime() : -1);
			String descrId = getGUID();

			pss.stm_insAgentDescr.setString(1, descrId);
			pss.stm_insAgentDescr.setString(2, agentName);
			pss.stm_insAgentDescr.setString(3, String.valueOf(lt));
			pss.stm_insAgentDescr.executeUpdate();
			
			
			// AID
			saveAID(agentAID);
			
			// Languages
			Iterator iter = dfd.getAllLanguages();
			if (iter.hasNext()) {
				pss.stm_insLanguage.clearBatch();
				while(iter.hasNext()){
					pss.stm_insLanguage.setString(1, descrId);
					pss.stm_insLanguage.setString(2, (String)iter.next());
					pss.stm_insLanguage.addBatch();
				}
				pss.stm_insLanguage.executeBatch();
			}
			
			// Ontologies
			iter = dfd.getAllOntologies();
			if (iter.hasNext()) {
				pss.stm_insOntology.clearBatch();
				while(iter.hasNext()){
					pss.stm_insOntology.setString(1, descrId);
					pss.stm_insOntology.setString(2, (String)iter.next());
					pss.stm_insOntology.addBatch();
				}
				pss.stm_insOntology.executeBatch();
			}
			
			// Protocols
			iter = dfd.getAllProtocols();
			if (iter.hasNext()) {
				pss.stm_insProtocol.clearBatch();
				while(iter.hasNext()){
					pss.stm_insProtocol.setString(1, descrId);
					pss.stm_insProtocol.setString(2, (String)iter.next());
					pss.stm_insProtocol.addBatch();
				}
				pss.stm_insProtocol.executeBatch();
			}
			
			// Services
			saveServices(descrId, dfd.getAllServices());    
			
			regsCnt++;
			// clear outdated entries after a certain number of new registrations
			if(regsCnt > MAX_REGISTER_WITHOUT_CLEAN){
				regsCnt = 0;
				clean();
			}
			
			conn.commit();		
		} catch (SQLException sqle) {
			// Rollback the transaction
			try {
				conn.rollback();
			} catch (SQLException se) {
				logger.log(Logger.SEVERE,"Rollback for incomplete insertion of DFD for agent "+dfd.getName() + " failed.", se);
			}
			// Re-throw the exception
			throw sqle;
		} 
		
		return dfdToReturn;
	}
	
	/**
	 * Remove the DFD object corresponding to the indicated AID.
	 * @return the removed DFD (if any) 
	 */
	protected Object removeSingle(Object name) throws SQLException {
		AID agentAID = (AID) name;
		String n = agentAID.getName();
		
		DFAgentDescription dfd = getDFD(n);
		if (dfd != null) {
			remove(n);
		}
		
		return dfd;
	}

	/**
	 * Retrieve the DFDs matching the given template
	 */
	protected List searchSingle(Object template, int maxResult) throws SQLException {
		List matchingAIDs = new ArrayList();
		
		// Get the names of all DFDs matching the template
		String select = null;
		ResultSet rs = null;
		Statement s = null;
		
		try {
			select = createSelect((DFAgentDescription) template);
					
			s = getConnectionWrapper().getConnection().createStatement();
			if (maxResult >= 0) {
				s.setMaxRows(maxResult);
				s.setFetchSize(maxResult);
			}
			rs = s.executeQuery(select);
			
			while(rs.next()) { 
				String aidS = rs.getString("aid");
				matchingAIDs.add(aidS);
			}			
		} 
		catch(SQLException sqle) {
			// Let it through
			throw sqle;
		}
		catch(Exception e) {
			logger.log(Logger.SEVERE, "Couldn't create the SQL SELECT statement.", e);
			throw new SQLException("Couldn't create the SQL SELECT statement. "+e.getMessage());
		}
		finally {
			closeResultSet(rs);
			closeStatement(s);
		}
		
		// For each matching AID reconstruct the complete DFD
		List dfds = new ArrayList(matchingAIDs.size());
		Iterator it = matchingAIDs.iterator();
		// FIXME: Define a proper constant and possibly a proper configuration option
		if (matchingAIDs.size() < 10) {
			while (it.hasNext()) {
				dfds.add(getDFD((String) it.next()));
			}
		}
		else {
			// If we found several matching agents we preload protocols languages and ontologies once for all 
			// instead of making several queries one per agent.
			PreparedStatements pss = getPreparedStatements();
			Map allLanguages = preloadIdValueTable(pss.stm_selCountAllLanguages, pss.stm_selAllLanguages);
			Map allOntologies = preloadIdValueTable(pss.stm_selCountAllOntologies, pss.stm_selAllOntologies);
			Map allProtocols = preloadIdValueTable(pss.stm_selCountAllProtocols, pss.stm_selAllProtocols);
			while (it.hasNext()) {
				dfds.add(getDFD((String) it.next(), allLanguages, allOntologies, allProtocols));
			}
		}
		
		return dfds;
	}
	
	private Map preloadIdValueTable(PreparedStatement cntStm, PreparedStatement stm) throws SQLException {
		Map m = null;
		ResultSet rs = cntStm.executeQuery();
		rs.next();
		long recordCount = rs.getLong(1);
		closeResultSet(rs);
		if (recordCount < MAX_PRELOAD_CNT) {
			// If there are more than MAX_PRELOAD_CNT elements return null. In fact it is more time consuming constructing the preload Map 
			// than performing all queries in the DB.
			rs = stm.executeQuery();
			
			if (true) {
				m = new HashMap();
				String currentId = null;
				List l = null;
				while(rs.next()){
					String id = rs.getString(1); // id (using the index is faster)
					if (!id.equals(currentId)) {
						l =  new ArrayList();
						m.put(id, l);
						currentId = id;
					}
					l.add(rs.getString(2)); // value (protocol, language, ontology ... depending on the passed statement)
				}
				closeResultSet(rs);
			}
		}
		return m;
	}


	/**
	 */
	protected KBIterator iteratorSingle(Object template) throws SQLException {
		String select = null;
		ResultSet rs = null;
		Statement s = null;
		
		try {
			select = createSelect((DFAgentDescription) template);
			
			s = getConnectionWrapper().getConnection().createStatement();
			rs = s.executeQuery(select);
			
			return new DFDBKBIterator(s, rs);
		} 
		catch(SQLException se){
			logger.log(Logger.SEVERE, "Error accessing DB: "+select, se);
			closeResultSet(rs);
			closeStatement(s);
			throw se;
		}
		catch(Exception e) {
			logger.log(Logger.SEVERE, "Error creating SQL SELECT statement.", e);
			throw new SQLException("Error creating SQL SELECT statement. "+e.getMessage());
		}
	}
	
	
	/**
	 Inner class DFDBKBIterator
	 */
	private class DFDBKBIterator implements KBIterator {
		private Statement s = null;
		private ResultSet rs = null;
		private boolean hasMoreElements = false;
		
		public DFDBKBIterator(Statement s, ResultSet rs) throws SQLException {
			this.s = s;
			this.rs = rs;
			if (rs != null) {
				// Move to the first row
				hasMoreElements = rs.next();
			}
		}
		
		public boolean hasNext() {
			return hasMoreElements;
		}
		
		public Object next() {
			if (hasMoreElements) {
				try {
					String name = rs.getString("aid");
					DFAgentDescription dfd = getDFD(name);
					hasMoreElements = rs.next();
					return dfd;
				}
				catch (SQLException sqle) {
					hasMoreElements = false;
					throw new NoSuchElementException("DB Error. "+sqle.getMessage());
				}
			}
			throw new NoSuchElementException("");
		}
		
		public void remove() {
			// Not implemented
		}
		
		public void close() {
			closeResultSet(rs);
			closeStatement(s);
		}
	} // END of inner class DFDBKBIterator
	
	
	/**
	 * Reconstructs an AID object corresponding to the given AID name
	 * @throws SQLException
	 */
	private AID getAID(String aidN) throws SQLException {
		
		ResultSet rs = null;
		AID id = new AID(aidN, AID.ISGUID);

		PreparedStatements pss = getPreparedStatements();
		// AID addresses
		pss.stm_selAgentAddresses.setString(1, aidN);
		rs = pss.stm_selAgentAddresses.executeQuery();
		while(rs.next()){
			id.addAddresses(rs.getString(1));
		}
		
		// AID resolvers 
		Collection resolvers = getResolverAIDs(aidN);
		Iterator iter = resolvers.iterator();
		while (iter.hasNext()) {
			id.addResolvers(getAID((String)iter.next()));
		}
		
		// AID User defined slots
		pss.stm_selAgentUserDefSlot.setString(1, aidN);
		rs = pss.stm_selAgentUserDefSlot.executeQuery();
		while(rs.next()) {
			String key = rs.getString("slotkey");
			String value = rs.getString("slotval");
			id.addUserDefinedSlot(key, value);
		}
		
		return id;
	}
	
	private DFAgentDescription getDFD(String aidN) throws SQLException {
		return getDFD(aidN, null, null, null);
	}
	
	/**
	 Reconstruct the DFD corresponding to the given AID name (if any)
	 */
	private DFAgentDescription getDFD(String aidN, Map allLanguages, Map allOntologies, Map allProtocols) throws SQLException {
		DFAgentDescription dfd = null;
		AID id = null;