Modified: webservices/sandesha/trunk/java/src/org/apache/sandesha2/util/SandeshaUtil.java URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/src/org/apache/sandesha2/util/SandeshaUtil.java?view=diff&rev=470218&r1=470217&r2=470218 ============================================================================== --- webservices/sandesha/trunk/java/src/org/apache/sandesha2/util/SandeshaUtil.java (original) +++ webservices/sandesha/trunk/java/src/org/apache/sandesha2/util/SandeshaUtil.java Wed Nov 1 20:08:26 2006 @@ -18,7 +18,9 @@ package org.apache.sandesha2.util; import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; import java.io.OutputStream; +import java.io.PrintWriter; import java.lang.reflect.Constructor; import java.util.ArrayList; import java.util.Iterator; @@ -220,14 +222,6 @@ sender.runSenderForTheSequence(context, sequenceID); } } - - private static void stopSenderForTheSequence(String sequenceID, ConfigurationContext context) { - Sender sender = (Sender) context.getProperty(Sandesha2Constants.SENDER); - - if (sender!=null) { - sender.stopSenderForTheSequence(sequenceID); - } - } public static void stopSender(ConfigurationContext context) { Sender sender = (Sender) context.getProperty(Sandesha2Constants.SENDER); @@ -267,12 +261,6 @@ if (pollingManager!=null) pollingManager.stopPolling (); } - - private static void stopInvokerForTheSequence(String sequenceID, ConfigurationContext context) { - Invoker invoker = (Invoker) context.getProperty(Sandesha2Constants.INVOKER); - if (invoker!=null) - invoker.stopInvokerForTheSequence(sequenceID); - } public static void stopInvoker(ConfigurationContext context) { Invoker invoker = (Invoker) context.getProperty(Sandesha2Constants.INVOKER); @@ -1148,5 +1136,22 @@ return newMsg; } + + /** + * Returns an Exception as a String + * + * @param exc + * @return + */ + public static String getStackTrace(Throwable exc) + { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + PrintWriter pw = new PrintWriter(baos); + exc.printStackTrace(pw); + pw.flush(); + String stackTrace = baos.toString(); + return stackTrace; + } + }
Modified: webservices/sandesha/trunk/java/src/org/apache/sandesha2/workers/SenderWorker.java URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/src/org/apache/sandesha2/workers/SenderWorker.java?view=diff&rev=470218&r1=470217&r2=470218 ============================================================================== --- webservices/sandesha/trunk/java/src/org/apache/sandesha2/workers/SenderWorker.java (original) +++ webservices/sandesha/trunk/java/src/org/apache/sandesha2/workers/SenderWorker.java Wed Nov 1 20:08:26 2006 @@ -11,7 +11,6 @@ import org.apache.axis2.context.OperationContext; import org.apache.axis2.description.TransportOutDescription; import org.apache.axis2.engine.AxisEngine; -import org.apache.axis2.transport.TransportSender; import org.apache.axis2.transport.TransportUtils; import org.apache.axis2.transport.http.HTTPConstants; import org.apache.commons.logging.Log; @@ -25,11 +24,12 @@ import org.apache.sandesha2.storage.StorageManager; import org.apache.sandesha2.storage.Transaction; import org.apache.sandesha2.storage.beanmanagers.SenderBeanMgr; +import org.apache.sandesha2.storage.beanmanagers.SequencePropertyBeanMgr; import org.apache.sandesha2.storage.beans.SenderBean; +import org.apache.sandesha2.storage.beans.SequencePropertyBean; import org.apache.sandesha2.util.AcknowledgementManager; import org.apache.sandesha2.util.MessageRetransmissionAdjuster; import org.apache.sandesha2.util.MsgInitializer; -import org.apache.sandesha2.util.RMMsgCreator; import org.apache.sandesha2.util.SandeshaUtil; import org.apache.sandesha2.util.TerminateManager; import org.apache.sandesha2.wsrm.TerminateSequence; @@ -52,6 +52,9 @@ public void run () { + if (log.isDebugEnabled()) + log.debug("Enter: SenderWorker::run"); + Transaction transaction = null; try { @@ -76,6 +79,8 @@ boolean continueSending = MessageRetransmissionAdjuster.adjustRetransmittion(rmMsgCtx, senderBean, configurationContext, storageManager); if (!continueSending) { + if (log.isDebugEnabled()) + log.debug("Exit: SenderWorker::run, !continueSending"); return; } @@ -88,6 +93,8 @@ String qualifiedForSending = (String) msgCtx.getProperty(Sandesha2Constants.QUALIFIED_FOR_SENDING); if (qualifiedForSending != null && !qualifiedForSending.equals(Sandesha2Constants.VALUE_TRUE)) { + if (log.isDebugEnabled()) + log.debug("Exit: SenderWorker::run, !qualified for sending"); return; } @@ -101,6 +108,8 @@ ArrayList msgsNotToSend = SandeshaUtil.getPropertyBean(msgCtx.getAxisOperation()).getMsgTypesToDrop(); if (msgsNotToSend != null && msgsNotToSend.contains(new Integer(rmMsgCtx.getMessageType()))) { + if (log.isDebugEnabled()) + log.debug("Exit: SenderWorker::run, message type to be dropped " + rmMsgCtx.getMessageType()); return; } @@ -171,7 +180,55 @@ } catch (Exception e) { String message = SandeshaMessageHelper.getMessage( SandeshaMessageKeys.sendMsgError, e.toString()); - log.error(message, e); + + if (log.isErrorEnabled()) + log.error(message, e); + // Store the Exception as a sequence property to enable the client to lookup the last + // exception time and timestamp. + + // Create a new Transaction + transaction = storageManager.getTransaction(); + + try + { + + // Get the internal sequence id from the context + String internalSequenceId = (String)rmMsgCtx.getProperty(Sandesha2Constants.MessageContextProperties.INTERNAL_SEQUENCE_ID); + + // Get the sequence property bean manager + SequencePropertyBeanMgr seqPropMgr = storageManager.getSequencePropertyBeanMgr(); + + // Add the new sequence property beans. + String exceptionStr = SandeshaUtil.getStackTrace(e); + SequencePropertyBean eBean = + new SequencePropertyBean(internalSequenceId, + Sandesha2Constants.SequenceProperties.LAST_FAILED_TO_SEND_ERROR, + exceptionStr); + + SequencePropertyBean etsBean = + new SequencePropertyBean(internalSequenceId, + Sandesha2Constants.SequenceProperties.LAST_FAILED_TO_SEND_ERROR_TIMESTAMP, + String.valueOf(System.currentTimeMillis())); + + + // Insert the exception bean + seqPropMgr.insert(eBean); + + // Insert the timestamp bean + seqPropMgr.insert(etsBean); + + // Commit the properties + transaction.commit(); + } + catch (Exception e1) + { + if (log.isErrorEnabled()) + log.error(e1); + + if (transaction != null && transaction.isActive()) + transaction.rollback(); + } + } finally { transaction = storageManager.getTransaction(); msgCtx.setProperty(Sandesha2Constants.WITHIN_TRANSACTION, @@ -224,16 +281,30 @@ msgCtx.setProperty(Sandesha2Constants.WITHIN_TRANSACTION, Sandesha2Constants.VALUE_FALSE); } catch (SandeshaStorageException e) { + if (log.isDebugEnabled()) + log.debug(e); if (transaction!=null && transaction.isActive()) transaction.rollback(); } catch (SandeshaException e) { - e.printStackTrace(); + if (log.isDebugEnabled()) + log.debug(e); + if (transaction!=null && transaction.isActive()) + transaction.rollback(); } catch (MissingResourceException e) { - e.printStackTrace(); + if (log.isFatalEnabled()) + log.fatal("Unable to load message bundle", e); + if (transaction!=null && transaction.isActive()) + transaction.rollback(); } catch (AxisFault e) { - e.printStackTrace(); + if (log.isDebugEnabled()) + log.debug(e); + if (transaction!=null && transaction.isActive()) + transaction.rollback(); } catch (Exception e) { - e.printStackTrace(); + if (log.isDebugEnabled()) + log.debug(e); + if (transaction!=null && transaction.isActive()) + transaction.rollback(); } finally { if (transaction!=null && transaction.isActive()) transaction.commit(); @@ -242,6 +313,9 @@ lock.removeWork(workId); } } + + if (log.isDebugEnabled()) + log.debug("Exit: SenderWorker::run"); } private void updateMessage(MessageContext msgCtx1) throws SandeshaException { @@ -347,7 +421,7 @@ log.debug("Exit: Sender::checkForSyncResponses"); } - private boolean isFaultEnvelope(SOAPEnvelope envelope) throws SandeshaException { + private boolean isFaultEnvelope(SOAPEnvelope envelope) { if (log.isDebugEnabled()) log.debug("Enter: Sender::isFaultEnvelope, " + envelope.getBody().getFault()); SOAPFault fault = envelope.getBody().getFault(); Modified: webservices/sandesha/trunk/java/test/src/org/apache/sandesha2/SandeshaClientTest.java URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/test/src/org/apache/sandesha2/SandeshaClientTest.java?view=diff&rev=470218&r1=470217&r2=470218 ============================================================================== --- webservices/sandesha/trunk/java/test/src/org/apache/sandesha2/SandeshaClientTest.java (original) +++ webservices/sandesha/trunk/java/test/src/org/apache/sandesha2/SandeshaClientTest.java Wed Nov 1 20:08:26 2006 @@ -18,14 +18,19 @@ import java.io.File; +import org.apache.axiom.om.OMAbstractFactory; +import org.apache.axiom.om.OMElement; +import org.apache.axiom.om.OMFactory; +import org.apache.axiom.om.OMNamespace; +import org.apache.axiom.soap.SOAP11Constants; import org.apache.axis2.AxisFault; import org.apache.axis2.Constants; +import org.apache.axis2.Constants.Configuration; import org.apache.axis2.addressing.EndpointReference; import org.apache.axis2.client.Options; import org.apache.axis2.client.ServiceClient; import org.apache.axis2.context.ConfigurationContext; import org.apache.axis2.context.ConfigurationContextFactory; -import org.apache.axis2.context.MessageContextConstants; import org.apache.axis2.transport.http.SimpleHTTPServer; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -40,18 +45,17 @@ private Log log = LogFactory.getLog(getClass()); int serverPort = DEFAULT_SERVER_TEST_PORT; + private static final String applicationNamespaceName = "http://tempuri.org/";; + private static final String ping = "ping"; + private static final String Text = "Text"; + + public SandeshaClientTest () { super ("SandeshaClientTest"); } - public void setUp () throws AxisFault { + public void setUp () { - String repoPath = "target" + File.separator + "repos" + File.separator + "server"; - String axis2_xml = "target" + File.separator + "repos" + File.separator + "server" + File.separator + "server_axis2.xml"; - - - ConfigurationContext configContext = ConfigurationContextFactory.createConfigurationContextFromFileSystem(repoPath,axis2_xml); - String serverPortStr = getTestProperty("test.server.port"); if (serverPortStr!=null) { @@ -62,6 +66,15 @@ } } + } + + private void startServer() throws AxisFault { + + String repoPath = "target" + File.separator + "repos" + File.separator + "server"; + String axis2_xml = "target" + File.separator + "repos" + File.separator + "server" + File.separator + "server_axis2.xml"; + + + ConfigurationContext configContext = ConfigurationContextFactory.createConfigurationContextFromFileSystem(repoPath,axis2_xml); httpServer = new SimpleHTTPServer (configContext,serverPort); httpServer.start(); try { @@ -69,6 +82,7 @@ } catch (InterruptedException e) { throw new SandeshaException ("sleep interupted"); } + } public void tearDown () throws SandeshaException { @@ -84,6 +98,8 @@ public void testCreateSequenceWithOffer () throws AxisFault,InterruptedException { + startServer(); + String to = "http://127.0.0.1:"; + serverPort + "/axis2/services/RMSampleService"; String transportTo = "http://127.0.0.1:"; + serverPort + "/axis2/services/RMSampleService"; @@ -95,7 +111,7 @@ Options clientOptions = new Options (); clientOptions.setTo(new EndpointReference (to)); - clientOptions.setProperty(MessageContextConstants.TRANSPORT_URL,transportTo); + clientOptions.setProperty(Configuration.TRANSPORT_URL,transportTo); String sequenceKey = SandeshaUtil.getUUID(); clientOptions.setProperty(SandeshaClientConstants.SEQUENCE_KEY,sequenceKey); @@ -156,5 +172,75 @@ // // } + /** + * Tests that the last error and timestamp are set for the simple case of the target service not being available + */ + public void testLastErrorAndTimestamp() throws Exception + { + String to = "http://127.0.0.1:"; + serverPort + "/axis2/services/RMSampleService"; + String transportTo = "http://127.0.0.1:"; + serverPort + "/axis2/services/RMSampleService"; + + String repoPath = "target" + File.separator + "repos" + File.separator + "client"; + String axis2_xml = "target" + File.separator + "repos" + File.separator + "client" + File.separator + "client_axis2.xml"; + + ConfigurationContext configContext = ConfigurationContextFactory.createConfigurationContextFromFileSystem(repoPath,axis2_xml); + + //clientOptions.setSoapVersionURI(SOAP12Constants.SOAP_ENVELOPE_NAMESPACE_URI); + Options clientOptions = new Options (); + clientOptions.setSoapVersionURI(SOAP11Constants.SOAP_ENVELOPE_NAMESPACE_URI); + + clientOptions.setTo(new EndpointReference (to)); + clientOptions.setProperty(Configuration.TRANSPORT_URL,transportTo); + + String sequenceKey = "sequence1"; + clientOptions.setProperty(SandeshaClientConstants.SEQUENCE_KEY,sequenceKey); + + ServiceClient serviceClient = new ServiceClient (configContext,null); + + serviceClient.setOptions(clientOptions); + + serviceClient.fireAndForget(getPingOMBlock("ping1")); + + //starting the server after a wait + Thread.sleep(10000); + + // Check that the last error and last error time stamp have been set + String lastSendError = SandeshaClient.getLastSendError(serviceClient); + long lastSendErrorTime = SandeshaClient.getLastSendErrorTimestamp(serviceClient); + + // Check the values are valid + assertNotNull(lastSendError); + assertTrue(lastSendErrorTime > -1); + + startServer(); + + clientOptions.setProperty(SandeshaClientConstants.LAST_MESSAGE, "true"); + serviceClient.fireAndForget(getPingOMBlock("ping2")); + + + Thread.sleep(10000); + + SequenceReport sequenceReport = SandeshaClient.getOutgoingSequenceReport(serviceClient); + assertTrue(sequenceReport.getCompletedMessages().contains(new Long(1))); + assertTrue(sequenceReport.getCompletedMessages().contains(new Long(2))); + assertEquals(sequenceReport.getSequenceStatus(),SequenceReport.SEQUENCE_STATUS_TERMINATED); + assertEquals(sequenceReport.getSequenceDirection(),SequenceReport.SEQUENCE_DIRECTION_OUT); + + configContext.getListenerManager().stop(); + serviceClient.cleanup(); + } + + private OMElement getPingOMBlock(String text) { + OMFactory fac = OMAbstractFactory.getOMFactory(); + OMNamespace namespace = fac.createOMNamespace(applicationNamespaceName,"ns1"); + OMElement pingElem = fac.createOMElement(ping, namespace); + OMElement textElem = fac.createOMElement(Text, namespace); + + textElem.setText(text); + pingElem.addChild(textElem); + + return pingElem; + } + } --------------------------------------------------------------------- To unsubscribe, e-mail: [EMAIL PROTECTED] For additional commands, e-mail: [EMAIL PROTECTED]
