KickJava   Java API By Example, From Geeks To Geeks.

Java > Open Source Codes > org > apache > servicemix > jms > multiplexing > MultiplexingConsumerProcessor


1 /*
2  * Licensed to the Apache Software Foundation (ASF) under one or more
3  * contributor license agreements. See the NOTICE file distributed with
4  * this work for additional information regarding copyright ownership.
5  * The ASF licenses this file to You under the Apache License, Version 2.0
6  * (the "License"); you may not use this file except in compliance with
7  * the License. You may obtain a copy of the License at
8  *
9  * http://www.apache.org/licenses/LICENSE-2.0
10  *
11  * Unless required by applicable law or agreed to in writing, software
12  * distributed under the License is distributed on an "AS IS" BASIS,
13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14  * See the License for the specific language governing permissions and
15  * limitations under the License.
16  */

17 package org.apache.servicemix.jms.multiplexing;
18
19 import java.util.Map JavaDoc;
20
21 import javax.jbi.messaging.DeliveryChannel;
22 import javax.jbi.messaging.ExchangeStatus;
23 import javax.jbi.messaging.MessageExchange;
24 import javax.jms.Destination JavaDoc;
25 import javax.jms.Message JavaDoc;
26 import javax.jms.MessageConsumer JavaDoc;
27 import javax.jms.MessageListener JavaDoc;
28 import javax.jms.MessageProducer JavaDoc;
29 import javax.jms.Session JavaDoc;
30 import javax.naming.InitialContext JavaDoc;
31 import javax.resource.spi.work.Work JavaDoc;
32 import javax.resource.spi.work.WorkException JavaDoc;
33
34 import org.apache.servicemix.common.BaseLifeCycle;
35 import org.apache.servicemix.jms.AbstractJmsProcessor;
36 import org.apache.servicemix.jms.JmsEndpoint;
37 import org.apache.servicemix.soap.Context;
38
39 import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentHashMap;
40
41 public class MultiplexingConsumerProcessor extends AbstractJmsProcessor implements MessageListener JavaDoc {
42
43     protected Session JavaDoc session;
44     protected Destination JavaDoc destination;
45     protected MessageConsumer JavaDoc consumer;
46     protected Map JavaDoc pendingMessages = new ConcurrentHashMap();
47     protected DeliveryChannel channel;
48
49     public MultiplexingConsumerProcessor(JmsEndpoint endpoint) {
50         super(endpoint);
51     }
52
53     protected void doStart(InitialContext JavaDoc ctx) throws Exception JavaDoc {
54         session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
55         destination = endpoint.getDestination();
56         if (destination == null) {
57             if (endpoint.getJndiDestinationName() != null) {
58                 destination = (Destination JavaDoc) ctx.lookup(endpoint.getJndiDestinationName());
59             } else if (endpoint.getJmsProviderDestinationName() != null) {
60                 if (STYLE_QUEUE.equals(endpoint.getDestinationStyle())) {
61                     destination = session.createQueue(endpoint.getJmsProviderDestinationName());
62                 } else {
63                     destination = session.createTopic(endpoint.getJmsProviderDestinationName());
64                 }
65             } else {
66                 throw new IllegalStateException JavaDoc("No destination provided");
67             }
68         }
69         consumer = session.createConsumer(destination);
70         consumer.setMessageListener(this);
71         channel = endpoint.getServiceUnit().getComponent().getComponentContext().getDeliveryChannel();
72     }
73
74     protected void doStop() throws Exception JavaDoc {
75         session = null;
76         destination = null;
77         consumer = null;
78         pendingMessages.clear();
79     }
80
81     public void onMessage(final Message JavaDoc message) {
82         try {
83             if (log.isDebugEnabled()) {
84                 log.debug("Received jms message " + message);
85             }
86             endpoint.getServiceUnit().getComponent().getWorkManager().scheduleWork(new Work JavaDoc() {
87                 public void release() {
88                 }
89                 public void run() {
90                     try {
91                         if (log.isDebugEnabled()) {
92                             log.debug("Handling jms message " + message);
93                         }
94                         Context JavaDoc context = createContext();
95                         MessageExchange exchange = toNMS(message, context);
96                         // TODO: copy protocol messages
97
//inMessage.setProperty(JbiConstants.PROTOCOL_HEADERS, getHeaders(message));
98
pendingMessages.put(exchange.getExchangeId(), context);
99                         BaseLifeCycle lf = (BaseLifeCycle) endpoint.getServiceUnit().getComponent().getLifeCycle();
100                         lf.sendConsumerExchange(exchange, MultiplexingConsumerProcessor.this.endpoint);
101                     } catch (Throwable JavaDoc e) {
102                         log.error("Error while handling jms message", e);
103                     }
104                 }
105             });
106         } catch (WorkException JavaDoc e) {
107             log.error("Error while handling jms message", e);
108         }
109     }
110
111     public void process(MessageExchange exchange) throws Exception JavaDoc {
112         Context JavaDoc context = (Context JavaDoc) pendingMessages.remove(exchange.getExchangeId());
113         Message JavaDoc message = (Message JavaDoc) context.getProperty(Message JavaDoc.class.getName());
114         MessageProducer JavaDoc producer = null;
115         Message JavaDoc response = null;
116         try {
117             response = fromNMSResponse(exchange, context, session);
118             if (response != null) {
119                 producer = session.createProducer(message.getJMSReplyTo());
120                 response.setJMSCorrelationID(message.getJMSCorrelationID());
121                 producer.send(response);
122             }
123         } finally {
124             if (producer != null) {
125                 producer.close();
126             }
127             if (exchange.getStatus() == ExchangeStatus.ACTIVE) {
128                 exchange.setStatus(ExchangeStatus.DONE);
129                 channel.send(exchange);
130             }
131         }
132     }
133
134 }
135
Popular Tags