Hi,
I am trying to use Camel (2.11.1) Redis component to subscribe to a channel
without success.
I have looked at the test case
org.apache.camel.component.redis.RedisConsumerIntegrationTest
By the way, the above test case is disabled by the Junit @Ignore.
My code below is a clone of RedisConsumerIntegrationTest . It receives no
message when I publish to mychannel from a Redis command prompt.
(There is no error.)
public class RedisSubscriberRouter {
public static void main(String[] args) throws Exception {
JedisShardInfo hostInfo = new JedisShardInfo("localhost", "6379");
hostInfo.setPassword("xxx");
final JedisConnectionFactory CONNECTION_FACTORY = new
JedisConnectionFactory(
hostInfo);
final RedisMessageListenerContainer LISTENER_CONTAINER = new
RedisMessageListenerContainer();
CONNECTION_FACTORY.afterPropertiesSet();
LISTENER_CONTAINER.setConnectionFactory(CONNECTION_FACTORY);
LISTENER_CONTAINER.afterPropertiesSet();
RedisTemplate<String, String> redisTemplate = new RedisTemplate<String,
String>();
redisTemplate.setConnectionFactory(CONNECTION_FACTORY);
redisTemplate.afterPropertiesSet();
SimpleRegistry registry = new SimpleRegistry();
registry.put("redisTemplate", redisTemplate);
registry.put("listenerContainer", LISTENER_CONTAINER);
// create CamelContext
CamelContext context = new DefaultCamelContext(registry);
context.addRoutes(new RouteBuilder() {
@Override
public void configure() throws Exception {
from(
"spring-redis://localhost:6379?command=SUBSCRIBE&channels=mychannel&listenerContainer=#listenerContainer")
.process(new Processor() {
@Override
public void process(Exchange exchange) throws
Exception {
String res =
exchange.getIn().getBody().toString();
System.out.println("************ " + res);
exchange.getOut().setBody(res);
}
})
.to("log:foo");
}
});
context.start();
System.out.println("Press any key to shutdown.");
BufferedReader br = new BufferedReader(new
InputStreamReader(System.in));
br.readLine();
context.stop();
}
}
What is wrong my code above ?
Thanks in advance for any assistance !
Shing
