testclientsyncsource.java

通过SyncML对异构数据库复制的代码。如何通过Sync4J作为服务进行同步。

package test.sync4j;

import sync4j.syncclient.spds.SyncException;
import sync4j.syncclient.spds.engine.SyncItem;
import sync4j.syncclient.spds.engine.SyncItemImpl;
import sync4j.syncclient.spds.engine.SyncItemKey;
import sync4j.syncclient.spds.engine.SyncItemProperty;
import sync4j.syncclient.spds.engine.SyncItemState;
import sync4j.syncclient.spds.engine.SyncSource;

import javax.naming.InitialContext;
import javax.naming.NamingException;
import javax.sql.DataSource;
import java.security.Principal;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

public class TestClientSyncSource implements SyncSource{
	private static final String JDBC_RECORDSET_MIMETYPE = "application/vnd.syncml+xmlrecordset";

	private static final String METADATA_COLUMN_NAME = "COLUMN_NAME";

	private static final int SQL_GET_ALL_ITEMS = 0;
	private static final int SQL_GET_ALL_ITEMS_SINCE = 1;
	private static final int SQL_GET_ITEMS = 2;
	private static final int SQL_GET_ITEMS_SINCE = 3;
	private static final int SQL_DELETE_ITEM = 4;
	private static final int SQL_ADD_ITEM = 5;
	private static final int SQL_UPDATE_ITEM = 6;

	private boolean initialised;

	private DataSource dataSource;
	private String[] columnNames;
	private String[] sqlStatements;

	private String name;
	private String type;
	private String sourceURI;

	private String jndiName;
	private String jdbcDriver;
	private String urlConnection;
	private String userConnection;
	private String passwordConnection;
	private String tableName;
	private String keyColumnName;
	private String timestampColumnName;
	private String stateColumnName;
	private String catalogName;
	private String schemaName;

	public TestClientSyncSource(){
		type = JDBC_RECORDSET_MIMETYPE;
		initialised = false;
	}

	public void init() throws SyncException{
		//Initialize the jndi datasource if provided
		if(jndiName != null && jndiName.length() > 0){
			try{
				dataSource = (DataSource)new InitialContext().lookup(jndiName);
			}
			catch(NamingException e){
				throw new SyncException("Data source not found: " + jndiName, e);
			}
		}

		//Get the table's column names (excluding key, timestamp, state columns)
		Connection connection = null;
		try{
			connection = getConnection();
			columnNames = getColumnNames(connection.getMetaData(), catalogName, schemaName, tableName);
		}
		catch(SQLException e){
			throw new SyncException("SQL exception initializing " + getClass().getName(), e);
		}
		finally{
			closeConnection(connection);
		}

		//Prepare sql statements
		sqlStatements = buildSql();
	}

	public String getName(){
		return name;
	}

	public void setName(String name){
		this.name = name;
	}

	public String getType(){
		return type;
	}

	public void setType(String type){
		this.type = type;
	}

	public String getSourceURI(){
		return sourceURI;
	}

	public void setSourceURI(String sourceURI){
		this.sourceURI = sourceURI;
	}

	public String getJndiName(){
		return jndiName;
	}

	public void setJndiName(String jndiName){
		this.jndiName = jndiName;
	}

	public String getJdbcDriver(){
		return jdbcDriver;
	}

	public void setJdbcDriver(String jdbcDriver){
		this.jdbcDriver = jdbcDriver;
	}

	public String getUrlConnection(){
		return urlConnection;
	}

	public void setUrlConnection(String urlConnection){
		this.urlConnection = urlConnection;
	}

	public String getUserConnection(){
		return userConnection;
	}

	public void setUserConnection(String userConnection){
		this.userConnection = userConnection;
	}

	public String getPasswordConnection(){
		return passwordConnection;
	}

	public void setPasswordConnection(String passwordConnection){
		this.passwordConnection = passwordConnection;
	}

