001 /** 002 * 003 * Licensed to the Apache Software Foundation (ASF) under one or more 004 * contributor license agreements. See the NOTICE file distributed with 005 * this work for additional information regarding copyright ownership. 006 * The ASF licenses this file to You under the Apache License, Version 2.0 007 * (the "License"); you may not use this file except in compliance with 008 * the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018 package org.apache.camel.spring; 019 020 import org.aopalliance.intercept.MethodInvocation; 021 import org.apache.camel.CamelContext; 022 import org.apache.camel.Consumer; 023 import org.apache.camel.Endpoint; 024 import org.apache.camel.EndpointInject; 025 import org.apache.camel.Exchange; 026 import org.apache.camel.MessageDriven; 027 import org.apache.camel.Processor; 028 import org.apache.camel.Producer; 029 import org.apache.camel.RuntimeCamelException; 030 import org.apache.camel.CamelTemplate; 031 import org.apache.camel.spring.util.BeanInfo; 032 import org.apache.camel.spring.util.DefaultMethodInvocationStrategy; 033 import org.apache.camel.spring.util.MethodInvocationStrategy; 034 import org.apache.camel.spring.util.ReflectionUtils; 035 import org.apache.camel.util.ObjectHelper; 036 import static org.apache.camel.util.ObjectHelper.isNotNullOrBlank; 037 import org.apache.commons.logging.Log; 038 import org.apache.commons.logging.LogFactory; 039 import org.springframework.beans.BeansException; 040 import org.springframework.beans.factory.NoSuchBeanDefinitionException; 041 import org.springframework.beans.factory.config.BeanPostProcessor; 042 import org.springframework.context.ApplicationContext; 043 import org.springframework.context.ApplicationContextAware; 044 045 import java.lang.reflect.Field; 046 import java.lang.reflect.Method; 047 import java.util.ArrayList; 048 import java.util.List; 049 050 /** 051 * A post processor to perform injection of {@link Endpoint} and {@link Producer} instances together with binding 052 * methods annotated with {@link @MessageDriven} to a Camel consumer. 053 * 054 * @version $Revision: 1.1 $ 055 */ 056 public class CamelBeanPostProcessor implements BeanPostProcessor, ApplicationContextAware { 057 private static final transient Log log = LogFactory.getLog(CamelBeanPostProcessor.class); 058 059 private CamelContext camelContext; 060 private ApplicationContext applicationContext; 061 private MethodInvocationStrategy invocationStrategy = new DefaultMethodInvocationStrategy(); 062 //private List<Consumer> consumers = new ArrayList<Consumer>(); 063 064 public CamelBeanPostProcessor() { 065 } 066 067 public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { 068 injectFields(bean); 069 injectMethods(bean); 070 return bean; 071 } 072 073 public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { 074 return bean; 075 } 076 077 // Properties 078 //------------------------------------------------------------------------- 079 080 public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { 081 this.applicationContext = applicationContext; 082 } 083 084 public MethodInvocationStrategy getInvocationStrategy() { 085 return invocationStrategy; 086 } 087 088 public void setInvocationStrategy(MethodInvocationStrategy invocationStrategy) { 089 this.invocationStrategy = invocationStrategy; 090 } 091 092 public CamelContext getCamelContext() { 093 return camelContext; 094 } 095 096 public void setCamelContext(CamelContext camelContext) { 097 this.camelContext = camelContext; 098 } 099 100 // Implementation methods 101 //------------------------------------------------------------------------- 102 103 /** 104 * A strategy method to allow implementations to perform some custom JBI based injection of the POJO 105 * 106 * @param bean the bean to be injected 107 */ 108 protected void injectFields(final Object bean) { 109 ReflectionUtils.doWithFields(bean.getClass(), new ReflectionUtils.FieldCallback() { 110 public void doWith(Field field) throws IllegalArgumentException, IllegalAccessException { 111 EndpointInject annotation = field.getAnnotation(EndpointInject.class); 112 if (annotation != null) { 113 ReflectionUtils.setField(field, bean, getEndpointInjectionValue(annotation, field.getType())); 114 } 115 } 116 }); 117 } 118 119 protected void injectMethods(final Object bean) { 120 ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() { 121 @SuppressWarnings("unchecked") 122 public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { 123 setterInjection(method, bean); 124 consumerInjection(method, bean); 125 } 126 }); 127 } 128 129 protected void setterInjection(Method method, Object bean) { 130 EndpointInject annoation = method.getAnnotation(EndpointInject.class); 131 if (annoation != null) { 132 Class<?>[] parameterTypes = method.getParameterTypes(); 133 if (parameterTypes != null) { 134 if (parameterTypes.length != 1) { 135 log.warn("Ignoring badly annotated method for injection due to incorrect number of parameters: " + method); 136 } 137 else { 138 Object value = getEndpointInjectionValue(annoation, parameterTypes[0]); 139 ObjectHelper.invokeMethod(method, bean, value); 140 } 141 } 142 } 143 } 144 145 protected void consumerInjection(final Object bean) { 146 ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() { 147 @SuppressWarnings("unchecked") 148 public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { 149 /* 150 151 TODO support callbacks? 152 153 if (method.getAnnotation(Callback.class) != null) { 154 try { 155 Expression e = ExpressionFactory.createExpression( 156 method.getAnnotation(Callback.class).condition()); 157 JexlContext jc = JexlHelper.createContext(); 158 jc.getVars().put("this", obj); 159 Object r = e.evaluate(jc); 160 if (!(r instanceof Boolean)) { 161 throw new RuntimeException("Expression did not returned a boolean value but: " + r); 162 } 163 Boolean oldVal = req.getCallbacks().get(method); 164 Boolean newVal = (Boolean) r; 165 if ((oldVal == null || !oldVal) && newVal) { 166 req.getCallbacks().put(method, newVal); 167 method.invoke(obj, new Object[0]); 168 // TODO: handle return value and sent it as the answer 169 } 170 } catch (Exception e) { 171 throw new RuntimeException("Unable to invoke callback", e); 172 } 173 } 174 */ 175 } 176 }); 177 } 178 179 protected void consumerInjection(Method method, Object bean) { 180 MessageDriven annotation = method.getAnnotation(MessageDriven.class); 181 if (annotation != null) { 182 log.info("Creating a consumer for: " + annotation); 183 184 // lets bind this method to a listener 185 Endpoint endpoint = getEndpointInjection(annotation.uri(), annotation.name()); 186 if (endpoint != null) { 187 try { 188 Processor processor = createConsumerProcessor(bean, method, endpoint); 189 log.info("Created processor: " + processor); 190 Consumer consumer = endpoint.createConsumer(processor); 191 consumer.start(); 192 addConsumer(consumer); 193 } 194 catch (Exception e) { 195 log.warn(e); 196 throw new RuntimeCamelException(e); 197 } 198 } 199 } 200 } 201 202 /** 203 * Create a processor which invokes the given method when an incoming message exchange is received 204 */ 205 protected Processor createConsumerProcessor(final Object pojo, final Method method, final Endpoint endpoint) { 206 final BeanInfo beanInfo = new BeanInfo(pojo.getClass(), invocationStrategy); 207 208 return new Processor() { 209 @Override 210 public String toString() { 211 return "Processor on " + endpoint; 212 } 213 214 public void process(Exchange exchange) throws Exception { 215 if (log.isDebugEnabled()) { 216 log.debug(">>>> invoking method for: " + exchange); 217 } 218 MethodInvocation invocation = beanInfo.createInvocation(method, pojo, exchange); 219 if (invocation == null) { 220 throw new IllegalStateException("No method invocation could be created"); 221 } 222 try { 223 invocation.proceed(); 224 } 225 catch (Exception e) { 226 throw e; 227 } 228 catch (Throwable throwable) { 229 throw new Exception(throwable); 230 } 231 } 232 }; 233 } 234 235 protected void addConsumer(Consumer consumer) { 236 log.debug("Adding consumer: " + consumer); 237 //consumers.add(consumer); 238 } 239 240 /** 241 * Creates the value for the injection point for the given annotation 242 */ 243 protected Object getEndpointInjectionValue(EndpointInject annotation, Class<?> type) { 244 Endpoint endpoint = getEndpointInjection(annotation.uri(), annotation.name()); 245 if (endpoint != null) { 246 if (type.isInstance(endpoint)) { 247 return endpoint; 248 } 249 else if (type.isAssignableFrom(Producer.class)) { 250 try { 251 return endpoint.createProducer(); 252 } 253 catch (Exception e) { 254 throw new RuntimeCamelException(e); 255 } 256 } 257 else if (type.isAssignableFrom(CamelTemplate.class)) { 258 return new CamelTemplate(getCamelContext(), endpoint); 259 } 260 } 261 return null; 262 } 263 264 protected Endpoint getEndpointInjection(String uri, String name) { 265 Endpoint endpoint = null; 266 if (isNotNullOrBlank(uri)) { 267 endpoint = camelContext.getEndpoint(uri); 268 } 269 else { 270 if (isNotNullOrBlank(name)) { 271 endpoint = (Endpoint) applicationContext.getBean(name); 272 if (endpoint == null) { 273 throw new NoSuchBeanDefinitionException(name); 274 } 275 } 276 else { 277 log.warn("No uri or name specified on @EndpointInject annotation!"); 278 } 279 } 280 return endpoint; 281 } 282 }