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