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.camel.processor.resequencer;
018
019 import java.lang.reflect.InvocationHandler;
020 import java.lang.reflect.Method;
021 import java.lang.reflect.Proxy;
022 import java.util.concurrent.BlockingQueue;
023
024 import org.apache.camel.Exchange;
025 import org.apache.camel.Processor;
026 import org.apache.commons.logging.Log;
027 import org.apache.commons.logging.LogFactory;
028
029 /**
030 * A thread that takes re-ordered {@link Exchange}s from a blocking queue and
031 * send them to the linked processor.
032 *
033 * @author Martin Krasser
034 *
035 * @version $Revision: 677549 $
036 */
037 public class SequenceSender extends Thread {
038
039 private static final transient Log LOG = LogFactory.getLog(SequenceSender.class);
040 private static final Exchange STOP = createStopSignal();
041
042 private BlockingQueue<Exchange> queue;
043 private Processor processor;
044
045 /**
046 * Creates a new {@link SequenceSender} thread.
047 *
048 * @param processor the processor to send re-ordered {@link Exchange}s.
049 */
050 public SequenceSender(Processor processor) {
051 this.processor = processor;
052 }
053
054 /**
055 * Sets the {@link BlockingQueue} to take messages from.
056 *
057 * @param queue the {@link BlockingQueue} to take messages from.
058 */
059 public void setQueue(BlockingQueue<Exchange> queue) {
060 this.queue = queue;
061 }
062
063 public void run() {
064 while (true) {
065 try {
066 Exchange exchange = queue.take();
067 if (exchange == STOP) {
068 LOG.info("Exit processing loop after cancellation");
069 return;
070 }
071 processor.process(exchange);
072 } catch (InterruptedException e) {
073 LOG.info("Exit processing loop after interrupt");
074 return;
075 } catch (Exception e) {
076 LOG.warn("Exception during exchange processing: " + e.getMessage());
077 }
078 }
079 }
080
081 /**
082 * Cancels this thread.
083 */
084 public void cancel() throws InterruptedException {
085 queue.put(STOP);
086 }
087
088 private static Exchange createStopSignal() {
089 return (Exchange)Proxy.newProxyInstance(SequenceSender.class.getClassLoader(),
090 new Class[] {Exchange.class}, createStopHandler());
091 }
092
093 private static InvocationHandler createStopHandler() {
094 return new InvocationHandler() {
095 public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
096 throw new RuntimeException("Illegal method invocation on stop signal");
097 }
098 };
099 }
100
101 }