smpp transaction doesn't work
Hi, I am trying to implement obtaining messaging over camel but have a
problem with transaction.
The idea is to leave message in the SMSC in case processing fail and retry
obtaining later. I suppose that we need implement transaction what was done.
But in case some error in time processing transaction doesn’t roll back
data and we just miss message.
Here my rout:
from(consumerUri)
.transacted()
.routeId("SMPP_ID_CONSUMER")
.process(new Processor(){
@Override
public void process(final Exchange
exchange) throws Exception {
Message message =
exchange.getIn();
if (logger.isTraceEnabled()) {
logger.trace("Receive
message from SMSC");
logger.trace("headers:
{}", message.getHeaders());
logger.trace("body:
{}", message.getBody());
}
throw new Exception();
////here I am throwing Exception to
check behavior
}
})
.choice()
.when(header(SmppConstants.MESSAGE_TYPE).isEqualTo(SmppMessageType.DeliverSm.toString()))
// process incoming message
.process(new Processor() {
@Override
public void
process(Exchange exchange) throws Exception {
Message message
= exchange.getIn();
logger.info("Receiving message: {}", message.getBody());
InboundMessage
inMsg = new
InboundMessage(InboundMessageType.SMS);
inMsg.addPart(MessagePart.FROM,
message.getHeader(SmppConstants.SOURCE_ADDR));
inMsg.addPart(MessagePart.BODY, message.getBody());
inMsg.addPart(MessagePart.HEADER, message.getHeaders());
message.setBody(inMsg);
}
})
.to("activemq:queue:test.sms")
// process delivery message
.when(header(SmppConstants.MESSAGE_TYPE).isEqualTo(SmppMessageType.DeliveryReceipt.toString()))
.choice()
// process successful
delivery
.when(header(SmppConstants.FINAL_STATUS).isEqualTo(DELIVERED_STATUS))
.process(new
Processor() {
@Override
public
void process(Exchange exchange) throws Exception {
Message message = exchange.getIn();
logger.debug("Message \"{}\" was successfully delivered",
message.getBody());
}
})
// process delivery fail
.when(header(SmppConstants.FINAL_STATUS).isNotEqualTo(DELIVERED_STATUS))
.process(new
DeliveryFailProcessor())
.to("activemq:test.sms")
.otherwise()
.process(new Processor() {
@Override
public void
process(Exchange exchange) throws Exception {
logger.info("Unsuported SMSC command: " +
exchange.getIn().getHeader(SmppConstants.MESSAGE_TYPE));
}
});
}
};
}
Here my configuration:
@Bean(name="CamelContext")
public CamelContext createCamelContext() throws Exception {
ApplicationContext spring = new
ClassPathXmlApplicationContext("classpath:oncall-sms-context.xml");
CamelContext camelContext =
SpringCamelContext.springCamelContext(spring);
PlatformTransactionManager transactionManager =
(PlatformTransactionManager) spring.getBean("txManager");
ConnectionFactory connectionFactory = (ConnectionFactory)
spring.getBean("jmsConnectionFactory");
JmsComponent component =
JmsComponent.jmsComponentTransacted(connectionFactory);
component.setTransacted(true);
component.setTransactionManager(transactionManager);
camelContext.addComponent("jms", component);
camelContext.start();
return camelContext;
Here my XML:
<bean id="txManager"
class="org.springframework.jms.connection.JmsTransactionManager">
<property name="connectionFactory" ref="jmsConnectionFactory" />
</bean>
<bean id="jmsConnectionFactory"
class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://MessageSms" />
</bean>
Here what I am getting:
Message History
---------------------------------------------------------------------------------------------------------------------------------------
RouteId ProcessorId Processor
Elapsed (ms)
[SMPP_ID_CONSUMER ] [SMPP_ID_CONSUMER ]
[smpp://[email protected]:2775?CamelSmppDestAddrNpi=0&CamelSmppDestAddrTon=]
[ 46]
[SMPP_ID_CONSUMER ] [transacted1 ] [transacted[]
] [ 44]
[SMPP_ID_CONSUMER ] [process1 ]
[com.solarwinds.oncall.sms.routs.SmppConsumerRouteBuilder$1$1@7bc0ad26
] [ 4]
Exchange
---------------------------------------------------------------------------------------------------------------------------------------
Exchange[
Id ID-Dakar-56706-1394108356753-0-2
ExchangePattern InOnly
Headers {breadcrumbId=ID-Dakar-56706-1394108356753-0-1,
CamelRedelivered=false, CamelRedeliveryCounter=0, CamelSmppCommandId=5,
CamelSmppDestAddr=null, CamelSmppDestAddrNpi=1, CamelSmppDestAddrTon=1,
CamelSmppMessageType=DeliverSm, CamelSmppScheduleDeliveryTime=null,
CamelSmppSequenceNumber=72, CamelSmppServiceType=null,
CamelSmppSourceAddr=4477665544, CamelSmppSourceAddrNpi=1,
CamelSmppSourceAddrTon=1, CamelSmppValidityPeriod=null}
BodyType String
Body Hello from SMPPSim
]
Stacktrace
---------------------------------------------------------------------------------------------------------------------------------------
java.lang.Exception: null
at
com.solarwinds.oncall.sms.routs.SmppConsumerRouteBuilder$1$1.process(SmppConsumerRouteBuilder.java:66)
~[classes/:na]
at
org.apache.camel.processor.DelegateSyncProcessor.process(DelegateSyncProcessor.java:63)
~[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:72)
[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:398)
[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:105)
[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:279)
[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.processor.DefaultErrorHandler.process(DefaultErrorHandler.java:56)
[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:166)
[camel-core-2.12.3.jar:2.12.3]
at org.apache.camel.processor.Pipeline.process(Pipeline.java:118)
[camel-core-2.12.3.jar:2.12.3]
at org.apache.camel.processor.Pipeline.process(Pipeline.java:80)
[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:398)
[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.spring.spi.TransactionErrorHandler.processByErrorHandler(TransactionErrorHandler.java:218)
[camel-spring-2.12.3.jar:2.12.3]
at
org.apache.camel.spring.spi.TransactionErrorHandler.process(TransactionErrorHandler.java:99)
[camel-spring-2.12.3.jar:2.12.3]
at
org.apache.camel.spring.spi.TransactionErrorHandler.process(TransactionErrorHandler.java:112)
[camel-spring-2.12.3.jar:2.12.3]
at
org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:72)
[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:398)
[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.spring.spi.TransactionErrorHandler.processByErrorHandler(TransactionErrorHandler.java:218)
[camel-spring-2.12.3.jar:2.12.3]
at
org.apache.camel.spring.spi.TransactionErrorHandler$1.doInTransactionWithoutResult(TransactionErrorHandler.java:181)
[camel-spring-2.12.3.jar:2.12.3]
at
org.springframework.transaction.support.TransactionCallbackWithoutResult.doInTransaction(TransactionCallbackWithoutResult.java:33)
[spring-tx-3.1.1.RELEASE.jar:3.1.1.RELEASE]
at
org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:130)
[spring-tx-3.1.1.RELEASE.jar:3.1.1.RELEASE]
at
org.apache.camel.spring.spi.TransactionErrorHandler.doInTransactionTemplate(TransactionErrorHandler.java:174)
[camel-spring-2.12.3.jar:2.12.3]
at
org.apache.camel.spring.spi.TransactionErrorHandler.processInTransaction(TransactionErrorHandler.java:134)
[camel-spring-2.12.3.jar:2.12.3]
at
org.apache.camel.spring.spi.TransactionErrorHandler.process(TransactionErrorHandler.java:103)
[camel-spring-2.12.3.jar:2.12.3]
at
org.apache.camel.spring.spi.TransactionErrorHandler.process(TransactionErrorHandler.java:112)
[camel-spring-2.12.3.jar:2.12.3]
at
org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:105)
[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:87)
[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.component.smpp.MessageReceiverListenerImpl.onAcceptDeliverSm(MessageReceiverListenerImpl.java:77)
[camel-smpp-2.12.3.jar:2.12.3]
at
org.jsmpp.session.SMPPSession.fireAcceptDeliverSm(SMPPSession.java:445)
[org.apache.servicemix.bundles.jsmpp-2.1.0_4.jar:na]
at org.jsmpp.session.SMPPSession.access$0(SMPPSession.java:443)
[org.apache.servicemix.bundles.jsmpp-2.1.0_4.jar:na]
at
org.jsmpp.session.SMPPSession$ResponseHandlerImpl.processDeliverSm(SMPPSession.java:462)
[org.apache.servicemix.bundles.jsmpp-2.1.0_4.jar:na]
at
org.jsmpp.session.state.SMPPSessionBoundRX.processDeliverSm0(SMPPSessionBoundRX.java:109)
[org.apache.servicemix.bundles.jsmpp-2.1.0_4.jar:na]
at
org.jsmpp.session.state.SMPPSessionBoundRX.processDeliverSm(SMPPSessionBoundRX.java:51)
[org.apache.servicemix.bundles.jsmpp-2.1.0_4.jar:na]
at org.jsmpp.session.PDUProcessTask.run(PDUProcessTask.java:81)
[org.apache.servicemix.bundles.jsmpp-2.1.0_4.jar:na]
at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
[na:1.7.0_07]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
[na:1.7.0_07]
at java.lang.Thread.run(Unknown Source) [na:1.7.0_07]
14:19:29.597 [pool-11-thread-2] ERROR o.j.session.state.SMPPSessionBoundRX -
Failed processing deliver_sm
org.jsmpp.extra.ProcessRequestException: java.lang.Exception
at
org.apache.camel.component.smpp.MessageReceiverListenerImpl.onAcceptDeliverSm(MessageReceiverListenerImpl.java:85)
~[camel-smpp-2.12.3.jar:2.12.3]
at
org.jsmpp.session.SMPPSession.fireAcceptDeliverSm(SMPPSession.java:445)
~[org.apache.servicemix.bundles.jsmpp-2.1.0_4.jar:na]
at org.jsmpp.session.SMPPSession.access$0(SMPPSession.java:443)
~[org.apache.servicemix.bundles.jsmpp-2.1.0_4.jar:na]
at
org.jsmpp.session.SMPPSession$ResponseHandlerImpl.processDeliverSm(SMPPSession.java:462)
~[org.apache.servicemix.bundles.jsmpp-2.1.0_4.jar:na]
at
org.jsmpp.session.state.SMPPSessionBoundRX.processDeliverSm0(SMPPSessionBoundRX.java:109)
[org.apache.servicemix.bundles.jsmpp-2.1.0_4.jar:na]
at
org.jsmpp.session.state.SMPPSessionBoundRX.processDeliverSm(SMPPSessionBoundRX.java:51)
[org.apache.servicemix.bundles.jsmpp-2.1.0_4.jar:na]
at org.jsmpp.session.PDUProcessTask.run(PDUProcessTask.java:81)
[org.apache.servicemix.bundles.jsmpp-2.1.0_4.jar:na]
at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
[na:1.7.0_07]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
[na:1.7.0_07]
at java.lang.Thread.run(Unknown Source) [na:1.7.0_07]
Caused by: org.apache.camel.RuntimeCamelException: java.lang.Exception
at
org.apache.camel.util.ObjectHelper.wrapRuntimeCamelException(ObjectHelper.java:1363)
~[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.spring.spi.TransactionErrorHandler$1.doInTransactionWithoutResult(TransactionErrorHandler.java:188)
~[camel-spring-2.12.3.jar:2.12.3]
at
org.springframework.transaction.support.TransactionCallbackWithoutResult.doInTransaction(TransactionCallbackWithoutResult.java:33)
~[spring-tx-3.1.1.RELEASE.jar:3.1.1.RELEASE]
at
org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:130)
~[spring-tx-3.1.1.RELEASE.jar:3.1.1.RELEASE]
at
org.apache.camel.spring.spi.TransactionErrorHandler.doInTransactionTemplate(TransactionErrorHandler.java:174)
~[camel-spring-2.12.3.jar:2.12.3]
at
org.apache.camel.spring.spi.TransactionErrorHandler.processInTransaction(TransactionErrorHandler.java:134)
~[camel-spring-2.12.3.jar:2.12.3]
at
org.apache.camel.spring.spi.TransactionErrorHandler.process(TransactionErrorHandler.java:103)
~[camel-spring-2.12.3.jar:2.12.3]
at
org.apache.camel.spring.spi.TransactionErrorHandler.process(TransactionErrorHandler.java:112)
~[camel-spring-2.12.3.jar:2.12.3]
at
org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
~[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
~[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:105)
~[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:87)
~[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.component.smpp.MessageReceiverListenerImpl.onAcceptDeliverSm(MessageReceiverListenerImpl.java:77)
~[camel-smpp-2.12.3.jar:2.12.3]
... 9 common frames omitted
Caused by: java.lang.Exception: null
at
com.solarwinds.oncall.sms.routs.SmppConsumerRouteBuilder$1$1.process(SmppConsumerRouteBuilder.java:66)
~[classes/:na]
at
org.apache.camel.processor.DelegateSyncProcessor.process(DelegateSyncProcessor.java:63)
~[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:72)
~[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:398)
~[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:105)
~[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:279)
~[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.processor.DefaultErrorHandler.process(DefaultErrorHandler.java:56)
~[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:166)
~[camel-core-2.12.3.jar:2.12.3]
at org.apache.camel.processor.Pipeline.process(Pipeline.java:118)
~[camel-core-2.12.3.jar:2.12.3]
at org.apache.camel.processor.Pipeline.process(Pipeline.java:80)
~[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:398)
~[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.spring.spi.TransactionErrorHandler.processByErrorHandler(TransactionErrorHandler.java:218)
~[camel-spring-2.12.3.jar:2.12.3]
at
org.apache.camel.spring.spi.TransactionErrorHandler.process(TransactionErrorHandler.java:99)
~[camel-spring-2.12.3.jar:2.12.3]
at
org.apache.camel.spring.spi.TransactionErrorHandler.process(TransactionErrorHandler.java:112)
~[camel-spring-2.12.3.jar:2.12.3]
at
org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:72)
~[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:398)
~[camel-core-2.12.3.jar:2.12.3]
at
org.apache.camel.spring.spi.TransactionErrorHandler.processByErrorHandler(TransactionErrorHandler.java:218)
~[camel-spring-2.12.3.jar:2.12.3]
at
org.apache.camel.spring.spi.TransactionErrorHandler$1.doInTransactionWithoutResult(TransactionErrorHandler.java:181)
~[camel-spring-2.12.3.jar:2.12.3]
... 20 common frames omitted
Exception in thread "Thread-17" java.util.ConcurrentModificationException
at java.util.ArrayList$Itr.checkForComodification(Unknown Source)
at java.util.ArrayList$Itr.next(Unknown Source)
at
org.jsmpp.session.AbstractSessionContext.fireStateChanged(AbstractSessionContext.java:76)
at
org.jsmpp.session.SMPPSessionContext.changeState(SMPPSessionContext.java:61)
at
org.jsmpp.session.AbstractSessionContext.close(AbstractSessionContext.java:62)
at org.jsmpp.session.AbstractSession.close(AbstractSession.java:198)
at
org.jsmpp.session.SMPPSession$PDUReaderWorker.readPDU(SMPPSession.java:575)
at
org.jsmpp.session.SMPPSession$PDUReaderWorker.run(SMPPSession.java:539)
Any idea?
I really need your help. Thanks ahead.
--
View this message in context:
http://camel.465427.n5.nabble.com/smpp-transaction-doesn-t-work-tp5748385.html
Sent from the Camel - Users mailing list archive at Nabble.com.
