Removing unwanted messages (Simple)

Not all messages are interesting and worth processing. Some applications generate too much noise, and only messages matching criteria should continue down a pipeline. In these kind of situations Message Filter pattern is used to eliminate undesirable messages, as shown in the following figure:

Getting ready

The complete source code for this tutorial is located under the following project: camel-message-routing-examples/removing-unwanted-messages.

How to do it...

Start with a filter definition, followed by a predicate expression and the processors to which this filter applies:

from("direct:start")
    .filter(header("userStatus").isEqualTo("valid"))
        .to("mock:valid")
    .end()
    .to("mock:all");

In the Java DSL, optionally we can mark the end of the filter, specifying up to where the filter applies. If not specified it applies for the rest of the route.

How it works...

This pattern is a simpler version of the Context-Based Router pattern demonstrated in the previous tutorial. It is conceptually similar to a Java if statement. If the incoming message satisfies the condition specified in the filter, the message is passed to the child processors. If the message fails to satisfy the condition, it is discarded and not passed to the child processor. In our example, we let only messages with valid userStatus header reach the mock:valid endpoint. None of the other messages will be passed to the mock:valid endpoint. Note that when a message is filtered out, it is not totally stopped from routing, it is not passed to child processors but it continues routing after the filter element. There is another command called stop that is used to completely stop a message from being processed further.

There's more...

Filtering messages is a very common operation in pipelines. Next, we will have a look at a special type of filter, the Idempotent consumer, and how to use Java beans for filtering messages, and as part of routes in general.

Idempotent consumer

One special kind of filter is the Idempotent consumer used to filter duplicate messages. The term idempotent describes a function that produces the same result if it is applied to itself. In computing, an idempotent operation has no additional effect when called more than once with the same input parameters. In messaging applications this means that receiving the same message once or multiple times has the same effect. Camel implements this pattern using the IdempotentConsumer class which uses an Expression to calculate a unique message ID for each message:

import static org.apache.camel.processor.idempotent.MemoryIdempotentRepository.memoryIdempotentRepository;
from("direct:start")
    .idempotentConsumer(
header("messageId"), memoryIdempotentRepository())
    .to("mock:result");

This ID is then looked up in IdempotentRepository to see if it has been processed before; if it has the message is discarded; if it hasn't then the message is processed and the ID is added to the repository to prevent processing other messages with the same ID.

Bean binding

When expressions are not flexible enough or we want to reuse existing Java code as part of EIPs, Camel bean language allows us to call Java methods directly from routes. Here is a filter that uses the method expression to call a bean method:

filter().method("myFilterBean", "isValidRequest")

Camel will look up in the Registry for a bean with the ID myFilterBean and call its isValidRequest method. In this case, because it is a predicate expression used in filter, the method has to return a Boolean or a value convertible to a Boolean. Instead of a reference to a bean, we can specify a class or omit the method name. Camel will instantiate an object from the class and if the method name is missing, it will attempt to choose the best matching method to call using its sophisticated bean binding algorithm. When a message arrives, it will try to map the Exchange content to method parameters and convert the result to an appropriate type. Some Camel specific types, such as Exchange, Message, Registry, CamelContext, and so on are always provided, if present, as method parameters. Camel will also try to bind the message body as the first parameter of the method by doing any type conversion optionally, unless it is one of the Camel specific types mentioned previously. We can influence the way parameters' values are created from the Exchange by using binding annotations in our beam:

public boolean isValidRequest(@Header("userId") String userId, @Body String body, Exchange exchange)

By annotating the method parameters, we tell Camel to bind the userId header to the first parameter, the message body to the second, and the Camel specific type, Exchange, to the last parameters. If we want to have cleaner POJOs, without any Camel dependencies, we could bind parameters using method options instead of introducing annotations in our beans. With this approach we don't need to touch the POJOs, instead we specify the parameter mappings in the route:

filter().method("myFilterBean", "isValidRequest(@Header("userId"), ${body}, null)")

Notice, with this approach we can additionally use literals (Booleans, numbers, string, null) or Simple language enclosed in curly brackets to pass values to the method.

Method construct is an expression, which can be used as part of EIPs (such as Message Filter, Context-Based Router, and so on), but Camel also has a bean component, which can be used as a processor in the routes using the same bean binding rules:

 from(...)
    .bean(OrderService.class, "doSomething(${body}, ${header.high})")

You can also use to instead of bean:

 from(...)  
    .to("bean:orderService?method=doSomething(${body.asXml}, ${header.high})")

In the preceding code snippet, the beans are not used as expressions in EIPs, they are processors in a route which actually can modify the Exchange or take some actions on each message. Bean binding is a very powerful Camel feature allowing us to execute Java code from anywhere in the routes. We can find out more about this feature at http://camel.apache.org/bean-binding.html.