Hi Team,
I have a requirement where I need to read the messages from a queue filter
out the duplicates and send to out queue. For filtering out the duplicates
based on MessageId I'm using Ehcache based Idempotent repository . It is
working fine as long as the camel application is up. When camel goes down
and later comes up earlier saved MessageIds are not referred to filter out
because of which I do have duplicate messages in out queue.
I want the cache to hold the MessageIds even when camel is down and refer
back the same list once camel is up. Please find my code below and suggest
me what i'm missing here.
*pom.xml:*
<dependency>
<groupId>org.apache.camel.springboot</groupId>
<artifactId>camel-ehcache-starter</artifactId>
</dependency>
Config.java
@Configuration
public class Config {
@Bean
public static EhcacheIdempotentRepository ehcacheIdempotentRepository()
throws MalformedURLException {
//CacheManager manager = CacheManagerBuilder.newCacheManager(new
XmlConfiguration("ehcache.xml"));
File myfile = new File("src/main/resources/ehcache.xml");
System.out.println("file:" + myfile.toString() );
URL myUrl = myfile.toURI().toURL();
//URL myUrl = getClass().getResource("src/main/resources/my-config.xml");
XmlConfiguration xmlConfig = new XmlConfiguration(myUrl);
CacheManager myCacheManager =
CacheManagerBuilder.newCacheManager(xmlConfig);
myCacheManager.init();
EhcacheIdempotentRepository ehcacheIdempotentRepository = new
EhcacheIdempotentRepository(myCacheManager, "idempotent-cache");
return ehcacheIdempotentRepository;
}
}
ehcache.xml:
<config xmlns:xsi='http://www.w3.org/2001/XMLSchema-instance'
xmlns='http://www.ehcache.org/v3'
xsi:schemaLocation="http://www.ehcache.org/v3
http://www.ehcache.org/schema/ehcache-core.xsd";>
<persistence directory="D:/cache" />
<cache alias="idempotent-cache">
<key-type>java.lang.String</key-type>
<value-type>java.lang.Boolean</value-type>
<expiry>
<ttl unit="days">7</ttl>
</expiry>
<resources>
<heap unit="entries">2000</heap>
<offheap unit="MB">10</offheap>
<disk persistent="true" unit="MB">20</disk>
</resources>
</cache>
</config>
CamelRoute:
public class SampleAutowiredAmqRoute extends RouteBuilder {
@Autowired
@Qualifier("ehcacheIdempotentRepository")
EhcacheIdempotentRepository ehcacheIdempotentRepository;
@Override
public void configure() throws Exception {
from("activemq:T24?connectionFactory=sourceConnectionFactory&acknowledgementModeName=CLIENT_ACKNOWLEDGE&concurrentConsumers=10")
.idempotentConsumer(header("JMSCorrelationID"),ehcacheIdempotentRepository )
.to("activemq:Equens?connectionFactory=destConnectionFactory");
}
}
Thanks in Advance!!
Regards,
Pooja
