dfdbkb.java

java实现的P2P多agent中间件

/**
 * JADE - Java Agent DEvelopment Framework is a framework to develop
 * multi-agent systems in compliance with the FIPA specifications.
 * Copyright (C) 2000 CSELT S.p.A.
 *
 * GNU Lesser General Public License
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation,
 * version 2.1 of the License.
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the
 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
 * Boston, MA  02111-1307, USA.
 */
package jade.domain;

//#J2ME_EXCLUDE_FILE
//#APIDOC_EXCLUDE_FILE

import jade.core.AID;
import jade.domain.FIPAAgentManagement.DFAgentDescription;
import jade.domain.FIPAAgentManagement.NotUnderstoodException;
import jade.domain.FIPAAgentManagement.Property;
import jade.domain.FIPAAgentManagement.ServiceDescription;
import jade.domain.KBManagement.*;
import jade.lang.acl.ACLMessage;
import jade.lang.acl.ACLCodec;
import jade.lang.acl.StringACLCodec;
import jade.proto.SubscriptionResponder;

import jade.util.leap.Collection;
import jade.util.leap.Iterator;
import jade.util.leap.ArrayList;
import jade.util.leap.List;
import jade.util.leap.Properties;
import jade.util.Logger;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.rmi.server.UID;
import java.security.MessageDigest;
import java.sql.*;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Map;
import java.util.Date;
import java.util.Vector;
import java.util.NoSuchElementException;

import org.apache.commons.codec.binary.Base64;


/**
 * This class implements a knowledge base used by the DF which stores its content
 * in an external database. 
 * 
 * @author Elisabetta Cortese - TILab
 * @author Roland Mungenast - Profactor
 */
public class DFDBKB extends DBKB {
	private static final int MAX_PRELOAD_CNT = 1000;
	private static final int MAX_REGISTER_WITHOUT_CLEAN = 100;
	private static final int MAX_PROP_LENGTH = 255;
	
	// Table names
	private static final String SUBSCRIPTION = "subscription";
	private static final String SERVICEPROTOCOL = "serviceprotocol";
	private static final String SERVICEONTOLOGY = "serviceontology";
	private static final String SERVICELANGUAGE = "servicelanguage";
	private static final String SERVICEPROPERTY = "serviceproperty";
	private static final String SERVICE = "service";
	private static final String LANGUAGE = "language";
	private static final String ONTOLOGY = "ontology";
	private static final String PROTOCOL = "protocol";
	private static final String AGENTUSERDEFSLOT = "agentuserdefslot";
	private static final String AGENTRESOLVER = "agentresolver";
	private static final String AGENTADDRESS = "agentaddress";
	private static final String DFAGENTDESCR = "dfagentdescr";
	
	// Number of registrations after the last lease-time-cleanup
	private int regsCnt = 0;
	
	private boolean tablesReady = false;
	
	/**
	 * Default data type for very long strings
	 */
	protected String DEFAULT_LONGVARCHAR_TYPE = "LONGVARCHAR";
	
	// This is used to generate unique IDs
	private String localIPAddress;
	
	private class PreparedStatements {
		// prepared SQL statements
		private PreparedStatement stm_selNrOfDescrForAID;
		private PreparedStatement stm_selAgentAddresses;
		private PreparedStatement stm_selAgentResolverAIDs;
		private PreparedStatement stm_selAgentUserDefSlot;
		private PreparedStatement stm_selLease;
		private PreparedStatement stm_selProtocols;
		private PreparedStatement stm_selLanguages;
		private PreparedStatement stm_selOntologies;
		private PreparedStatement stm_selServices;
		private PreparedStatement stm_selServiceProtocols;
		private PreparedStatement stm_selServiceLanguages;
		private PreparedStatement stm_selServiceOntologies;
		private PreparedStatement stm_selServiceProperties;
		private PreparedStatement stm_selExpiredDescr;
		private PreparedStatement stm_selSubscriptions;
		
		private PreparedStatement stm_selAllProtocols;
		private PreparedStatement stm_selCountAllProtocols;
		private PreparedStatement stm_selAllLanguages;
		private PreparedStatement stm_selCountAllLanguages;
		private PreparedStatement stm_selAllOntologies;
		private PreparedStatement stm_selCountAllOntologies;
		
