[CAMEL-20889] camel-core: Stream is not reset when Message.getBody(class) is invoked ans stream caching is enabled - ASF JIRA

Attach files

Attach Screenshot

Voters

Watch issue

Watchers

Create sub-task

Link

Clone

Update Comment Author

Replace String in Comment

Update Comment Visibility

Delete Comments

XML

Word

Printable

JSON

Details

Type: Bug
Status: Resolved
Priority: Major
Resolution: Fixed
Affects Version/s: None
Fix Version/s: 4.7.0
Component/s: came-core
Labels:
None

Estimated Complexity:
Unknown

Description

Having the following Camel route on Camel 2.x

from("timer:mytimer2?repeatCount=1&delay=1000")
    .routeId("generate-route-marshall")
    .streamCaching()
    .transform(constant("\{\"maskme\":\"json payload\"}"))
    .marshal().json(JsonLibrary.Jackson)
    .process(exchange ->
    {
         LOGGER.info("body class name -> " + exchange.getIn().getBody().getClass().getName());
         LOGGER.info("consumed once  using .getBody(String.class) -> " + exchange.getIn().getBody(String.class));
        LOGGER.info("consumed twice using .getBody(String.class) -> " + exchange.getIn().getBody(String.class));
    })
    .to("jms:queue:INCOMING");

Shows that the consumption of the message payload in the processor via

exchange.getIn().getBody(String.class));\{code}
 works as expected, as the body is shown twice on the logs, as follows.
{code:java}
body class name -> org.apache.camel.converter.stream.InputStreamCache
consumed once  using .getBody(String.class) -> "\{\"maskme\":\"json payload\"}"
consumed twice using .getBody(String.class) -> "\{\"maskme\":\"json payload\"}"

But if you run this same route from Camel 3.x to the latest 4.x one, you will get the following output:

body class name -> org.apache.camel.converter.stream.InputStreamCache
consumed once  using .getBody(String.class) -> "\{\"maskme\":\"json payload\"}"
consumed once  using .getBody(String.class) ->

The problem is that on Camel 2.x, when
{{ exchange.getIn().getBody(String.class))}}
is called, the org.apache.camel.converter.IOConverterOptimised is triggered to convert the body from org.apache.camel.converter.stream.InputStreamCache to java.lang.String, which at the beginning of the convertion, it reset the stream:

IOConverterOptimised.java

public final class IOConverterOptimised {
    private IOConverterOptimised() {
    }

    public static Object convertTo(final Class<?> type, final Exchange exchange, final Object value) throws Exception {
        Class fromType = value.getClass();
        if (value instanceof StreamCache) {
            ((StreamCache)value).reset();      // <------ HERE
        }