KickJava   Java API By Example, From Geeks To Geeks.

Java > Open Source Codes > org > apache > sandesha > server > msgprocessors > CompositeProcessor


1 /*
2  * Copyright 1999-2004 The Apache Software Foundation.
3  *
4  * Licensed under the Apache License, Version 2.0 (the "License");
5  * you may not use this file except in compliance with the License.
6  * You may obtain a copy of the License at
7  *
8  * http://www.apache.org/licenses/LICENSE-2.0
9  *
10  * Unless required by applicable law or agreed to in writing, software
11  * distributed under the License is distributed on an "AS IS" BASIS,
12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13  * See the License for the specific language governing permissions and
14  * limitations under the License.
15  *
16  */
17 package org.apache.sandesha.server.msgprocessors;
18
19 import org.apache.axis.AxisFault;
20 import org.apache.axis.Message;
21 import org.apache.axis.MessageContext;
22 import org.apache.axis.components.logger.LogFactory;
23 import org.apache.axis.message.addressing.AddressingHeaders;
24 import org.apache.axis.message.addressing.RelatesTo;
25 import org.apache.commons.logging.Log;
26 import org.apache.sandesha.Constants;
27 import org.apache.sandesha.IStorageManager;
28 import org.apache.sandesha.RMMessageContext;
29 import org.apache.sandesha.storage.dao.SandeshaQueueDAO;
30 import org.apache.sandesha.ws.rm.RMHeaders;
31
32 import javax.xml.namespace.QName JavaDoc;
33
34 /**
35  * This will process messages with sequence elements. Mainly the requests/responses.
36  *
37  * @author Jaliya Ekanayake
38  */
39 public class CompositeProcessor implements IRMMessageProcessor {
40
41     private IStorageManager storageManager;
42     private static final Log log = LogFactory.getLog(SandeshaQueueDAO.class.getName());
43
44     public CompositeProcessor(IStorageManager storageManger) {
45         storageManager = storageManger;
46     }
47
48     public boolean processMessage(RMMessageContext rmMessageContext) throws AxisFault {
49
50         RMHeaders rmHeaders = rmMessageContext.getRMHeaders();
51         AddressingHeaders addrHeaders = rmMessageContext.getAddressingHeaders();
52         AcknowledgementProcessor ackProcessor = new AcknowledgementProcessor(storageManager);
53         if (rmHeaders.getSequenceAcknowledgement() != null) {
54             ackProcessor.processMessage(rmMessageContext);
55         }
56
57         if (rmHeaders.getSequence() != null) {
58             if (rmHeaders.getSequence().getMessageNumber() != null) {
59                 String JavaDoc sequenceUUID = rmHeaders.getSequence().getIdentifier().getIdentifier();
60                 long messageNumber = rmHeaders.getSequence().getMessageNumber().getMessageNumber();
61
62                 String JavaDoc seqId = storageManager.getOutgoingSeqenceIdOfIncomingMsg(rmMessageContext);
63                 boolean hasSequence = storageManager.isSequenceExist(seqId);
64
65                 if (addrHeaders.getRelatesTo() != null && !addrHeaders.getRelatesTo().isEmpty()) {
66                     RelatesTo relatesTo = (RelatesTo) addrHeaders.getRelatesTo().get(0);
67                     String JavaDoc messageId = relatesTo.getURI().toString();
68                     seqId = storageManager.getOutgoingSeqOfMsg(messageId);
69                 }
70                 if (!hasSequence) {
71                     storageManager.addIncomingSequence(seqId);
72                 }
73                 if (!storageManager.isMessageExist(seqId, messageNumber)) {
74                     //Create a copy of the RMMessageContext.
75 RMMessageContext rmMsgContext = new RMMessageContext();
76                     //Copy the RMMEssageContext
77 rmMessageContext.copyContents(rmMsgContext);
78                     rmMsgContext.setSequenceID(sequenceUUID);
79                     rmMsgContext.setMsgNumber(messageNumber);
80                     try {
81                         MessageContext msgContext = new MessageContext(rmMessageContext.getMsgContext().getAxisEngine());
82                         RMMessageContext.copyMessageContext(rmMessageContext.getMsgContext(),
83                                 msgContext);
84                         String JavaDoc soapMsg = rmMessageContext.getMsgContext().getRequestMessage()
85                                 .getSOAPEnvelope()
86                                 .toString();
87                         Message reqMsg = new Message(soapMsg);
88
89                         msgContext.setRequestMessage(reqMsg);
90                         rmMsgContext.setMsgContext(msgContext);
91                         rmMsgContext.setMessageType(Constants.MSG_TYPE_SERVICE_REQUEST);
92                     } catch (Exception JavaDoc e) {
93                         log.error(e);
94                         throw new AxisFault(new QName JavaDoc(Constants.FaultCodes.WSRM_SERVER_INTERNAL_ERROR),
95                                 Constants.FaultMessages.SERVER_INTERNAL_ERROR, null, null);
96                     }
97                     storageManager.insertIncomingMessage(rmMsgContext);
98                 }
99
100                 // refresh the ack for every message arrived
101 //But send only if needed.
102 if (rmHeaders.getAckRequest() != null ||
103                         rmHeaders.getSequence().getLastMessage() != null) {
104                     storageManager.sendAck(sequenceUUID);
105                     return ackProcessor.sendAcknowledgement(rmMessageContext);
106                 } else {
107                     boolean sync = ackProcessor.sendAcknowledgement(rmMessageContext);
108                     return sync;
109                 }
110             }
111         }
112         return false;
113     }
114
115 }
Popular Tags