001 /*
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements. See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License. You may obtain a copy of the License at
008 *
009 * http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017 package org.apache.servicemix.common;
018
019 import javax.jbi.component.ComponentContext;
020 import javax.jbi.messaging.DeliveryChannel;
021 import javax.jbi.messaging.MessageExchange;
022 import javax.jbi.messaging.MessageExchangeFactory;
023 import javax.jbi.messaging.MessagingException;
024 import javax.jbi.messaging.ExchangeStatus;
025 import javax.jbi.servicedesc.ServiceEndpoint;
026 import javax.xml.namespace.QName;
027 import javax.transaction.Transaction;
028 import javax.transaction.TransactionManager;
029 import javax.transaction.InvalidTransactionException;
030 import javax.transaction.SystemException;
031
032 /**
033 * This class is a wrapper around an existing DeliveryChannel
034 * that will be given to service engine endpoints so that
035 * they are able to send messages and to interact with the
036 * JBI container.
037 *
038 * @author gnodet
039 */
040 public class EndpointDeliveryChannel implements DeliveryChannel {
041
042 private static final ThreadLocal<Endpoint> ENDPOINT_TLS = new ThreadLocal<Endpoint>();
043
044 private final DeliveryChannel channel;
045 private final Endpoint endpoint;
046
047 public EndpointDeliveryChannel(Endpoint endpoint) throws MessagingException {
048 this.endpoint = endpoint;
049 this.channel = endpoint.getServiceUnit().getComponent().getComponentContext().getDeliveryChannel();
050 }
051
052 public EndpointDeliveryChannel(ComponentContext context) throws MessagingException {
053 this.endpoint = null;
054 this.channel = context.getDeliveryChannel();
055 }
056
057 public MessageExchange accept() throws MessagingException {
058 throw new UnsupportedOperationException();
059 }
060
061 public MessageExchange accept(long timeout) throws MessagingException {
062 throw new UnsupportedOperationException();
063 }
064
065 public void close() throws MessagingException {
066 throw new UnsupportedOperationException();
067 }
068
069 public MessageExchangeFactory createExchangeFactory() {
070 return channel.createExchangeFactory();
071 }
072
073 public MessageExchangeFactory createExchangeFactory(QName interfaceName) {
074 return channel.createExchangeFactory(interfaceName);
075 }
076
077 public MessageExchangeFactory createExchangeFactory(ServiceEndpoint endpoint) {
078 return channel.createExchangeFactory(endpoint);
079 }
080
081 public MessageExchangeFactory createExchangeFactoryForService(QName serviceName) {
082 return channel.createExchangeFactoryForService(serviceName);
083 }
084
085 public void send(MessageExchange exchange) throws MessagingException {
086 prepareExchange(exchange);
087 handleExchange(exchange, exchange.getStatus() == ExchangeStatus.ACTIVE);
088 channel.send(exchange);
089 }
090
091 public boolean sendSync(MessageExchange exchange, long timeout) throws MessagingException {
092 boolean processed = false;
093 try {
094 prepareExchange(exchange);
095 handleExchange(exchange, exchange.getStatus() == ExchangeStatus.ACTIVE);
096 boolean ret = channel.sendSync(exchange, timeout);
097 handleExchange(exchange, exchange.getStatus() == ExchangeStatus.ACTIVE);
098 if (ret) {
099 resumeTx(exchange);
100 processed = true;
101 }
102 return ret;
103 } finally {
104 if (!processed) {
105 handleExchange(exchange, false);
106 }
107 }
108 }
109
110 public boolean sendSync(MessageExchange exchange) throws MessagingException {
111 boolean processed = false;
112 try {
113 prepareExchange(exchange);
114 handleExchange(exchange, exchange.getStatus() == ExchangeStatus.ACTIVE);
115 boolean ret = channel.sendSync(exchange);
116 handleExchange(exchange, exchange.getStatus() == ExchangeStatus.ACTIVE);
117 if (ret) {
118 resumeTx(exchange);
119 processed = true;
120 }
121 return ret;
122 } finally {
123 if (!processed) {
124 handleExchange(exchange, false);
125 }
126 }
127 }
128
129 private void resumeTx(MessageExchange exchange) throws MessagingException {
130 if (!getEndpoint().getServiceUnit().getComponent().getContainer().handleTransactions()) {
131 Transaction tx = (Transaction) exchange.getProperty(MessageExchange.JTA_TRANSACTION_PROPERTY_NAME);
132 if (tx != null) {
133 TransactionManager txmgr = (TransactionManager) endpoint.getServiceUnit().getComponent().getComponentContext().getTransactionManager();
134 try {
135 txmgr.resume(tx);
136 } catch (InvalidTransactionException e) {
137 throw new MessagingException(e);
138 } catch (SystemException e) {
139 throw new MessagingException(e);
140 }
141 }
142 }
143 }
144
145 protected void prepareExchange(MessageExchange exchange) throws MessagingException {
146 Endpoint ep = getEndpoint();
147 ep.getServiceUnit().getComponent().prepareExchange(exchange, ep);
148 }
149
150 protected void handleExchange(MessageExchange exchange, boolean add) throws MessagingException {
151 Endpoint ep = getEndpoint();
152 ep.getServiceUnit().getComponent().handleExchange(ep, exchange, add);
153 }
154
155 protected Endpoint getEndpoint() {
156 if (endpoint != null) {
157 return endpoint;
158 }
159 return ENDPOINT_TLS.get();
160 }
161
162 public static void setEndpoint(Endpoint endpoint) {
163 ENDPOINT_TLS.set(endpoint);
164 }
165 }