Camel Request/Reply Correlation

Question

I have a CXFRS endpoint, where a request such as "GET /files/x" should return file "x" from a particular directory and then delete it. The files are output from another process and then quickly consumed via this route. Thus, it would be acceptable for them to be polled and kept in memory temporarily, if need be.

Below is my web service.

@Component
@Path("/")
public class WebService {
 @GET
 @Path("files/{id}")
 public String getFile(@PathParam("id") String id) {
  return null;
 }
}

Below is the incomplete route.

<route>
 <from uri="cxfrs://bean:webService"/>
 <choice>
  <when>
   <simple>${in.headers.operationName} == 'getFile'</simple>
   <setHeader headerName="correlationId">
    <simple>mandatoryBodyAs(java.lang.String)</simple>
   </setHeader>
   ???
  </when>
 <choice>
</route>

I have investigated the content enricher pattern, but this will not help because the original exchange cannot be accessed by the consumer within the enricher. Thus, the file name cannot be dynamically determined from the input message. In other words, the example below would not read file "x", because the header is never seen by the file endpoint.

...setHeader(Exchange.FILE_NAME, "x").pollEnrich("file://dir")...

I also attempted to use the aggregator pattern between the web service route and a separate file route, with the aggregation strategy shown below.

@Component
public class Aggregator implements AggregationStrategy {
 public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
  if (oldExchange == null) {
   return newExchange;
  } else {
   oldExchange.getOut().setBody(newExchange.getIn().getBody());
   return oldExchange;
  }
 } 
}

This does not work because the aggregator pattern seems to be for "in only" exchanges; with "in out" exchanges, the reply is returned to the client as soon as the message reaches the aggregator, not after aggregation is complete. Subsequently, I moved onto writing the process below.

@Component
public class FileEnricher implements Processor {

 @Value("${folder}")
 private String folder;

 public void process(Exchange exchange) throws Exception {

  Endpoint endpoint = exchange.getContext().getEndpoint(String.format(
   "file://%s?fileName=%s",
   folder,
   exchange.getIn().getHeader("correlationId")
  ));
  PollingConsumer consumer = endpoint.createPollingConsumer();
  PollEnricher enricher = new PollEnricher(consumer);
  enricher.setTimeout(10000);

  consumer.start();
  enricher.process(exchange);
  enricher.shutdown();
  consumer.stop();
 }
}

This process does nothing more than allow me to dynamically configure a content enricher. I find it hard to believe there is not a better way to do it. In particular, I am concerned about threading and constantly adding/removing components/endpoints in the context.

Can anyone suggest a better solution to this problem?

Answer 1

Just set the message body with the java.io.File for the file to enrich, for example from a java bean. And then route the message to the bean:

public File whichFileToPick(@Header("id") String id) {
   return new File("somedir/" + id");
}

In the future the pollEnrich will be enhanced so it will be easier to do directly in a Camel route. Christian came up with a good idea of allowing to specify a Camel expression as the uri for the pollEnrich.