	public String getTableName(){
		return tableName;
	}

	public void setTableName(String tableName){
		this.tableName = tableName;
	}

	public String getKeyColumnName(){
		return keyColumnName;
	}

	public void setKeyColumnName(String keyColumnName){
		this.keyColumnName = keyColumnName;
	}

	public String getTimestampColumnName(){
		return timestampColumnName;
	}

	public void setTimestampColumnName(String timestampColumnName){
		this.timestampColumnName = timestampColumnName;
	}

	public String getStateColumnName(){
		return stateColumnName;
	}

	public void setStateColumnName(String stateColumnName){
		this.stateColumnName = stateColumnName;
	}

	public String getCatalogName(){
		return catalogName;
	}

	public void setCatalogName(String catalogName){
		this.catalogName = catalogName;
	}

	public String getSchemaName(){
		return schemaName;
	}

	public void setSchemaName(String schemaName){
		this.schemaName = schemaName;
	}

	public String toString(){
		StringBuffer sb = new StringBuffer(super.toString());
		sb.append(" - { ");
		sb.append("name: ").append(name).append(", ");
		sb.append("jndiName: ").append(jndiName).append(", ");
		sb.append("jdbcDriver: ").append(jdbcDriver).append(", ");
		sb.append("urlConnection: ").append(urlConnection).append(", ");
		sb.append("userConnection: ").append(userConnection).append(", ");
		sb.append("passwordConnection: ").append(passwordConnection).append(", ");
		sb.append("tableName: ").append(tableName).append(", ");
		sb.append("keyColumnName: ").append(keyColumnName).append(", ");
		sb.append("timestampColumnName: ").append(timestampColumnName).append(", ");
		sb.append("stateColumnName: ").append(stateColumnName).append(" }");
		return sb.toString();
	}

	public void beginSync(int type) throws SyncException{
		if(!initialised){
			init();
			initialised = true;
		}
	}

	public void commitSync() throws SyncException{
	}

	public SyncItem[] getAllSyncItems(Principal principal) throws SyncException{
		//Get all sync items
		return getSyncItems(SyncItemState.UNKNOWN, null);
	}

	public SyncItem[] getDeletedSyncItems(Principal principal, Date since) throws SyncException{
		//Get all deleted sync items since the specified date
		return getSyncItems(SyncItemState.DELETED, new Timestamp(since.getTime()));
	}

	public SyncItem[] getNewSyncItems(Principal principal, Date since) throws SyncException{
		//Get all new sync items since the specified date
		return getSyncItems(SyncItemState.NEW, new Timestamp(since.getTime()));
	}

	public SyncItem[] getUpdatedSyncItems(Principal principal, Date since) throws SyncException{
		//Get all updated sync items since the specified date
		return getSyncItems(SyncItemState.UPDATED, new Timestamp(since.getTime()));
	}

	public void removeSyncItem(Principal principal, SyncItem syncItem) throws SyncException{
		//Delete a sync item
		Connection connection = null;
		try{
			connection = getConnection();
			PreparedStatement statement = connection.prepareStatement(sqlStatements[SQL_DELETE_ITEM]);
			statement.setTimestamp(1, new Timestamp(System.currentTimeMillis()));
			statement.setString(2, syncItem.getKey().getKeyAsString());
			statement.executeUpdate();
		}
		catch(SQLException e){
			throw new SyncException("Error removing item " +
				syncItem.getKey().getKeyAsString(), e);
		}
		finally{
			closeConnection(connection);
		}
	}

