-
Notifications
You must be signed in to change notification settings - Fork 4.9k
/
Enricher.java
349 lines (295 loc) · 13 KB
/
Enricher.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.camel.processor;
import java.util.Map;
import org.apache.camel.AggregationStrategy;
import org.apache.camel.AsyncCallback;
import org.apache.camel.CamelContext;
import org.apache.camel.CamelContextAware;
import org.apache.camel.CamelExchangeException;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
import org.apache.camel.ExchangePropertyKey;
import org.apache.camel.Expression;
import org.apache.camel.spi.EndpointUtilizationStatistics;
import org.apache.camel.spi.HeadersMapFactory;
import org.apache.camel.spi.IdAware;
import org.apache.camel.spi.ProcessorExchangeFactory;
import org.apache.camel.spi.RouteIdAware;
import org.apache.camel.support.AsyncProcessorSupport;
import org.apache.camel.support.DefaultExchange;
import org.apache.camel.support.ExchangeHelper;
import org.apache.camel.support.MessageHelper;
import org.apache.camel.support.service.ServiceHelper;
import static org.apache.camel.support.ExchangeHelper.copyResultsPreservePattern;
/**
* A content enricher that enriches input data by first obtaining additional data from a <i>resource</i> represented by
* an endpoint <code>producer</code> and second by aggregating input data and additional data. Aggregation of input data
* and additional data is delegated to an {@link AggregationStrategy} object.
* <p/>
* Uses a {@link org.apache.camel.Producer} to obtain the additional data as opposed to {@link PollEnricher} that uses a
* {@link org.apache.camel.PollingConsumer}.
*
* @see PollEnricher
*/
public class Enricher extends AsyncProcessorSupport implements IdAware, RouteIdAware, CamelContextAware {
private CamelContext camelContext;
private String id;
private String routeId;
private final Expression expression;
private final String uri;
private String variableSend;
private String variableReceive;
private AggregationStrategy aggregationStrategy;
private boolean aggregateOnException;
private boolean shareUnitOfWork;
private int cacheSize;
private boolean ignoreInvalidEndpoint;
private boolean allowOptimisedComponents = true;
private boolean autoStartupComponents = true;
private HeadersMapFactory headersMapFactory;
private ProcessorExchangeFactory processorExchangeFactory;
private SendDynamicProcessor sendDynamicProcessor;
public Enricher(Expression expression, String uri) {
this.expression = expression;
this.uri = uri;
}
@Override
public CamelContext getCamelContext() {
return camelContext;
}
@Override
public void setCamelContext(CamelContext camelContext) {
this.camelContext = camelContext;
}
@Override
public String getId() {
return id;
}
@Override
public void setId(String id) {
this.id = id;
}
@Override
public String getRouteId() {
return routeId;
}
@Override
public void setRouteId(String routeId) {
this.routeId = routeId;
}
public Expression getExpression() {
return expression;
}
public EndpointUtilizationStatistics getEndpointUtilizationStatistics() {
return sendDynamicProcessor.getEndpointUtilizationStatistics();
}
public String getVariableSend() {
return variableSend;
}
public void setVariableSend(String variableSend) {
this.variableSend = variableSend;
}
public String getVariableReceive() {
return variableReceive;
}
public void setVariableReceive(String variableReceive) {
this.variableReceive = variableReceive;
}
public void setAggregationStrategy(AggregationStrategy aggregationStrategy) {
this.aggregationStrategy = aggregationStrategy;
}
public AggregationStrategy getAggregationStrategy() {
return aggregationStrategy;
}
public boolean isAggregateOnException() {
return aggregateOnException;
}
public void setAggregateOnException(boolean aggregateOnException) {
this.aggregateOnException = aggregateOnException;
}
public boolean isShareUnitOfWork() {
return shareUnitOfWork;
}
public void setShareUnitOfWork(boolean shareUnitOfWork) {
this.shareUnitOfWork = shareUnitOfWork;
}
public int getCacheSize() {
return cacheSize;
}
public void setCacheSize(int cacheSize) {
this.cacheSize = cacheSize;
}
public boolean isIgnoreInvalidEndpoint() {
return ignoreInvalidEndpoint;
}
public void setIgnoreInvalidEndpoint(boolean ignoreInvalidEndpoint) {
this.ignoreInvalidEndpoint = ignoreInvalidEndpoint;
}
public boolean isAllowOptimisedComponents() {
return allowOptimisedComponents;
}
public void setAllowOptimisedComponents(boolean allowOptimisedComponents) {
this.allowOptimisedComponents = allowOptimisedComponents;
}
public boolean isAutoStartupComponents() {
return autoStartupComponents;
}
public void setAutoStartupComponents(boolean autoStartupComponents) {
this.autoStartupComponents = autoStartupComponents;
}
@Override
public boolean process(final Exchange exchange, final AsyncCallback callback) {
final Exchange resourceExchange = createResourceExchange(exchange, ExchangePattern.InOut);
// if we should store the received message body in a variable,
// then we need to preserve the original message body
Object body = null;
Map<String, Object> headers = null;
if (variableReceive != null) {
try {
body = exchange.getMessage().getBody();
// do a defensive copy of the headers
headers = headersMapFactory.newMap(exchange.getMessage().getHeaders());
} catch (Exception throwable) {
exchange.setException(throwable);
callback.done(true);
return true;
}
}
final Object originalBody = body;
final Map<String, Object> originalHeaders = headers;
return sendDynamicProcessor.process(resourceExchange, new AsyncCallback() {
@Override
public void done(boolean doneSync) {
if (!isAggregateOnException() && resourceExchange.isFailed()) {
// copy resource exchange onto original exchange (preserving pattern)
copyResultsWithoutCorrelationId(exchange, resourceExchange);
} else {
prepareResult(exchange);
try {
// prepare the exchanges for aggregation
ExchangeHelper.prepareAggregation(exchange, resourceExchange);
MessageHelper.resetStreamCache(exchange.getIn());
Exchange aggregatedExchange = aggregationStrategy.aggregate(exchange, resourceExchange);
if (aggregatedExchange != null) {
if (ExchangeHelper.shouldSetVariableResult(aggregatedExchange, variableReceive)) {
// result should be stored in variable instead of message body
ExchangeHelper.setVariableFromMessageBodyAndHeaders(aggregatedExchange, variableReceive,
aggregatedExchange.getMessage());
aggregatedExchange.getMessage().setBody(originalBody);
aggregatedExchange.getMessage().setHeaders(originalHeaders);
}
// copy aggregation result onto original exchange (preserving pattern)
copyResultsWithoutCorrelationId(exchange, aggregatedExchange);
// handover any synchronization (if unit of work is not shared)
if (resourceExchange != null && !isShareUnitOfWork()) {
resourceExchange.getExchangeExtension().handoverCompletions(exchange);
}
}
} catch (Exception e) {
// if the aggregationStrategy threw an exception, set it on the original exchange
exchange.setException(new CamelExchangeException("Error occurred during aggregation", exchange, e));
}
}
// and release resource exchange back in pool
processorExchangeFactory.release(resourceExchange);
callback.done(doneSync);
}
});
}
/**
* Creates a new {@link DefaultExchange} instance from the given <code>exchange</code>. The resulting exchange's
* pattern is defined by <code>pattern</code>.
*
* @param source exchange to copy from.
* @param pattern exchange pattern to set.
* @return created exchange.
*/
protected Exchange createResourceExchange(Exchange source, ExchangePattern pattern) {
// copy exchange, and do not share the unit of work
Exchange target = processorExchangeFactory.createCorrelatedCopy(source, false);
target.setPattern(pattern);
// if we share unit of work, we need to prepare the resource exchange
if (isShareUnitOfWork()) {
target.setProperty(ExchangePropertyKey.PARENT_UNIT_OF_WORK, source.getUnitOfWork());
// and then share the unit of work
target.getExchangeExtension().setUnitOfWork(source.getUnitOfWork());
}
return target;
}
private static void prepareResult(Exchange exchange) {
if (exchange.getPattern().isOutCapable()) {
exchange.getOut().copyFrom(exchange.getIn());
}
}
private static AggregationStrategy defaultAggregationStrategy() {
return new CopyAggregationStrategy();
}
@Override
public String toString() {
return id;
}
@Override
protected void doBuild() throws Exception {
// use send dynamic to send to endpoint
this.sendDynamicProcessor = new SendDynamicProcessor(uri, expression);
this.sendDynamicProcessor.setCamelContext(camelContext);
this.sendDynamicProcessor.setCacheSize(cacheSize);
this.sendDynamicProcessor.setIgnoreInvalidEndpoint(ignoreInvalidEndpoint);
this.sendDynamicProcessor.setAllowOptimisedComponents(allowOptimisedComponents);
this.sendDynamicProcessor.setAutoStartupComponents(autoStartupComponents);
this.sendDynamicProcessor.setVariableSend(variableSend);
// create a per processor exchange factory
this.processorExchangeFactory = getCamelContext().getCamelContextExtension()
.getProcessorExchangeFactory().newProcessorExchangeFactory(this);
this.processorExchangeFactory.setRouteId(getRouteId());
this.processorExchangeFactory.setId(getId());
if (aggregationStrategy == null) {
aggregationStrategy = defaultAggregationStrategy();
}
CamelContextAware.trySetCamelContext(aggregationStrategy, camelContext);
ServiceHelper.buildService(processorExchangeFactory, sendDynamicProcessor);
}
@Override
protected void doInit() throws Exception {
headersMapFactory = camelContext.getCamelContextExtension().getHeadersMapFactory();
}
@Override
protected void doStart() throws Exception {
ServiceHelper.startService(processorExchangeFactory, aggregationStrategy, sendDynamicProcessor);
}
@Override
protected void doStop() throws Exception {
ServiceHelper.stopService(aggregationStrategy, processorExchangeFactory, sendDynamicProcessor);
}
private static void copyResultsWithoutCorrelationId(Exchange target, Exchange source) {
Object correlationId = target.removeProperty(ExchangePropertyKey.CORRELATION_ID);
copyResultsPreservePattern(target, source);
if (correlationId != null) {
target.setProperty(ExchangePropertyKey.CORRELATION_ID, correlationId);
}
}
private static class CopyAggregationStrategy implements AggregationStrategy {
@Override
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
if (newExchange != null) {
copyResultsWithoutCorrelationId(oldExchange, newExchange);
}
return oldExchange;
}
}
}