Camel: How to stream a from SQL component using StreamList

Question

I am trying to use Camels SQL component to stream from a database using outputType=StreamList. I get the ResultIterator from a Java class with ConsumerTemplate :

public Flux<String> CreateFlux () {
ConsumerTemplate consumer = camelContext.createConsumerTemplate();

    ResultSetIterator resultSetIterator = consumer.receiveBody(
            "sql:SELECT DATA FROM TRANSAKSJON WHERE REQ_ID='" + recId + "'?outputType=StreamList", ResultSetIterator.class);
    ...
    while (result.hasNext()) {
        Map<String, String> map = (Map<String, String>) result.next();
        String data = map.get("DATA");

    }
}

I get the following error when trying to iterate the ResultsetIterator:

org.h2.jdbc.JdbcSQLException: The object is already closed [90007-197]

Upon inspection I see that the connection is closed. connection = {HikariProxyConnection@16287} "HikariProxyConnection@1048081993 wrapping com.zaxxer.hikari.pool.ProxyConnection.ClosedConnection"

How can I use the camel SQL component to stream? I have to use it from a bean that is not inside a camel route. I see that the streaming works only if I am using the SQL component inside a camel route.

Camel version is: 2.24.1

Update1: After looking at the source code it is intended. onDone closes the conncetion. I am trying to set UnitOfWork on my defaultExchange to keep the connection open by marking the exchange as not done.

Update2: managed to get it to work by setting UnitOf Work:

        ProducerTemplate pTmp = camelContext.createProducerTemplate();

        DefaultExchange defaultExchange = new DefaultExchange(camelContext);
        UnitOfWork unitOfWork = new DefaultUnitOfWork(defaultExchange);
        defaultExchange.setUnitOfWork(unitOfWork);
        pTmp.send("direct:DbStream", defaultExchange);

The route DbStream executes the SQL select described above

Claus Ibsen Claus Ibsen · Accepted Answer · 2020-02-14T10:46:17

Do not use receiveBody, but just receive to get the Exchange back. Then you can get the iterator from its message body, and the after use, you can done the exchange (see the javadoc)

Camel: How to stream a from SQL component using StreamList

1 Answers