		private PreparedStatement stm_insAgentDescr;
		private PreparedStatement stm_insAgentAddress;
		private PreparedStatement stm_insAgentUserDefSlot;
		private PreparedStatement stm_insAgentResolverAID;
		private PreparedStatement stm_insLanguage;
		private PreparedStatement stm_insOntology;
		private PreparedStatement stm_insProtocol;
		private PreparedStatement stm_insService;
		private PreparedStatement stm_insServiceProtocol;
		private PreparedStatement stm_insServiceOntology;
		private PreparedStatement stm_insServiceLanguage;
		private PreparedStatement stm_insServiceProperty;
		private PreparedStatement stm_insSubscription;
		private PreparedStatement stm_delAgentDescr;
		private PreparedStatement stm_delAgentUserDefSlot;
		private PreparedStatement stm_delAgentResolver;
		private PreparedStatement stm_delAgentAddress;
		
		private PreparedStatement stm_selDescrId; 
		private PreparedStatement stm_selServiceId;
		private PreparedStatement stm_delService;
		private PreparedStatement stm_delLanguage;
		private PreparedStatement stm_delProtocol;
		private PreparedStatement stm_delOntology;
		
		private PreparedStatement stm_delServiceLanguage;
		private PreparedStatement stm_delServiceOntology;
		private PreparedStatement stm_delServiceProtocol;
		private PreparedStatement stm_delServiceProperty;
		private PreparedStatement stm_delSubscription;

