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 java.lang.reflect.Method;
020 import java.util.Map;
021 import java.util.Set;
022 import java.util.HashSet;
023
024 import javax.jbi.JBIException;
025 import javax.jbi.component.ComponentContext;
026 import javax.jbi.component.ComponentLifeCycle;
027 import javax.jbi.management.LifeCycleMBean;
028 import javax.jbi.messaging.DeliveryChannel;
029 import javax.jbi.messaging.ExchangeStatus;
030 import javax.jbi.messaging.MessageExchange;
031 import javax.jbi.messaging.MessagingException;
032 import javax.jbi.messaging.MessageExchange.Role;
033 import javax.jbi.servicedesc.ServiceEndpoint;
034 import javax.management.MBeanServer;
035 import javax.management.ObjectName;
036 import javax.transaction.Status;
037 import javax.transaction.Transaction;
038 import javax.transaction.TransactionManager;
039 import javax.transaction.SystemException;
040 import javax.xml.namespace.QName;
041
042 import org.apache.commons.logging.Log;
043 import org.apache.servicemix.executors.Executor;
044 import org.apache.servicemix.executors.ExecutorFactory;
045 import org.apache.servicemix.executors.impl.ExecutorFactoryImpl;
046
047 import java.util.concurrent.ConcurrentHashMap;
048 import java.util.concurrent.atomic.AtomicBoolean;
049
050 /**
051 * Base class for life cycle management of components. This class may be used as
052 * is.
053 *
054 * @author Guillaume Nodet
055 * @version $Revision: 399873 $
056 * @since 3.0
057 */
058 public class AsyncBaseLifeCycle implements ComponentLifeCycle {
059
060 public static final String INITIALIZED = "Initialized";
061
062 protected transient Log logger;
063
064 protected ServiceMixComponent component;
065
066 protected ComponentContext context;
067
068 protected ObjectName mbeanName;
069
070 protected ExecutorFactory executorFactory;
071
072 protected Executor executor;
073
074 protected AtomicBoolean running;
075
076 protected DeliveryChannel channel;
077
078 protected Thread poller;
079
080 protected AtomicBoolean polling;
081
082 protected TransactionManager transactionManager;
083
084 protected boolean workManagerCreated;
085
086 protected ThreadLocal<String> correlationId;
087
088 protected String currentState = LifeCycleMBean.UNKNOWN;
089
090 protected Container container;
091
092 protected Map<String, Set<String>> knownExchanges;
093
094 public AsyncBaseLifeCycle() {
095 this.running = new AtomicBoolean(false);
096 this.polling = new AtomicBoolean(false);
097 this.correlationId = new ThreadLocal<String>();
098 this.knownExchanges = new ConcurrentHashMap<String, Set<String>>();
099 }
100
101 public AsyncBaseLifeCycle(ServiceMixComponent component) {
102 this();
103 setComponent(component);
104 }
105
106 public Container getContainer() {
107 return container;
108 }
109
110 protected void setComponent(ServiceMixComponent component) {
111 this.component = component;
112 this.logger = component.getLogger();
113 }
114
115 /*
116 * (non-Javadoc)
117 *
118 * @see javax.jbi.component.ComponentLifeCycle#getExtensionMBeanName()
119 */
120 public ObjectName getExtensionMBeanName() {
121 return mbeanName;
122 }
123
124 protected Object getExtensionMBean() throws Exception {
125 return null;
126 }
127
128 protected ObjectName createExtensionMBeanName() throws Exception {
129 return this.context.getMBeanNames().createCustomComponentMBeanName("Configuration");
130 }
131
132 public QName getEPRServiceName() {
133 return null;
134 }
135
136 public String getCurrentState() {
137 return currentState;
138 }
139
140 protected void setCurrentState(String currentState) {
141 this.currentState = currentState;
142 }
143
144 public boolean isStarted(){
145 return currentState != null && currentState.equals(LifeCycleMBean.STARTED);
146 }
147
148 /**
149 * @return true if the object is stopped
150 */
151 public boolean isStopped(){
152 return currentState != null && currentState.equals(LifeCycleMBean.STOPPED);
153 }
154
155 /**
156 * @return true if the object is shutDown
157 */
158 public boolean isShutDown(){
159 return currentState != null && currentState.equals(LifeCycleMBean.SHUTDOWN);
160 }
161
162 /**
163 * @return true if the object is shutDown
164 */
165 public boolean isInitialized(){
166 return currentState != null && currentState.equals(INITIALIZED);
167 }
168
169 /**
170 * @return true if the object is shutDown
171 */
172 public boolean isUnknown(){
173 return currentState == null || currentState.equals(LifeCycleMBean.UNKNOWN);
174 }
175
176 /*
177 * (non-Javadoc)
178 *
179 * @see javax.jbi.component.ComponentLifeCycle#init(javax.jbi.component.ComponentContext)
180 */
181 public void init(ComponentContext context) throws JBIException {
182 try {
183 if (logger.isDebugEnabled()) {
184 logger.debug("Initializing component");
185 }
186 Thread.currentThread().setContextClassLoader(component.getClass().getClassLoader());
187 this.context = context;
188 this.channel = context.getDeliveryChannel();
189 try {
190 this.transactionManager = (TransactionManager) context.getTransactionManager();
191 } catch (Throwable e) {
192 // Ignore, this is just a safeguard against non compliant
193 // JBI implementation which throws an exception instead of
194 // return null
195 }
196 container = Container.detect(context);
197 doInit();
198 setCurrentState(INITIALIZED);
199 if (logger.isDebugEnabled()) {
200 logger.debug("Component initialized");
201 }
202 } catch (JBIException e) {
203 throw e;
204 } catch (Exception e) {
205 throw new JBIException("Error calling init", e);
206 }
207 }
208
209 protected void doInit() throws Exception {
210 // Register extension mbean
211 Object mbean = getExtensionMBean();
212 if (mbean != null) {
213 MBeanServer server = this.context.getMBeanServer();
214 if (server == null) {
215 // TODO: log a warning ?
216 // throw new JBIException("null mBeanServer");
217 } else {
218 this.mbeanName = createExtensionMBeanName();
219 if (server.isRegistered(this.mbeanName)) {
220 server.unregisterMBean(this.mbeanName);
221 }
222 server.registerMBean(mbean, this.mbeanName);
223 }
224 }
225 // Obtain or create the work manager
226 // When using the WorkManager from ServiceMix,
227 // some class loader problems can appear when
228 // trying to uninstall the components.
229 // Some threads owned by the work manager have a
230 // security context referencing the component class loader
231 // so that every loaded classes are locked
232 // this.workManager = findWorkManager();
233 if (this.executorFactory == null) {
234 this.executorFactory = findExecutorFactory();
235 }
236 if (this.executorFactory == null) {
237 this.executorFactory = createExecutorFactory();
238 }
239 this.executor = this.executorFactory.createExecutor("component." + getContext().getComponentName());
240 }
241
242 /*
243 * (non-Javadoc)
244 *
245 * @see javax.jbi.component.ComponentLifeCycle#shutDown()
246 */
247 public void shutDown() throws JBIException {
248 try {
249 if (logger.isDebugEnabled()) {
250 logger.debug("Shutting down component");
251 }
252 Thread.currentThread().setContextClassLoader(component.getClass().getClassLoader());
253 doShutDown();
254 setCurrentState(LifeCycleMBean.SHUTDOWN);
255 this.context = null;
256 if (logger.isDebugEnabled()) {
257 logger.debug("Component shut down");
258 }
259 } catch (JBIException e) {
260 throw e;
261 } catch (Exception e) {
262 throw new JBIException("Error calling shutdown", e);
263 }
264 }
265
266 protected void doShutDown() throws Exception {
267 // Unregister mbean
268 if (this.mbeanName != null) {
269 MBeanServer server = this.context.getMBeanServer();
270 if (server == null) {
271 throw new JBIException("null mBeanServer");
272 }
273 if (server.isRegistered(this.mbeanName)) {
274 server.unregisterMBean(this.mbeanName);
275 }
276 }
277 // Destroy excutor
278 executor.shutdown();
279 executor = null;
280 }
281
282 /*
283 * (non-Javadoc)
284 *
285 * @see javax.jbi.component.ComponentLifeCycle#start()
286 */
287 public void start() throws JBIException {
288 try {
289 if (logger.isDebugEnabled()) {
290 logger.debug("Starting component");
291 }
292 Thread.currentThread().setContextClassLoader(component.getClass().getClassLoader());
293 if (this.running.compareAndSet(false, true)) {
294 doStart();
295 setCurrentState(LifeCycleMBean.STARTED);
296 }
297 if (logger.isDebugEnabled()) {
298 logger.debug("Component started");
299 }
300 } catch (JBIException e) {
301 throw e;
302 } catch (Exception e) {
303 throw new JBIException("Error calling start", e);
304 }
305 }
306
307 protected void doStart() throws Exception {
308 if (container.getType() != Container.Type.ServiceMix3) {
309 synchronized (this.polling) {
310 executor.execute(new Runnable() {
311 public void run() {
312 poller = Thread.currentThread();
313 pollDeliveryChannel();
314 }
315 });
316 polling.wait();
317 }
318 }
319 }
320
321 protected void pollDeliveryChannel() {
322 synchronized (polling) {
323 polling.set(true);
324 polling.notify();
325 }
326 while (running.get()) {
327 try {
328 final MessageExchange exchange = channel.accept(1000L);
329 if (exchange != null) {
330 final Transaction tx = (Transaction) exchange
331 .getProperty(MessageExchange.JTA_TRANSACTION_PROPERTY_NAME);
332 if (tx != null && container.handleTransactions()) {
333 if (transactionManager == null) {
334 throw new IllegalStateException(
335 "Exchange is enlisted in a transaction, but no transaction manager is available");
336 }
337 transactionManager.suspend();
338 }
339 executor.execute(new Runnable() {
340 public void run() {
341 processExchangeInTx(exchange, tx);
342 }
343 });
344 }
345 } catch (Throwable t) {
346 if (running.get() == false) {
347 // Should have been interrupted, discard the throwable
348 if (logger.isDebugEnabled()) {
349 logger.debug("Polling thread will stop");
350 }
351 } else {
352 logger.error("Error polling delivery channel", t);
353 }
354 }
355 }
356 synchronized (polling) {
357 polling.set(false);
358 polling.notify();
359 }
360 }
361
362 /*
363 * (non-Javadoc)
364 *
365 * @see javax.jbi.component.ComponentLifeCycle#stop()
366 */
367 public void stop() throws JBIException {
368 try {
369 if (logger.isDebugEnabled()) {
370 logger.debug("Stopping component");
371 }
372 Thread.currentThread().setContextClassLoader(component.getClass().getClassLoader());
373 if (this.running.compareAndSet(true, false)) {
374 doStop();
375 setCurrentState(LifeCycleMBean.STOPPED);
376 }
377 if (logger.isDebugEnabled()) {
378 logger.debug("Component stopped");
379 }
380 } catch (JBIException e) {
381 throw e;
382 } catch (Exception e) {
383 throw new JBIException("Error calling stop", e);
384 }
385 }
386
387 protected void doStop() throws Exception {
388 // Interrupt the polling thread and await termination
389 try {
390 synchronized (polling) {
391 if (polling.get()) {
392 poller.interrupt();
393 polling.wait();
394 }
395 }
396 } finally {
397 poller = null;
398 }
399 }
400
401 /**
402 * @return Returns the context.
403 */
404 public ComponentContext getContext() {
405 return context;
406 }
407
408 public Executor getExecutor() {
409 return executor;
410 }
411
412 public void setExecutor(Executor executor) {
413 this.executor = executor;
414 }
415
416 public ExecutorFactory getExecutorFactory() {
417 return executorFactory;
418 }
419
420 public void setExecutorFactory(ExecutorFactory executorFactory) {
421 this.executorFactory = executorFactory;
422 }
423
424 protected ExecutorFactory createExecutorFactory() {
425 // Create a very simple one
426 return new ExecutorFactoryImpl();
427 }
428
429 public Object getSmx3Container() {
430 if (container instanceof Container.Smx3Container) {
431 return ((Container.Smx3Container) container).getSmx3Container();
432 }
433 return null;
434 }
435
436 protected ExecutorFactory findExecutorFactory() {
437 // If inside ServiceMix, retrieve its executor factory
438 try {
439 Object container = getSmx3Container();
440 if (container != null) {
441 Method getWorkManagerMth = container.getClass().getMethod("getExecutorFactory", new Class[0]);
442 return (ExecutorFactory) getWorkManagerMth.invoke(container, new Object[0]);
443 }
444 } catch (Throwable t) {
445 // Ignore
446 }
447 // TODO: should look in jndi for an existing ExecutorFactory
448 return null;
449 }
450
451 protected void processExchangeInTx(MessageExchange exchange, Transaction tx) {
452 try {
453 if (tx != null) {
454 transactionManager.resume(tx);
455 }
456 processExchange(exchange);
457 } catch (Exception e) {
458 logger.error("Error processing exchange " + exchange, e);
459 try {
460 // If we are transacted, check if this exception should
461 // rollback the transaction
462 if (transactionManager != null && transactionManager.getStatus() == Status.STATUS_ACTIVE) {
463 if (exceptionShouldRollbackTx(e)) {
464 transactionManager.setRollbackOnly();
465 }
466 if (!container.handleTransactions()) {
467 transactionManager.suspend();
468 }
469 }
470 exchange.setError(e);
471 channel.send(exchange);
472 } catch (Exception inner) {
473 logger.error("Error setting exchange status to ERROR", inner);
474 }
475 } finally {
476 try {
477 // Check transaction status
478 if (tx != null) {
479 int status = transactionManager.getStatus();
480 // We use pull delivery, so the transaction should already
481 // have been transfered to another thread because the
482 // component
483 // must have answered.
484 if (status != Status.STATUS_NO_TRANSACTION) {
485 logger.error("Transaction is still active after exchange processing. Trying to rollback transaction.");
486 try {
487 transactionManager.rollback();
488 } catch (Throwable t) {
489 logger.error("Error trying to rollback transaction.", t);
490 }
491 }
492 }
493 } catch (Throwable t) {
494 logger.error("Error checking transaction status.", t);
495 }
496 }
497 }
498
499 protected boolean exceptionShouldRollbackTx(Exception e) {
500 return false;
501 }
502
503 public void onMessageExchange(MessageExchange exchange) {
504 if (!container.handleTransactions()) {
505 final Transaction tx = (Transaction) exchange.getProperty(MessageExchange.JTA_TRANSACTION_PROPERTY_NAME);
506 processExchangeInTx(exchange, tx);
507 return;
508 }
509 try {
510 processExchange(exchange);
511 } catch (Exception e) {
512 logger.error("Error processing exchange " + exchange, e);
513 try {
514 // If we are transacted and this is a runtime exception
515 // try to mark transaction as rollback
516 if (transactionManager != null &&
517 transactionManager.getStatus() == Status.STATUS_ACTIVE &&
518 exceptionShouldRollbackTx(e)) {
519 transactionManager.setRollbackOnly();
520 if (!container.handleTransactions()) {
521 transactionManager.suspend();
522 }
523 }
524 exchange.setError(e);
525 channel.send(exchange);
526 } catch (Exception inner) {
527 logger.error("Error setting exchange status to ERROR", inner);
528 }
529 }
530 }
531
532 protected void processExchange(MessageExchange exchange) throws Exception {
533 if (logger.isDebugEnabled()) {
534 logger.debug("Received exchange: status: " + exchange.getStatus() + ", role: "
535 + (exchange.getRole() == Role.CONSUMER ? "consumer" : "provider"));
536 }
537 if (exchange.getRole() == Role.PROVIDER) {
538 boolean dynamic = false;
539 ServiceEndpoint endpoint = exchange.getEndpoint();
540 String key = EndpointSupport.getKey(exchange.getEndpoint());
541 Endpoint ep = this.component.getRegistry().getEndpoint(key);
542 if (ep == null) {
543 if (endpoint.getServiceName().equals(getEPRServiceName())) {
544 ep = getResolvedEPR(exchange.getEndpoint());
545 ep.activate();
546 ep.start();
547 dynamic = true;
548 }
549 if (ep == null) {
550 throw new IllegalStateException("Endpoint not found: " + key);
551 }
552 }
553 try {
554 doProcess(ep, exchange);
555 } finally {
556 // If the endpoint is dynamic, deactivate it
557 if (dynamic) {
558 ep.stop();
559 ep.deactivate();
560 }
561 }
562 } else {
563 Endpoint ep = null;
564 if (exchange.getProperty(JbiConstants.SENDER_ENDPOINT) != null) {
565 String key = exchange.getProperty(JbiConstants.SENDER_ENDPOINT).toString();
566 ep = this.component.getRegistry().getEndpoint(key);
567 }
568 if (ep == null) {
569 throw new IllegalStateException("Endpoint not found for: " + exchange.getExchangeId());
570 }
571 doProcess(ep, exchange);
572 }
573
574 }
575
576 /**
577 * Thin wrapper around the call to the processor to ensure that the Endpoints
578 * classloader is used where available
579 *
580 */
581 private void doProcess(Endpoint ep, MessageExchange exchange) throws Exception {
582 ClassLoader oldCl = Thread.currentThread().getContextClassLoader();
583 boolean processed = false;
584 try {
585 ClassLoader cl = (ep != null) ? ep.getServiceUnit().getConfigurationClassLoader() : null;
586 if (cl != null) {
587 Thread.currentThread().setContextClassLoader(cl);
588 }
589 // Read the correlation id from the exchange and set it in the correlation id property
590 String correlationID = (String)exchange.getProperty(JbiConstants.CORRELATION_ID);
591 if (correlationID != null) {
592 // Set the id in threadlocal variable
593 correlationId.set(correlationID);
594 }
595 if (logger.isDebugEnabled()) {
596 logger.debug("Retrieved correlation id: " + correlationID);
597 }
598 EndpointDeliveryChannel.setEndpoint(ep);
599 handleExchange(ep, exchange, exchange.getStatus() == ExchangeStatus.ACTIVE);
600 ep.process(exchange);
601 processed = true;
602 } finally {
603 if (!processed) {
604 handleExchange(ep, exchange, false);
605 }
606 EndpointDeliveryChannel.setEndpoint(null);
607 Thread.currentThread().setContextClassLoader(oldCl);
608 // Clean the threadlocal variable
609 correlationId.set(null);
610 }
611 }
612
613 public void prepareExchange(MessageExchange exchange, Endpoint endpoint) throws MessagingException {
614 if (exchange.getRole() == Role.CONSUMER) {
615 // Check if a correlation id is already set on the exchange, otherwise create it
616 String correlationIDValue = (String) exchange.getProperty(JbiConstants.CORRELATION_ID);
617 if (correlationIDValue == null) {
618 // Retrieve correlation id from thread local variable, if exist
619 correlationIDValue = correlationId.get();
620 if (correlationIDValue == null) {
621 // Set a correlation id property that have to be propagated in all components
622 // to trace the process instance
623 correlationIDValue = exchange.getExchangeId();
624 exchange.setProperty(JbiConstants.CORRELATION_ID, exchange.getExchangeId());
625 if (logger.isDebugEnabled()) {
626 logger.debug("Created correlation id: " + correlationIDValue);
627 }
628 } else {
629 // Use correlation id retrieved from previous message exchange
630 exchange.setProperty(JbiConstants.CORRELATION_ID, correlationIDValue);
631 if (logger.isDebugEnabled()) {
632 logger.debug("Correlation id retrieved from ThreadLocal: " + correlationIDValue);
633 }
634 }
635 }
636 // Set the sender endpoint property
637 exchange.setProperty(JbiConstants.SENDER_ENDPOINT, endpoint.getKey());
638 }
639 // Handle transaction
640 if (!container.handleTransactions()) {
641 try {
642 if ((exchange.getRole() == Role.CONSUMER && exchange.getStatus() == ExchangeStatus.ACTIVE) || exchange.getRole() == Role.PROVIDER) {
643 if (transactionManager != null) {
644 exchange.setProperty(MessageExchange.JTA_TRANSACTION_PROPERTY_NAME, transactionManager.suspend());
645 }
646 }
647 } catch (SystemException e) {
648 throw new MessagingException("Error handling transaction", e);
649 }
650 }
651 }
652
653 public void prepareShutdown(Endpoint endpoint) throws InterruptedException {
654 Set<String> exchanges = getKnownExchanges(endpoint);
655 synchronized (exchanges) {
656 if (!exchanges.isEmpty()) {
657 exchanges.wait();
658 }
659 }
660 }
661
662 protected Set<String> getKnownExchanges(Endpoint endpoint) {
663 Set<String> exchanges = knownExchanges.get(endpoint.getKey());
664 if (exchanges == null) {
665 synchronized (knownExchanges) {
666 exchanges = knownExchanges.get(endpoint.getKey());
667 if (exchanges == null) {
668 exchanges = new HashSet<String>();
669 knownExchanges.put(endpoint.getKey(), exchanges);
670 }
671 }
672 }
673 return exchanges;
674 }
675
676 public void handleExchange(Endpoint endpoint, MessageExchange exchange, boolean add) {
677 Set<String> exchanges = getKnownExchanges(endpoint);
678 synchronized (exchanges) {
679 if (add) {
680 exchanges.add(exchange.getExchangeId());
681 } else {
682 exchanges.remove(exchange.getExchangeId());
683 }
684 exchanges.notifyAll();
685 }
686 }
687
688 /**
689 * Handle an exchange sent to an EPR resolved by this component
690 *
691 * @param ep the service endpoint
692 * @return an endpoint to use for handling the exchange
693 * @throws Exception
694 */
695 protected Endpoint getResolvedEPR(ServiceEndpoint ep) throws Exception {
696 throw new UnsupportedOperationException("Component does not handle EPR exchanges");
697 }
698
699 }