	public SyncItem setSyncItem(Principal principal, SyncItem syncItem) throws SyncException{
		//Set a sync item with specified item data
		String[] values = xmlToValues(new String((byte[])syncItem.getPropertyValue(SyncItem.PROPERTY_BINARY_CONTENT)));
		Date timestamp = (Date)syncItem.getPropertyValue(SyncItem.PROPERTY_TIMESTAMP);
		Connection connection = null;
		try{
			//Try to update item
			connection = getConnection();
			PreparedStatement statement = connection.prepareStatement(sqlStatements[SQL_UPDATE_ITEM]);
			statement.setString(1, String.valueOf(SyncItemState.UPDATED));
			statement.setTimestamp(2, new Timestamp(timestamp.getTime()));
			for(int i = 0; i < values.length; ++i){
				statement.setString(i + 3, values[i]);
			}
			statement.setString(values.length + 3, syncItem.getKey().getKeyAsString());
			if(statement.executeUpdate() != 0){
				return syncItem;
			}

			//Update failed, add new item
			try{
				statement.close();
			}
			catch(SQLException e){
			}
			statement = connection.prepareStatement(sqlStatements[SQL_ADD_ITEM]);
			statement.setString(1, String.valueOf(SyncItemState.NEW));
			statement.setTimestamp(2, new Timestamp(timestamp.getTime()));
			statement.setString(3, syncItem.getKey().getKeyAsString());
			for(int i = 0; i < columnNames.length; ++i){
				statement.setString(i + 4, values[i]);
			}
			statement.executeUpdate();
		}
		catch(SQLException e){
			throw new SyncException("Error setting item " +
				syncItem.getKey().getKeyAsString(), e);
		}
		finally{
			closeConnection(connection);
		}
		return syncItem;
	}

	private static String[] xmlToValues(String xml){
		//Fix for Sync4J 2.0, remove for newer releases
		if(xml != null) xml = xml.replaceAll("&lt;", "<");

		//Extract values from xml
		List values = new ArrayList();
		for(int end, offset = 0; offset < xml.length(); offset = end + 8){
			offset = xml.indexOf("<field>", offset);
			if(offset < 0){
				break;
			}
			offset += 7;
			end = xml.indexOf("</field>", offset);
			if(end < 0){
				break;
			}
			values.add(xml.substring(offset, end));
		}
		return (String[])values.toArray(new String[values.size()]);
	}

	public SyncItem[] setSyncItems(Principal principal, SyncItem[] syncItems) throws SyncException{
		//Set sync items with specified data
		if(syncItems != null){
			for(int i = 0; i < syncItems.length; i++){
				setSyncItem(principal, syncItems[i]);
			}
		}
		return syncItems;
	}

	public SyncItem getSyncItemFromId(Principal principal, SyncItemKey syncItemKey) throws SyncException{
		return null;
	}

	public SyncItem[] getSyncItemsFromIds(Principal principal, SyncItemKey[] syncItemsKeys) throws SyncException{
		return new SyncItem[0];
	}

	public SyncItem getSyncItemFromTwin(Principal principal, SyncItem syncItem) throws SyncException{
		return null;
	}

	public SyncItem[] getSyncItemsFromTwins(Principal principal, SyncItem[] syncItem) throws SyncException{
		return new SyncItem[0];
	}

	private SyncItem[] getSyncItems(char state, Timestamp since) throws SyncException{
		//Get sync items based on state and last modification timestamp
		List items = new ArrayList();
		Connection connection = null;
		try{
			connection = getConnection();
			PreparedStatement statement;

			if(state == SyncItemState.UNKNOWN){
				//Get all items
				if(since == null){
					statement = connection.prepareStatement(sqlStatements[SQL_GET_ALL_ITEMS]);
				}
				//Get all items since timestamp
				else{
					statement = connection.prepareStatement(sqlStatements[SQL_GET_ALL_ITEMS_SINCE]);
					statement.setTimestamp(1, since);
				}
			}
			else{
				//Get items with state
				if(since == null){
					statement = connection.prepareStatement(sqlStatements[SQL_GET_ITEMS]);
					statement.setString(1, String.valueOf(state));
				}
				//Get items with state since timestamp
				else{
					statement = connection.prepareStatement(sqlStatements[SQL_GET_ITEMS_SINCE]);
					statement.setString(1, String.valueOf(state));
					statement.setTimestamp(2, since);
				}
			}

			//Build sync items from the resultset
			for(ResultSet rs = statement.executeQuery(); rs.next();){
				items.add(rowToSyncItem(rs));
			}
		}
		catch(SQLException e){
			throw new SyncException("Error getting items", e);
		}
		finally{
			closeConnection(connection);
		}
		return (SyncItem[])items.toArray(new SyncItem[items.size()]);
	}