		private PreparedStatements(Connection conn) throws SQLException {
			// select statements
			stm_selNrOfDescrForAID = conn.prepareStatement("SELECT COUNT(*) FROM dfagentdescr WHERE aid = ?");
			stm_selAgentAddresses = conn.prepareStatement("SELECT address FROM agentaddress WHERE aid = ?");
			stm_selAgentResolverAIDs = conn.prepareStatement("SELECT resolveraid FROM agentresolver WHERE aid = ?");
			stm_selAgentUserDefSlot = conn.prepareStatement("SELECT slotkey, slotval FROM agentuserdefslot WHERE aid = ?");
			stm_selLease = conn.prepareStatement("SELECT id, lease FROM dfagentdescr WHERE aid = ?");
			stm_selProtocols = conn.prepareStatement("SELECT protocol FROM protocol WHERE descrid = ?");
			stm_selLanguages = conn.prepareStatement("SELECT language FROM language WHERE descrid = ?");
			stm_selOntologies = conn.prepareStatement("SELECT ontology FROM ontology WHERE descrid = ?");
			stm_selServices = conn.prepareStatement("SELECT id, sname, stype, sownership FROM service WHERE descrid = ?");
			stm_selServiceProtocols = conn.prepareStatement("SELECT protocol FROM serviceprotocol WHERE serviceid = ?");
			stm_selServiceLanguages = conn.prepareStatement("SELECT ontology FROM serviceontology WHERE serviceid = ?");
			stm_selServiceOntologies = conn.prepareStatement("SELECT language FROM servicelanguage WHERE serviceid = ?");
			stm_selServiceProperties = conn.prepareStatement("SELECT propkey, propval_str, propval_obj FROM serviceproperty WHERE serviceid = ?");
			stm_selDescrId = conn.prepareStatement("SELECT id FROM dfagentdescr WHERE aid = ?"); 
			stm_selServiceId = conn.prepareStatement("SELECT id FROM service WHERE descrid = ?");
			stm_selExpiredDescr = conn.prepareStatement("SELECT aid FROM dfagentdescr WHERE lease < ? AND lease <> '-1'");
			stm_selSubscriptions = conn.prepareStatement("SELECT * FROM subscription");

			stm_selAllProtocols = conn.prepareStatement("SELECT descrid, protocol FROM protocol ORDER BY descrid");
			stm_selCountAllProtocols = conn.prepareStatement("SELECT COUNT(*) FROM protocol");
			stm_selAllLanguages = conn.prepareStatement("SELECT descrid, language FROM language ORDER BY descrid");
			stm_selCountAllLanguages = conn.prepareStatement("SELECT COUNT(*) FROM language");
			stm_selAllOntologies = conn.prepareStatement("SELECT descrid, ontology FROM ontology ORDER BY descrid");
			stm_selCountAllOntologies = conn.prepareStatement("SELECT COUNT(*) FROM ontology");
			
			stm_insAgentDescr = conn.prepareStatement("INSERT INTO dfagentdescr VALUES (?, ?, ?)");
			stm_insAgentAddress = conn.prepareStatement("INSERT INTO agentaddress VALUES (?, ?, ?)");
			stm_insAgentUserDefSlot = conn.prepareStatement("INSERT INTO agentuserdefslot VALUES (?, ?, ?, ?)");
			stm_insAgentResolverAID = conn.prepareStatement("INSERT INTO agentresolver VALUES (?, ?, ?)");
			stm_insLanguage = conn.prepareStatement("INSERT INTO language VALUES (?, ?)");
			stm_insOntology = conn.prepareStatement("INSERT INTO ontology VALUES (?, ?)");
			stm_insProtocol = conn.prepareStatement("INSERT INTO protocol VALUES (?, ?)");
			stm_insService = conn.prepareStatement("INSERT INTO service VALUES (?, ?, ?, ?, ?)");
			stm_insServiceProtocol = conn.prepareStatement("INSERT INTO serviceprotocol VALUES (?, ?)");
			stm_insServiceOntology = conn.prepareStatement("INSERT INTO serviceontology VALUES (?, ?)");
			stm_insServiceLanguage = conn.prepareStatement("INSERT INTO servicelanguage VALUES (?, ?)");
			stm_insServiceProperty = conn.prepareStatement("INSERT INTO serviceproperty VALUES (?, ?, ?, ?, ?)");
			stm_insSubscription = conn.prepareStatement("INSERT INTO subscription VALUES (?, ?)");
			
			
			// delete statements
			stm_delAgentDescr = conn.prepareStatement("DELETE FROM dfagentdescr WHERE id = ?");
			stm_delAgentUserDefSlot = conn.prepareStatement("DELETE FROM agentuserdefslot WHERE aid = ?");
			stm_delAgentResolver = conn.prepareStatement("DELETE FROM agentresolver WHERE aid = ?");
			stm_delAgentAddress = conn.prepareStatement("DELETE FROM agentaddress WHERE aid = ?");
			stm_delLanguage = conn.prepareStatement("DELETE FROM language WHERE descrid = ?");
			stm_delProtocol = conn.prepareStatement("DELETE FROM protocol WHERE descrid = ?");
			stm_delOntology = conn.prepareStatement("DELETE FROM ontology WHERE descrid = ?");
			stm_delService = conn.prepareStatement("DELETE FROM service WHERE descrid = ?"); 
			stm_delServiceLanguage = conn.prepareStatement("DELETE FROM servicelanguage WHERE serviceid = ?");
			stm_delServiceOntology = conn.prepareStatement("DELETE FROM serviceontology WHERE serviceid = ?");
			stm_delServiceProtocol = conn.prepareStatement("DELETE FROM serviceprotocol WHERE serviceid = ?");
			stm_delServiceProperty = conn.prepareStatement("DELETE FROM serviceproperty WHERE serviceid = ?");
			stm_delSubscription = conn.prepareStatement("DELETE FROM subscription WHERE id = ?");
		}
	}
	
	/**
	 * Constructor
	 * @param maxResultLimit internal limit for the number of maximum search results.
	 * @param drv database driver
	 * @param url database URL
	 * @param user database user name
	 * @param passwd database password
	 * @param cleanTables specifies whether the KB should delete all existing tables for the DF at startup
	 * @throws SQLException an error occured while opening a connection to the database
	 */
	public DFDBKB(int maxResultLimit, String drv, String url, String user, String passwd, boolean cleanTables) throws SQLException {
		super(drv, url, user, passwd, maxResultLimit, cleanTables);
	}
	
