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 */ 017package org.apache.camel.component.event; 018 019import org.apache.camel.Exchange; 020import org.apache.camel.Processor; 021import org.apache.camel.Producer; 022import org.apache.camel.impl.DefaultEndpoint; 023import org.apache.camel.impl.DefaultProducer; 024import org.apache.camel.processor.loadbalancer.LoadBalancer; 025import org.apache.camel.processor.loadbalancer.TopicLoadBalancer; 026import org.apache.camel.spi.UriEndpoint; 027import org.apache.camel.spi.UriPath; 028import org.apache.camel.util.ObjectHelper; 029import org.springframework.beans.BeansException; 030import org.springframework.context.ApplicationContext; 031import org.springframework.context.ApplicationContextAware; 032import org.springframework.context.ApplicationEvent; 033 034import static org.apache.camel.util.ObjectHelper.wrapRuntimeCamelException; 035 036/** 037 * An <a href="http://camel.apache.org/event.html">Event Endpoint</a> 038 * for working with Spring ApplicationEvents 039 * 040 * @version 041 */ 042@UriEndpoint(scheme = "spring-event", title = "Spring Event", syntax = "spring-event:name", consumerClass = EventConsumer.class, label = "spring,eventbus") 043public class EventEndpoint extends DefaultEndpoint implements ApplicationContextAware { 044 private LoadBalancer loadBalancer; 045 private ApplicationContext applicationContext; 046 047 @UriPath(description = "Name of endpoint") 048 private String name; 049 050 public EventEndpoint(String endpointUri, EventComponent component, String name) { 051 super(endpointUri, component); 052 this.applicationContext = component.getApplicationContext(); 053 this.name = name; 054 } 055 056 /** 057 * <b>Note:</b> It is preferred to create endpoints using the associated 058 * component. 059 * @deprecated 060 */ 061 @Deprecated 062 public EventEndpoint(String endpointUri) { 063 super(endpointUri); 064 } 065 066 public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { 067 this.applicationContext = applicationContext; 068 } 069 070 public ApplicationContext getApplicationContext() { 071 return applicationContext; 072 } 073 074 public String getName() { 075 return name; 076 } 077 078 public void setName(String name) { 079 this.name = name; 080 } 081 082 public boolean isSingleton() { 083 return true; 084 } 085 086 public Producer createProducer() throws Exception { 087 ObjectHelper.notNull(applicationContext, "applicationContext"); 088 return new DefaultProducer(this) { 089 public void process(Exchange exchange) throws Exception { 090 ApplicationEvent event = toApplicationEvent(exchange); 091 applicationContext.publishEvent(event); 092 } 093 }; 094 } 095 096 public EventConsumer createConsumer(Processor processor) throws Exception { 097 ObjectHelper.notNull(applicationContext, "applicationContext"); 098 EventConsumer answer = new EventConsumer(this, processor); 099 configureConsumer(answer); 100 return answer; 101 } 102 103 public void onApplicationEvent(ApplicationEvent event) { 104 Exchange exchange = createExchange(); 105 exchange.getIn().setBody(event); 106 try { 107 getLoadBalancer().process(exchange); 108 } catch (Exception e) { 109 throw wrapRuntimeCamelException(e); 110 } 111 } 112 113 public LoadBalancer getLoadBalancer() { 114 if (loadBalancer == null) { 115 loadBalancer = createLoadBalancer(); 116 } 117 return loadBalancer; 118 } 119 120 public void setLoadBalancer(LoadBalancer loadBalancer) { 121 this.loadBalancer = loadBalancer; 122 } 123 124 @Override 125 public EventComponent getComponent() { 126 return (EventComponent) super.getComponent(); 127 } 128 129 // Implementation methods 130 // ------------------------------------------------------------------------- 131 public synchronized void consumerStarted(EventConsumer consumer) { 132 getComponent().consumerStarted(this); 133 getLoadBalancer().addProcessor(consumer.getProcessor()); 134 } 135 136 public synchronized void consumerStopped(EventConsumer consumer) { 137 getComponent().consumerStopped(this); 138 getLoadBalancer().removeProcessor(consumer.getProcessor()); 139 } 140 141 protected LoadBalancer createLoadBalancer() { 142 return new TopicLoadBalancer(); 143 } 144 145 protected ApplicationEvent toApplicationEvent(Exchange exchange) { 146 ApplicationEvent event = exchange.getIn().getBody(ApplicationEvent.class); 147 if (event != null) { 148 return event; 149 } 150 return new CamelEvent(this, exchange); 151 } 152}