in modules/core/src/main/java/org/apache/synapse/core/axis2/SynapseCallbackReceiver.java [251:474]
private void handleMessage(String messageID ,MessageContext response,
org.apache.synapse.MessageContext synapseOutMsgCtx, AsyncCallback callback) throws AxisFault {
Object o = response.getProperty(SynapseConstants.SENDING_FAULT);
if (o != null && Boolean.TRUE.equals(o)) {
Pipe pipe = (Pipe) ((Axis2MessageContext) synapseOutMsgCtx).getAxis2MessageContext()
.getProperty(PassThroughConstants.PASS_THROUGH_PIPE);
if (pipe != null && pipe.isSerializationComplete()) {
NHttpServerConnection conn = (NHttpServerConnection) ((Axis2MessageContext) synapseOutMsgCtx).
getAxis2MessageContext().getProperty(PassThroughConstants.PASS_THROUGH_SOURCE_CONNECTION);
SourceConfiguration sourceConfiguration = (SourceConfiguration) ((Axis2MessageContext) synapseOutMsgCtx)
.getAxis2MessageContext().getProperty(PassThroughConstants.PASS_THROUGH_SOURCE_CONFIGURATION);
Pipe newPipe = new Pipe(conn, sourceConfiguration.getBufferFactory().getBuffer(),
PassThroughConstants.SOURCE, sourceConfiguration);
((Axis2MessageContext) synapseOutMsgCtx).getAxis2MessageContext()
.setProperty(PassThroughConstants.PASS_THROUGH_PIPE, newPipe);
}
StatisticsReporter.reportFaultForAll(synapseOutMsgCtx,
ErrorLogFactory.createErrorLog(response));
// there is a sending fault. propagate the fault to fault handlers.
Stack faultStack = synapseOutMsgCtx.getFaultStack();
if (faultStack != null && !faultStack.isEmpty()) {
// if we have access to the full synapseOutMsgCtx.getEnvelope(), then let
// it flow with the error details. Else, replace its envelope with the
// fault envelope
try {
synapseOutMsgCtx.getEnvelope().build();
} catch (OMException x) {
synapseOutMsgCtx.setEnvelope(response.getEnvelope());
}
Exception e = (Exception) response.getProperty(SynapseConstants.ERROR_EXCEPTION);
synapseOutMsgCtx.setProperty(SynapseConstants.SENDING_FAULT, Boolean.TRUE);
synapseOutMsgCtx.setProperty(SynapseConstants.ERROR_CODE,
response.getProperty(SynapseConstants.ERROR_CODE));
synapseOutMsgCtx.setProperty(SynapseConstants.ERROR_MESSAGE,
response.getProperty(SynapseConstants.ERROR_MESSAGE));
synapseOutMsgCtx.setProperty(SynapseConstants.ERROR_DETAIL,
response.getProperty(SynapseConstants.ERROR_DETAIL));
synapseOutMsgCtx.setProperty(SynapseConstants.ERROR_EXCEPTION, e);
if (log.isDebugEnabled()) {
log.debug("[Failed Request Message ID : " + messageID + "]" +
" [New to be Retried Request Message ID : " +
synapseOutMsgCtx.getMessageID() + "]");
}
int errorCode = (Integer)response.getProperty(SynapseConstants.ERROR_CODE);
// If a timeout has occurred and the timeout action of the callback is to
// discard the message
if (errorCode == SynapseConstants.NHTTP_CONNECTION_TIMEOUT &&
callback.getTimeOutAction() == SynapseConstants.DISCARD) {
// Do not execute any fault sequences. Discard message
log.warn("Synapse timed out for the request with Message ID : " + messageID +
". Ignoring fault handlers since the timeout action is DISCARD.");
faultStack.removeAllElements();
} else {
((FaultHandler) faultStack.pop()).handleFault(synapseOutMsgCtx, null);
}
}
} else {
// there can always be only one instance of an Endpoint in the faultStack of a message
// if the send was successful, so remove it before we proceed any further
Stack faultStack = synapseOutMsgCtx.getFaultStack();
if (faultStack != null && !faultStack.isEmpty()
&& faultStack.peek() instanceof Endpoint) {
Endpoint successfulEndpoint = (Endpoint) faultStack.pop();
successfulEndpoint.onSuccess();
}
if (log.isDebugEnabled()) {
log.debug("Synapse received an asynchronous response message");
log.debug("Received To: " +
(response.getTo() != null ? response.getTo().getAddress() : "null"));
log.debug("SOAPAction: " +
(response.getSoapAction() != null ? response.getSoapAction() : "null"));
log.debug("WSA-Action: " +
(response.getWSAAction() != null ? response.getWSAAction() : "null"));
String[] cids = response.getAttachmentMap().getAllContentIDs();
if (cids != null && cids.length > 0) {
for (String cid : cids) {
log.debug("Attachment : " + cid);
}
}
log.debug("Body : \n" + response.getEnvelope());
}
MessageContext axisOutMsgCtx =
((Axis2MessageContext) synapseOutMsgCtx).getAxis2MessageContext();
//Processes 'Accept-Encoding'
ResponseAcceptEncodingProcessor.process(response, axisOutMsgCtx);
response.setServiceContext(null);
response.setOperationContext(axisOutMsgCtx.getOperationContext());
response.setAxisMessage(axisOutMsgCtx.getAxisOperation().getMessage(
WSDLConstants.MESSAGE_LABEL_OUT_VALUE));
// set properties on response
response.setServerSide(true);
response.setProperty(SynapseConstants.ISRESPONSE_PROPERTY, Boolean.TRUE);
response.setProperty(MessageContext.TRANSPORT_OUT,
axisOutMsgCtx.getProperty(MessageContext.TRANSPORT_OUT));
response.setProperty(org.apache.axis2.Constants.OUT_TRANSPORT_INFO,
axisOutMsgCtx.getProperty(org.apache.axis2.Constants.OUT_TRANSPORT_INFO));
response.setTransportIn(axisOutMsgCtx.getTransportIn());
response.setTransportOut(axisOutMsgCtx.getTransportOut());
// If request is REST assume that the response is REST too
response.setDoingREST(axisOutMsgCtx.isDoingREST());
if (axisOutMsgCtx.isDoingMTOM()) {
response.setDoingMTOM(true);
response.setProperty(
org.apache.axis2.Constants.Configuration.ENABLE_MTOM,
org.apache.axis2.Constants.VALUE_TRUE);
}
if (axisOutMsgCtx.isDoingSwA()) {
response.setDoingSwA(true);
response.setProperty(
org.apache.axis2.Constants.Configuration.ENABLE_SWA,
org.apache.axis2.Constants.VALUE_TRUE);
}
// when axis2 receives a soap message without addressing headers it users
// DISABLE_ADDRESSING_FOR_OUT_MESSAGES property to keep it and hence avoid addressing
// headers on the response. this causes a problem for synapse if the original message
// it receives (from client) has addressing and the synapse service invocation has not
// engage addressing. in this case when synapse receives the response from the server
// addressing In handler disable addressing since that response does not have addressing
// headers. synapse sends the response to its original client using the same message
// context. Then this response does not have addressing headers since it already
// disable. to avoid this we need to set the DISABLE_ADDRESSING_FOR_OUT_MESSAGES
// property state to original state.
if (axisOutMsgCtx.getProperty(
AddressingConstants.DISABLE_ADDRESSING_FOR_OUT_MESSAGES) != null) {
response.setProperty(AddressingConstants.DISABLE_ADDRESSING_FOR_OUT_MESSAGES,
axisOutMsgCtx.getProperty(
AddressingConstants.DISABLE_ADDRESSING_FOR_OUT_MESSAGES));
} else {
response.removeProperty(AddressingConstants.DISABLE_ADDRESSING_FOR_OUT_MESSAGES);
}
Object messageType = axisOutMsgCtx.getProperty(
org.apache.axis2.Constants.Configuration.MESSAGE_TYPE);
if (!HTTPConstants.MEDIA_TYPE_X_WWW_FORM.equals(messageType)) {
// copy the message type property that's used by the out message to the
// response message
response.setProperty(org.apache.axis2.Constants.Configuration.MESSAGE_TYPE,
messageType);
}
// compare original received message (axisOutMsgCtx) soap version with the response
// if they are different change to original version
if(axisOutMsgCtx.isSOAP11() != response.isSOAP11()) {
if(axisOutMsgCtx.isSOAP11()) {
SOAPUtils.convertSOAP12toSOAP11(response);
} else {
SOAPUtils.convertSOAP11toSOAP12(response);
}
}
if (axisOutMsgCtx.getMessageID() != null) {
response.setRelationships(
new RelatesTo[]{new RelatesTo(axisOutMsgCtx.getMessageID())});
}
response.setReplyTo(axisOutMsgCtx.getReplyTo());
response.setFaultTo(axisOutMsgCtx.getFaultTo());
if (axisOutMsgCtx.isPropertyTrue(NhttpConstants.IGNORE_SC_ACCEPTED)) {
response.setProperty(NhttpConstants.FORCE_SC_ACCEPTED, Constants.VALUE_TRUE);
}
// create the synapse message context for the response
Axis2MessageContext synapseInMessageContext =
new Axis2MessageContext(
response,
synapseOutMsgCtx.getConfiguration(),
synapseOutMsgCtx.getEnvironment());
synapseInMessageContext.setResponse(true);
synapseInMessageContext.setTo(
new EndpointReference(AddressingConstants.Final.WSA_ANONYMOUS_URL));
synapseInMessageContext.setTracingState(synapseOutMsgCtx.getTracingState());
// set the properties of the original MC to the new MC
for (Object key : synapseOutMsgCtx.getPropertyKeySet()) {
synapseInMessageContext.setProperty(
(String) key, synapseOutMsgCtx.getProperty((String) key));
}
// If this response is related to session affinity endpoints -Server initiated session
Dispatcher dispatcher =
(Dispatcher) synapseOutMsgCtx.getProperty(
SynapseConstants.PROP_SAL_ENDPOINT_CURRENT_DISPATCHER);
if (dispatcher != null && dispatcher.isServerInitiatedSession()) {
dispatcher.updateSession(synapseInMessageContext);
}
StatisticsReporter.reportForAllOnResponseReceived(synapseInMessageContext);
// send the response message through the synapse mediation flow
try {
synapseOutMsgCtx.getEnvironment().injectMessage(synapseInMessageContext);
} catch (SynapseException syne) {
Stack stack = synapseInMessageContext.getFaultStack();
if (stack != null &&
!stack.isEmpty()) {
((FaultHandler) stack.pop()).handleFault(synapseInMessageContext, syne);
} else {
log.error("Synapse encountered an exception, " +
"No error handlers found - [Message Dropped]\n" + syne.getMessage());
}
}
}
}