	/**
	 * Initializes all used SQL statements, the DB tables and the logging
	 */
	public void setup() throws SQLException {
		logger = Logger.getMyLogger(this.getClass().getName());
		
		try {
			localIPAddress = InetAddress.getLocalHost().getHostAddress();
		} catch (Exception e) {
			localIPAddress = "localhost";
		} 
		
		ConnectionWrapper wrapper = getConnectionWrapper();
		Connection conn = wrapper.getConnection();
		try {
			conn.setAutoCommit(false); // deactivate auto commit for better performance
		} catch (Exception e) {
			if(logger.isLoggable(Logger.WARNING)) {
				logger.log(Logger.WARNING, "Disabling auto-commit failed.");
			}
		}
		
		if (cleanTables) {
			// Drop all existing tables for the DF if required
			dropDFTables();
		}
		createDFTables();
		tablesReady = true;
		PreparedStatements ps = new PreparedStatements(conn);
		wrapper.setInfo(ps);
		clean();
	}
	
	protected void initConnectionWrapper(ConnectionWrapper wrapper) throws SQLException {
		Connection conn = wrapper.getConnection();
		try {
			conn.setAutoCommit(false); // deactivate auto commit for better performance
		} catch (Exception e) {
			if(logger.isLoggable(Logger.WARNING)) {
				logger.log(Logger.WARNING, "Disabling auto-commit failed.");
			}
		}

		// We cannot initialize the prepared-statements if the tables are not there yet
		if (tablesReady) {
			PreparedStatements ps = new PreparedStatements(conn);
			wrapper.setInfo(ps);
		}
	}

	private PreparedStatements getPreparedStatements() throws SQLException {
		ConnectionWrapper wrapper = getConnectionWrapper();
		return (PreparedStatements)wrapper.getInfo();
	}

	/**
	 * Returns the name of the SQL type used in the
	 * database to represent very long strings.
	 */
	protected String getLongVarCharType() {
		String bestMatch = DEFAULT_LONGVARCHAR_TYPE;
		try {
			// get the datatype with the highest precision from the meta data information
			DatabaseMetaData md = getConnectionWrapper().getConnection().getMetaData();
			ResultSet typeInfo = md.getTypeInfo();
			long maxPrecision = -1;
			while (typeInfo.next()) {
				long jdbcType = Long.parseLong(typeInfo.getString("DATA_TYPE"));
				long precision = Long.parseLong(typeInfo.getString("PRECISION"));
				
				if (jdbcType == Types.LONGVARCHAR && precision > maxPrecision) {
					maxPrecision = precision;
					bestMatch = typeInfo.getString("TYPE_NAME");
				}
			}
		} catch (SQLException e) {
			// ignored --> default value will be returned
		}
		return bestMatch;
	}
		
	/**
	 * Returns a global unique identifier
	 */
	protected String getGUID() {
		UID uid = new UID();
		return localIPAddress + ":" + uid;
	}
	
	/**
	 * Drops a DB table or does nothing if the table doesn't exist.
	 */                
	protected void dropTable(Statement stmt, String tableName) {
		try {
			stmt.execute("DROP TABLE " + tableName + " CASCADE CONSTRAINTS"); 
			getConnectionWrapper().getConnection().commit();
		} catch (SQLException e) {
			// Check if the exception is because the table does not exist
			if (tableExists(tableName)) {
				logger.log(Logger.WARNING, "Cannot clean table "+tableName, e);
			}
		}
	}
	
	/**
	 * Drops all existing DB tables used by the DF.
	 */
	protected void dropDFTables() throws SQLException {    
		
		logger.log(Logger.INFO, "Cleaning DF tables...");
		
		Statement stmt = getConnectionWrapper().getConnection().createStatement();
		
		dropTable(stmt, SUBSCRIPTION); 
		dropTable(stmt, SERVICEPROTOCOL);
		dropTable(stmt, SERVICEONTOLOGY);
		dropTable(stmt, SERVICELANGUAGE);
		dropTable(stmt, SERVICEPROPERTY);
		dropTable(stmt, SERVICE);
		dropTable(stmt, LANGUAGE);
		dropTable(stmt, ONTOLOGY);
		dropTable(stmt, PROTOCOL);
		dropTable(stmt, AGENTUSERDEFSLOT);
		dropTable(stmt, AGENTRESOLVER);   
		dropTable(stmt, AGENTADDRESS);
		dropTable(stmt, DFAGENTDESCR); 
		
		stmt.close();
	}
	
	/**
	 * Check whether a database table already exists
	 * 
	 * @param name name of the table
	 */
	protected boolean tableExists(String name) {
		Statement stmt = null;
		
		try {