Example 3: External Update without Compensation
This custom procedure uses an independent transaction with a transactional data source in the server. Compensating logic is defined for the independent transaction.
package proc;
import com.compositesw.extension.*;
import java.sql.*;
public class ExternalUpdate
implements CustomProcedure, java.io.Serializable
{
private static final String ORDERS_URL =
"jdbc:mysql://localhost:3306/Orders";
private transient ExecutionEnvironment qenv;
private transient Connection conn;
private transient int numRowsUpdated;
private boolean isUpdate;
private int id;
private String firstName;
private String lastName;
private String companyName;
private String phoneNumber;
public ExternalUpdate() { }
/**
* This is called once just after constructing the class. The
* environment contains methods used to interact with the server.
*/
public void initialize(ExecutionEnvironment qenv)
throws SQLException
{
this.qenv = qenv;
conn = DriverManager.getConnection(ORDERS_URL, "tutorial", "tutorial");
conn.setAutoCommit(false);
}
/**
* Called during introspection to get the description of the input
* and output parameters. Should not return null.
*/
public ParameterInfo[] getParameterInfo() {
return new ParameterInfo[] {
new ParameterInfo("Id", Types.INTEGER, DIRECTION_IN),
new ParameterInfo("FirstName", Types.VARCHAR, DIRECTION_IN),
new ParameterInfo("LastName", Types.VARCHAR, DIRECTION_IN),
new ParameterInfo("CompanyName", Types.VARCHAR, DIRECTION_IN),
new ParameterInfo("PhoneNumber", Types.VARCHAR, DIRECTION_IN),
};
}
/**
* Called to invoke the stored procedure. Will only be called a
* single time per instance. Can throw CustomProcedureException or
* SQLException if there is an error during invoke.
*/
public void invoke(Object[] inputValues)
throws CustomProcedureException, SQLException
{
Statement stmt = conn.createStatement();
//
// Save away the current values to be used for compensation
//
ResultSet rs = stmt.executeQuery(
"SELECT ContactFirstName, ContactLastName, CompanyName, PhoneNumber " +
"FROM customers WHERE CustomerID=" + inputValues[0]);
if (rs.next()) {
isUpdate = true;
id = ((Integer)inputValues[0]).intValue();
firstName = rs.getString(1);
lastName = rs.getString(2);
companyName = rs.getString(3);
phoneNumber = rs.getString(4);
}
rs.close();
//
// Perform the insert or update
//
if (isUpdate) {
numRowsUpdated = stmt.executeUpdate(
"UPDATE customers" +
" SET ContactFirstName='" + inputValues[1] +
"', ContactLastName='" + inputValues[2] +
"', CompanyName='" + inputValues[3] +
"', PhoneNumber='" + inputValues[4] +
"' WHERE CustomerID=" + inputValues[0]);
}
else {
numRowsUpdated = stmt.executeUpdate(
"INSERT into customers (CustomerID, ContactFirstName, " +
"ContactLastName, CompanyName, PhoneNumber) VALUES (" +
inputValues[0] + ", '" + inputValues[1] + "', '" +
inputValues[2] + "', '" + inputValues[3] + "', '" +
inputValues[4] + "')");
}
stmt.close();
}
/**
* Called to retrieve the number of rows that were inserted,
* updated, or deleted during the execution of the procedure. A
* return value of -1 indicates that the number of affected rows is
* unknown. Can throw CustomProcedureException or SQLException if
* there is an error when getting the number of affected rows.
*/
public int getNumAffectedRows() {
return numRowsUpdated;
}
/**
* Called to retrieve the output values. The returned objects
* should obey the Java to SQL typing conventions as defined in the
* table above. Output cursors can be returned as either
* CustomCursor or java.sql.ResultSet. Can throw
* CustomProcedureException or SQLException if there is an error
* when getting the output values. Should not return null.
*/
public Object[] getOutputValues() {
return new Object[] { };
}
/**
* Called when the procedure reference is no longer needed. Close
* can be called without retrieving any of the output values (such
* as cursors) or even invoking, so this needs to do any remaining
* cleanup. Close can be called concurrently with any other call
* such as "invoke" or "getOutputValues". In this case, any pending
* methods should immediately throw a CustomProcedureException.
*/
public void close()
throws SQLException
{ }
//
// Introspection methods
//
/**
* Called during introspection to get the short name of the stored
* procedure. This name can be overridden during configuration.
* Should not return null.
*/
public String getName() {
return "ExternalUpdate";
}
/**
* Called during introspection to get the description of the stored
* procedure. Should not return null.
*/
public String getDescription() {
return "This procedure performs an update to an external transactional " +
"data source using JDBC.";
}
//
// Transaction methods
//
/**
* Returns true if the custom procedure uses transactions. If this
* method returns false then commit and rollback will not be called.
*/
public boolean canCommit() {
return true;
}
/**
* Commit any open transactions
*/
public void commit()
throws SQLException
{
conn.commit();
conn.close();
conn = null;
}
/**
* Rollback any open transactions.
*/
public void rollback()
throws SQLException
{
conn.rollback();
conn.close();
conn = null;
}
/**
* Returns true if the transaction can be compensated.
*/
public boolean canCompensate() {
return true;
}
/**
* Compensate any committed transactions (if supported).
*/
public void compensate(ExecutionEnvironment qenv)
throws SQLException
{
conn = DriverManager.getConnection(ORDERS_URL);
conn.setAutoCommit(false);
Statement stmt = conn.createStatement();
if (isUpdate) {
numRowsUpdated = stmt.executeUpdate(
"UPDATE customers" +
" SET ContactFirstName='" + firstName +
"', ContactLastName='" + lastName +
"', CompanyName='" + companyName +
"', PhoneNumber='" + phoneNumber +
"' WHERE CustomerID=" + id);
}
else {
stmt.executeUpdate("DELETE from customers WHERE CustomerID=" + id);
}
stmt.close();
conn.commit();
conn.close();
conn = null;
}
}