	private SyncItem rowToSyncItem(ResultSet rs) throws SQLException{
		//Convert a database record to a sync item
		StringBuffer xml = new StringBuffer();
		xml.append("<![CDATA[\n");
		xml.append("<record>\n");
		for(int i = 0; i < columnNames.length; i++){
			xml.append("<field>").append(rs.getString(columnNames[i])).append("</field>\n");
		}
		xml.append("</record>\n");
		xml.append("]]>\n");

		SyncItem item = new SyncItemImpl(this, rs.getString(keyColumnName),
			rs.getString(stateColumnName).charAt(0));
		item.setProperty(new SyncItemProperty(SyncItem.PROPERTY_BINARY_CONTENT,
			xml.toString().getBytes()));
		return item;
	}

	private Connection getConnection() throws SQLException{
		if(dataSource != null){
			return dataSource.getConnection();
		}
		try{
			Class.forName(jdbcDriver);
		}
		catch(ClassNotFoundException e){
			e.printStackTrace();
		}
		return DriverManager.getConnection(urlConnection, userConnection, passwordConnection);
	}

	private static void closeConnection(Connection connection){
		if(connection != null){
			try{
				connection.close();
			}
			catch(SQLException e){
			}
		}
	}

	private String[] getColumnNames(DatabaseMetaData metaData, String catalog, String schema, String tableName) throws SQLException{
		String catalogName = (catalog == null || catalog.length() <= 0) ? null : catalog;
		String schemaName = (schema == null || schema.length() <= 0) ? metaData.getUserName() : schema;
		ResultSet rs = metaData.getColumns(catalogName, schemaName, tableName, "%");
		List columnNames = new ArrayList();
		while(rs.next()){
			String columnName = rs.getString(METADATA_COLUMN_NAME);
			//Skip key, timestamp and state columns
			if(!(keyColumnName.equalsIgnoreCase(columnName) ||
				timestampColumnName.equalsIgnoreCase(columnName) ||
				stateColumnName.equalsIgnoreCase(columnName)) &&
				!columnNames.contains(columnName)){
				columnNames.add(columnName);
			}
		}
		try{
			rs.close();
		}
		catch(SQLException e){
		}
		return (String[])columnNames.toArray(new String[columnNames.size()]);
	}

	private String[] buildSql(){
		StringBuffer columns = new StringBuffer();
		StringBuffer columnValues = new StringBuffer();
		StringBuffer columnAssignments = new StringBuffer();
		for(int i = 0; i < columnNames.length; ++i){
			columns.append(", ").append(columnNames[i]);
			columnValues.append(", ?");
			columnAssignments.append(", ").append(columnNames[i]).append(" = ?");
		}
		Object[] args = new Object[]{
			tableName,
			stateColumnName,
			timestampColumnName,
			keyColumnName,
			columns.toString(),
			columnValues.toString(),
			columnAssignments.toString()
		};

		String[] sql = new String[]{
			"select * from {0}",
			"select * from {0} where {2} > ?",
			"select * from {0} where {1} = ?",
			"select * from {0} where {1} = ? and {2} > ?",
			"update {0} set {1} = ''D'', {2} = ? where {3} = ?",
			"insert into {0} ({1}, {2}, {3}{4}) values(?, ?, ?{5})",
			"update {0} set {1} = ?, {2} = ?{6} where {3} = ?"
		};
		for(int i = 0; i < sql.length; i++){
			sql[i] = MessageFormat.format(sql[i], args);
		}
		return sql;
	}
}