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.model;
018
019 import java.util.ArrayList;
020 import java.util.List;
021 import java.util.concurrent.ExecutorService;
022
023 import javax.xml.bind.annotation.XmlAccessType;
024 import javax.xml.bind.annotation.XmlAccessorType;
025 import javax.xml.bind.annotation.XmlAttribute;
026 import javax.xml.bind.annotation.XmlElement;
027 import javax.xml.bind.annotation.XmlElementRef;
028 import javax.xml.bind.annotation.XmlRootElement;
029 import javax.xml.bind.annotation.XmlTransient;
030
031 import org.apache.camel.Endpoint;
032 import org.apache.camel.ExchangePattern;
033 import org.apache.camel.Expression;
034 import org.apache.camel.Processor;
035 import org.apache.camel.Producer;
036 import org.apache.camel.processor.CamelInternalProcessor;
037 import org.apache.camel.processor.WireTapProcessor;
038 import org.apache.camel.spi.RouteContext;
039 import org.apache.camel.util.CamelContextHelper;
040
041 /**
042 * Represents an XML <wireTap/> element
043 */
044 @XmlRootElement(name = "wireTap")
045 @XmlAccessorType(XmlAccessType.FIELD)
046 public class WireTapDefinition<Type extends ProcessorDefinition<Type>> extends NoOutputDefinition<WireTapDefinition<Type>>
047 implements ExecutorServiceAwareDefinition<WireTapDefinition<Type>>, EndpointRequiredDefinition {
048 @XmlAttribute
049 protected String uri;
050 @XmlAttribute
051 protected String ref;
052 @XmlTransient
053 protected Endpoint endpoint;
054 @XmlTransient
055 private Processor newExchangeProcessor;
056 @XmlAttribute(name = "processorRef")
057 private String newExchangeProcessorRef;
058 @XmlElement(name = "body")
059 private ExpressionSubElementDefinition newExchangeExpression;
060 @XmlElementRef
061 private List<SetHeaderDefinition> headers = new ArrayList<SetHeaderDefinition>();
062 @XmlTransient
063 private ExecutorService executorService;
064 @XmlAttribute
065 private String executorServiceRef;
066 @XmlAttribute
067 private Boolean copy;
068 @XmlAttribute
069 private String onPrepareRef;
070 @XmlTransient
071 private Processor onPrepare;
072
073 public WireTapDefinition() {
074 }
075
076 public WireTapDefinition(String uri) {
077 setUri(uri);
078 }
079
080 public WireTapDefinition(Endpoint endpoint) {
081 setEndpoint(endpoint);
082 }
083
084 @Override
085 public String getEndpointUri() {
086 if (uri != null) {
087 return uri;
088 } else if (endpoint != null) {
089 return endpoint.getEndpointUri();
090 } else {
091 return null;
092 }
093 }
094
095 @Override
096 public Processor createProcessor(RouteContext routeContext) throws Exception {
097 // executor service is mandatory for wire tap
098 boolean shutdownThreadPool = ProcessorDefinitionHelper.willCreateNewThreadPool(routeContext, this, true);
099 ExecutorService threadPool = ProcessorDefinitionHelper.getConfiguredExecutorService(routeContext, "WireTap", this, true);
100
101 // create the producer to send to the wire tapped endpoint
102 Endpoint endpoint = resolveEndpoint(routeContext);
103 Producer producer = endpoint.createProducer();
104
105 // create error handler we need to use for processing the wire tapped
106 Processor target = wrapInErrorHandler(routeContext, producer);
107
108 // and wrap in unit of work
109 String routeId = routeContext.getRoute().idOrCreate(routeContext.getCamelContext().getNodeIdFactory());
110 CamelInternalProcessor internal = new CamelInternalProcessor(target);
111 internal.addAdvice(new CamelInternalProcessor.UnitOfWorkProcessorAdvice(routeId));
112
113 WireTapProcessor answer = new WireTapProcessor(endpoint, internal, getPattern(), threadPool, shutdownThreadPool);
114 answer.setCopy(isCopy());
115 if (newExchangeProcessorRef != null) {
116 newExchangeProcessor = routeContext.mandatoryLookup(newExchangeProcessorRef, Processor.class);
117 }
118 if (newExchangeProcessor != null) {
119 answer.addNewExchangeProcessor(newExchangeProcessor);
120 }
121 if (newExchangeExpression != null) {
122 answer.setNewExchangeExpression(newExchangeExpression.createExpression(routeContext));
123 }
124 if (headers != null && !headers.isEmpty()) {
125 for (SetHeaderDefinition header : headers) {
126 Processor processor = createProcessor(routeContext, header);
127 answer.addNewExchangeProcessor(processor);
128 }
129 }
130 if (onPrepareRef != null) {
131 onPrepare = CamelContextHelper.mandatoryLookup(routeContext.getCamelContext(), onPrepareRef, Processor.class);
132 }
133 if (onPrepare != null) {
134 answer.setOnPrepare(onPrepare);
135 }
136
137 return answer;
138 }
139
140 public ExchangePattern getPattern() {
141 return ExchangePattern.InOnly;
142 }
143
144 @Override
145 public String toString() {
146 return "WireTap[" + description() + "]";
147 }
148
149 protected String description() {
150 return FromDefinition.description(getUri(), getRef(), getEndpoint());
151 }
152
153 @Override
154 public String getShortName() {
155 return "wireTap";
156 }
157
158 @Override
159 public String getLabel() {
160 return "wireTap[" + description() + "]";
161 }
162
163 @Override
164 @SuppressWarnings("unchecked")
165 public Type end() {
166 // allow end() to return to previous type so you can continue in the DSL
167 return (Type) super.end();
168 }
169
170 @Override
171 public void addOutput(ProcessorDefinition<?> output) {
172 // add outputs on parent as this wiretap does not support outputs
173 getParent().addOutput(output);
174 }
175
176 public Endpoint resolveEndpoint(RouteContext context) {
177 if (endpoint == null) {
178 return context.resolveEndpoint(getUri(), getRef());
179 } else {
180 return endpoint;
181 }
182 }
183
184 // Fluent API
185 // -------------------------------------------------------------------------
186
187 /**
188 * Uses a custom thread pool
189 *
190 * @param executorService a custom {@link ExecutorService} to use as thread pool
191 * for sending tapped exchanges
192 * @return the builder
193 */
194 public WireTapDefinition<Type> executorService(ExecutorService executorService) {
195 setExecutorService(executorService);
196 return this;
197 }
198
199 /**
200 * Uses a custom thread pool
201 *
202 * @param executorServiceRef reference to lookup a custom {@link ExecutorService}
203 * to use as thread pool for sending tapped exchanges
204 * @return the builder
205 */
206 public WireTapDefinition<Type> executorServiceRef(String executorServiceRef) {
207 setExecutorServiceRef(executorServiceRef);
208 return this;
209 }
210
211 /**
212 * Uses a copy of the original exchange
213 *
214 * @return the builder
215 */
216 public WireTapDefinition<Type> copy() {
217 setCopy(true);
218 return this;
219 }
220
221 /**
222 * Uses a copy of the original exchange
223 *
224 * @param copy if it is true camel will copy the original exchange,
225 * if it is false camel will not copy the original exchange
226 * @return the builder
227 */
228 public WireTapDefinition<Type> copy(boolean copy) {
229 setCopy(copy);
230 return this;
231 }
232
233 /**
234 * @deprecated will be removed in Camel 3.0 Instead use {@link #newExchangeBody(org.apache.camel.Expression)}
235 */
236 @Deprecated
237 public WireTapDefinition<Type> newExchange(Expression expression) {
238 return newExchangeBody(expression);
239 }
240
241 /**
242 * Sends a <i>new</i> Exchange, instead of tapping an existing, using {@link ExchangePattern#InOnly}
243 *
244 * @param expression expression that creates the new body to send
245 * @return the builder
246 * @see #newExchangeHeader(String, org.apache.camel.Expression)
247 */
248 public WireTapDefinition<Type> newExchangeBody(Expression expression) {
249 setNewExchangeExpression(expression);
250 return this;
251 }
252
253 /**
254 * Sends a <i>new</i> Exchange, instead of tapping an existing, using {@link ExchangePattern#InOnly}
255 *
256 * @param ref reference to the {@link Processor} to lookup in the {@link org.apache.camel.spi.Registry} to
257 * be used for preparing the new exchange to send
258 * @return the builder
259 */
260 public WireTapDefinition<Type> newExchangeRef(String ref) {
261 setNewExchangeProcessorRef(ref);
262 return this;
263 }
264
265 /**
266 * Sends a <i>new</i> Exchange, instead of tapping an existing, using {@link ExchangePattern#InOnly}
267 *
268 * @param processor processor preparing the new exchange to send
269 * @return the builder
270 * @see #newExchangeHeader(String, org.apache.camel.Expression)
271 */
272 public WireTapDefinition<Type> newExchange(Processor processor) {
273 setNewExchangeProcessor(processor);
274 return this;
275 }
276
277 /**
278 * Sets a header on the <i>new</i> Exchange, instead of tapping an existing, using {@link ExchangePattern#InOnly}.
279 * <p/>
280 * Use this together with the {@link #newExchange(org.apache.camel.Expression)} or {@link #newExchange(org.apache.camel.Processor)}
281 * methods.
282 *
283 * @param headerName the header name
284 * @param expression the expression setting the header value
285 * @return the builder
286 */
287 public WireTapDefinition<Type> newExchangeHeader(String headerName, Expression expression) {
288 headers.add(new SetHeaderDefinition(headerName, expression));
289 return this;
290 }
291
292 /**
293 * Uses the {@link Processor} when preparing the {@link org.apache.camel.Exchange} to be send.
294 * This can be used to deep-clone messages that should be send, or any custom logic needed before
295 * the exchange is send.
296 *
297 * @param onPrepare the processor
298 * @return the builder
299 */
300 public WireTapDefinition<Type> onPrepare(Processor onPrepare) {
301 setOnPrepare(onPrepare);
302 return this;
303 }
304
305 /**
306 * Uses the {@link Processor} when preparing the {@link org.apache.camel.Exchange} to be send.
307 * This can be used to deep-clone messages that should be send, or any custom logic needed before
308 * the exchange is send.
309 *
310 * @param onPrepareRef reference to the processor to lookup in the {@link org.apache.camel.spi.Registry}
311 * @return the builder
312 */
313 public WireTapDefinition<Type> onPrepareRef(String onPrepareRef) {
314 setOnPrepareRef(onPrepareRef);
315 return this;
316 }
317
318 public String getUri() {
319 return uri;
320 }
321
322 public void setUri(String uri) {
323 this.uri = uri;
324 }
325
326 public String getRef() {
327 return ref;
328 }
329
330 public void setRef(String ref) {
331 this.ref = ref;
332 }
333
334 public Endpoint getEndpoint() {
335 return endpoint;
336 }
337
338 public void setEndpoint(Endpoint endpoint) {
339 this.endpoint = endpoint;
340 }
341
342 public Processor getNewExchangeProcessor() {
343 return newExchangeProcessor;
344 }
345
346 public void setNewExchangeProcessor(Processor processor) {
347 this.newExchangeProcessor = processor;
348 }
349
350 public String getNewExchangeProcessorRef() {
351 return newExchangeProcessorRef;
352 }
353
354 public void setNewExchangeProcessorRef(String ref) {
355 this.newExchangeProcessorRef = ref;
356 }
357
358 public ExpressionSubElementDefinition getNewExchangeExpression() {
359 return newExchangeExpression;
360 }
361
362 public void setNewExchangeExpression(ExpressionSubElementDefinition expression) {
363 this.newExchangeExpression = expression;
364 }
365
366 public void setNewExchangeExpression(Expression expression) {
367 this.newExchangeExpression = new ExpressionSubElementDefinition(expression);
368 }
369
370 public ExecutorService getExecutorService() {
371 return executorService;
372 }
373
374 public void setExecutorService(ExecutorService executorService) {
375 this.executorService = executorService;
376 }
377
378 public String getExecutorServiceRef() {
379 return executorServiceRef;
380 }
381
382 public void setExecutorServiceRef(String executorServiceRef) {
383 this.executorServiceRef = executorServiceRef;
384 }
385
386 public Boolean getCopy() {
387 return copy;
388 }
389
390 public void setCopy(Boolean copy) {
391 this.copy = copy;
392 }
393
394 public boolean isCopy() {
395 // should default to true if not configured
396 return copy != null ? copy : true;
397 }
398
399 public String getOnPrepareRef() {
400 return onPrepareRef;
401 }
402
403 public void setOnPrepareRef(String onPrepareRef) {
404 this.onPrepareRef = onPrepareRef;
405 }
406
407 public Processor getOnPrepare() {
408 return onPrepare;
409 }
410
411 public void setOnPrepare(Processor onPrepare) {
412 this.onPrepare = onPrepare;
413 }
414
415 public List<SetHeaderDefinition> getHeaders() {
416 return headers;
417 }
418
419 public void setHeaders(List<SetHeaderDefinition> headers) {
420 this.headers = headers;
421 }
422 }