enqueue 到oracle AQ 的 调用方式

ParameterMap param = new ParameterMap();
param.addValue("MESSAGE", request);
param.addValue("QUEUE_NAME", this.destination);

this.jdbcTemplate.withSqlMap("{CALL GCAL_PKG_ENQUEUE.TEXT_MESSAGE_ENQUEUE(:MESSAGE, :QUEUE_NAME)}", param).call();


jdbcTemplate
{

public JdbcTemplate withSqlMap(String sql, ParameterMap parameterMap)
{
JdbcOperation jdbcOperation = createJdbcOperation(null);
compileToSqlParameterMap(jdbcOperation, sql, parameterMap);
return this;
}

public Map<String, Object> call()
{
if (this.localJdbcOperation.get() == null) {
throw new DatabaseException("There is no database operation yet.");
}
Exception exception = null;
MetricsTimer metricsTimer = COLLECTOR.startMetricsTimer("call");
try {
Map localMap = ((JdbcOperation)this.localJdbcOperation.get()).call(metricsTimer);
return localMap;
}
catch (RuntimeException e)
{
exception = e;
throw e;
}
finally {
metricsTimer.stop(exception);
this.localJdbcOperation.set(null); } throw localObject;
}
}


JdbcOperation{

protected Map<String, Object> call(MetricsTimer metricsTimer)
{
Map resultMap = new HashMap();
CallableStatement callableStatement = null;
try
{
metricsTimer.addMetrics("compiledSql", this.compiledSql);
callableStatement = createCallableStatement(this.compiledSql);
setCallableParameters(callableStatement, this.compiledValues);
callableStatement.executeUpdate();
for (int i = 0; i < this.compiledValues.length; i++) {
if ((this.compiledValues == null) || (this.compiledValues.getInOut() == ParameterData.InOutType.IN)) {
continue;
}
retrieveCallResult(callableStatement, i, resultMap);
}
}
catch (SQLException e)
{
String error = new StringBuilder().append("Failed to call SQL ").append(this.compiledSql).append(" with error ").append(e.getMessage()).append(" errorCode ").append(e.getErrorCode()).toString();
LOGGER.error(error, e);
throw new DatabaseException(error, e);
}
finally {
closeResources(callableStatement, null);
}
return resultMap;
}

}


create or replace
PACKAGE                                         GCAL_PKG_ENQUEUE AS
/* TODO enter package declarations (types, exceptions, methods etc) here */
PROCEDURE TEXT_MESSAGE_ENQUEUE
(
P_TEXT_MESSAGE            IN  VARCHAR2,
P_QUEUE_NAME              IN  VARCHAR2
);
PROCEDURE TEXT_ENQUEUE_WITH_PRIORITY
(
P_TEXT_MESSAGE            IN  VARCHAR2,
P_QUEUE_NAME              IN  VARCHAR2,
P_PRIORITY                IN  NUMBER
);
PROCEDURE TEXT_ENQUEUE_WITH_CORRELATION
(
P_TEXT_MESSAGE            IN  VARCHAR2,
P_QUEUE_NAME              IN  VARCHAR2,
P_PRIORITY                IN  NUMBER,
P_CORRELATION             IN  NUMBER
);

END GCAL_PKG_ENQUEUE;

create or replace
PACKAGE BODY                                                       GCAL_PKG_ENQUEUE AS
PROCEDURE TEXT_MESSAGE_ENQUEUE
(
P_TEXT_MESSAGE            IN  VARCHAR2,
P_QUEUE_NAME              IN  VARCHAR2
) AS
v_defaultPriority  NUMBER;
BEGIN
v_defaultPriority := 5;
TEXT_ENQUEUE_WITH_PRIORITY(P_TEXT_MESSAGE, P_QUEUE_NAME, v_defaultPriority);
END TEXT_MESSAGE_ENQUEUE;
PROCEDURE TEXT_ENQUEUE_WITH_PRIORITY
(
P_TEXT_MESSAGE            IN  VARCHAR2,
P_QUEUE_NAME              IN  VARCHAR2,
P_PRIORITY                IN  NUMBER
) AS
BEGIN
TEXT_ENQUEUE_WITH_CORRELATION(P_TEXT_MESSAGE, P_QUEUE_NAME, P_PRIORITY, NULL);
END TEXT_ENQUEUE_WITH_PRIORITY;

PROCEDURE TEXT_ENQUEUE_WITH_CORRELATION
(
P_TEXT_MESSAGE            IN  VARCHAR2,
P_QUEUE_NAME              IN  VARCHAR2,
P_PRIORITY                IN  NUMBER,
P_CORRELATION             IN  NUMBER
) AS
AGENT              SYS.AQ$_AGENT := SYS.AQ$_AGENT(' ', NULL, 0);
MESSAGE            SYS.AQ$_JMS_TEXT_MESSAGE;
ENQUEUE_OPTIONS    DBMS_AQ.ENQUEUE_OPTIONS_T;
MESSAGE_PROPERTIES DBMS_AQ.MESSAGE_PROPERTIES_T;
MSGID              RAW(16);
JAVA_EXP           EXCEPTION;
PRAGMA             EXCEPTION_INIT(JAVA_EXP, -24197);
BEGIN
MESSAGE := SYS.AQ$_JMS_TEXT_MESSAGE.CONSTRUCT;
MESSAGE.SET_TEXT(P_TEXT_MESSAGE);
MESSAGE.SET_REPLYTO(AGENT);
MESSAGE.SET_TYPE('JMS_TEXT_MESSAGE');
MESSAGE.SET_USERID(SYS_CONTEXT('USERENV','SESSION_USER'));
MESSAGE.SET_APPID(SYS_CONTEXT('USERENV','HOST'));
MESSAGE_PROPERTIES.PRIORITY := P_PRIORITY;
IF P_CORRELATION IS NOT NULL THEN
MESSAGE_PROPERTIES.CORRELATION := P_CORRELATION;
END IF;
-- Enqueue this message into AQ queue using DBMS_AQ package
DBMS_AQ.ENQUEUE(QUEUE_NAME => P_QUEUE_NAME,
ENQUEUE_OPTIONS => ENQUEUE_OPTIONS,
MESSAGE_PROPERTIES => MESSAGE_PROPERTIES,
PAYLOAD => MESSAGE,
MSGID => MSGID);
EXCEPTION
WHEN JAVA_EXP THEN
RAISE;
WHEN OTHERS THEN
RAISE;
END TEXT_ENQUEUE_WITH_CORRELATION;
END GCAL_PKG_ENQUEUE;