camel-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Oleg <Z_O...@mail.ru>
Subject smpp transaction doesn't work
Date Thu, 06 Mar 2014 12:47:14 GMT
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://smppclient1@127.0.0.1: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.

Mime
View raw message