Modified: webservices/sandesha/trunk/java/modules/persistence/src/main/java/org/apache/sandesha2/storage/jdbc/PersistentStorageManager.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/modules/persistence/src/main/java/org/apache/sandesha2/storage/jdbc/PersistentStorageManager.java?rev=708805&r1=708804&r2=708805&view=diff
==============================================================================
--- webservices/sandesha/trunk/java/modules/persistence/src/main/java/org/apache/sandesha2/storage/jdbc/PersistentStorageManager.java (original)
+++ webservices/sandesha/trunk/java/modules/persistence/src/main/java/org/apache/sandesha2/storage/jdbc/PersistentStorageManager.java Tue Oct 28 23:41:11 2008
@@ -47,39 +47,39 @@
/**
* A Storage Manager implementation for managing Sandesha2 beans.
- *
+ * <p/>
* Needs this parameter in module.xml or axis2.xml :
- *
- * db.driver JDBC Driver class name
- * db.connectionstring JDBC connection string
- * db.user Data Base user name
- * db.password Data Base user password
- *
- * Transactions are supposed to be attached to a thread (see inMemoryStorageManager)
- * hence the ThreadLocal threadTransaction variable (instead of the transactions HashMap
- * used by inMemoryStorageManager).
- *
- * MessageContexts are stored in a HashMap, as in inMemoryStorageManager, AND in DataBase
- * as backup in case of failure.
+ * <p/>
+ * db.driver JDBC Driver class name
+ * db.connectionstring JDBC connection string
+ * db.user Data Base user name
+ * db.password Data Base user password
+ * <p/>
+ * Transactions are supposed to be attached to a thread (see inMemoryStorageManager)
+ * hence the ThreadLocal threadTransaction variable (instead of the transactions HashMap
+ * used by inMemoryStorageManager).
+ * <p/>
+ * MessageContexts are stored in a HashMap, as in inMemoryStorageManager, AND in DataBase
+ * as backup in case of failure.
*/
public class PersistentStorageManager extends StorageManager {
-
+
private Connection DbConnection = null;
private String DbConnectionString = null;
private String DbDriver = null;
private String DbUser = null;
private String DbPassword = null;
- private PersistentRMSBeanMgr pRMSBeanMgr = null;
- private PersistentRMDBeanMgr pRMDBeanMgr = null;
- private PersistentSenderBeanMgr pSenderBeanMgr = null;
- private PersistentInvokerBeanMgr pInvokerBeanMgr = null;
- private Sender sender = null;
- private Invoker invoker = null;
- private PollingManager pollingManager = null;
- private boolean useSerialization = false;
- private HashMap<String,MessageContext> storageMap = null;
- private static ThreadLocal threadTransaction = null;
+ private PersistentRMSBeanMgr pRMSBeanMgr = null;
+ private PersistentRMDBeanMgr pRMDBeanMgr = null;
+ private PersistentSenderBeanMgr pSenderBeanMgr = null;
+ private PersistentInvokerBeanMgr pInvokerBeanMgr = null;
+ private Sender sender = null;
+ private Invoker invoker = null;
+ private PollingManager pollingManager = null;
+ private boolean useSerialization = false;
+ private HashMap<String, MessageContext> storageMap = null;
+ private static ThreadLocal threadTransaction = null;
private static final Log log = LogFactory.getLog(PersistentStorageManager.class);
public SandeshaThread getInvoker() {
@@ -94,76 +94,72 @@
return sender;
}
-
+
public PersistentStorageManager(ConfigurationContext context)
- throws SandeshaException
- {
- super (context);
+ throws SandeshaException {
+ super(context);
log.info("create PersistentStorageManager");
- storageMap = new HashMap ();
+ storageMap = new HashMap();
threadTransaction = new ThreadLocal();
- pRMSBeanMgr = new PersistentRMSBeanMgr (this);
- pRMDBeanMgr = new PersistentRMDBeanMgr (this);
- pSenderBeanMgr = new PersistentSenderBeanMgr (this);
- pInvokerBeanMgr = new PersistentInvokerBeanMgr (this);
+ pRMSBeanMgr = new PersistentRMSBeanMgr(this);
+ pRMDBeanMgr = new PersistentRMDBeanMgr(this);
+ pSenderBeanMgr = new PersistentSenderBeanMgr(this);
+ pInvokerBeanMgr = new PersistentInvokerBeanMgr(this);
sender = new Sender();
// Note that while inOrder is a global property we can decide if we need the
// invoker thread at this point. If we change this to be a sequence-level
// property then we'll need to revisit this.
SandeshaPolicyBean policy = SandeshaUtil.getPropertyBean(context.getAxisConfiguration());
useSerialization = policy.isUseMessageSerialization();
- if(policy.isInOrder()) invoker = new Invoker();
- if(policy.isEnableMakeConnection()) pollingManager = new PollingManager();
- ModuleConfiguration mc = context.getAxisConfiguration().getModuleConfig("sandesha2");
- Parameter param = mc.getParameter("db.connectionstring");
- if ( param != null ) {
- DbConnectionString = (String)param.getValue();
- log.debug(param.getName() + "=" + DbConnectionString);
+ if (policy.isInOrder()) invoker = new Invoker();
+ if (policy.isEnableMakeConnection()) pollingManager = new PollingManager();
+ ModuleConfiguration mc = context.getAxisConfiguration().getModuleConfig("sandesha2");
+ Parameter param = mc.getParameter("db.connectionstring");
+ if (param != null) {
+ DbConnectionString = (String) param.getValue();
+ log.debug(param.getName() + "=" + DbConnectionString);
}
param = mc.getParameter("db.driver");
- if ( param != null ) {
- DbDriver = (String)param.getValue();
- log.debug(param.getName() + "=" + DbDriver);
+ if (param != null) {
+ DbDriver = (String) param.getValue();
+ log.debug(param.getName() + "=" + DbDriver);
}
param = mc.getParameter("db.user");
- if ( param != null ) {
- DbUser = (String)param.getValue();
- log.debug(param.getName() + "=" + DbUser);
+ if (param != null) {
+ DbUser = (String) param.getValue();
+ log.debug(param.getName() + "=" + DbUser);
}
param = mc.getParameter("db.password");
- if ( param != null ) {
- DbPassword = (String)param.getValue();
- log.debug(param.getName() + "=" + DbPassword);
+ if (param != null) {
+ DbPassword = (String) param.getValue();
+ log.debug(param.getName() + "=" + DbPassword);
}
}
- public void shutdown()
- {
- if ( DbConnection != null ) {
+
+ public void shutdown() {
+ if (DbConnection != null) {
try {
DbConnection.close();
DbConnection = null;
- } catch (Exception ex) {}
+ } catch (Exception ex) {
+ }
}
super.shutdown();
}
- public InvokerBeanMgr getInvokerBeanMgr()
- {
+ public InvokerBeanMgr getInvokerBeanMgr() {
return pInvokerBeanMgr;
}
- public RMDBeanMgr getRMDBeanMgr()
- {
+ public RMDBeanMgr getRMDBeanMgr() {
return pRMDBeanMgr;
}
- public RMSBeanMgr getRMSBeanMgr()
- {
+ public RMSBeanMgr getRMSBeanMgr() {
return pRMSBeanMgr;
}
- public SenderBeanMgr getSenderBeanMgr()
- {
+ public SenderBeanMgr getSenderBeanMgr() {
return pSenderBeanMgr;
}
@@ -171,15 +167,13 @@
return useSerialization;
}
- public boolean hasUserTransaction(MessageContext msg)
- {
+ public boolean hasUserTransaction(MessageContext msg) {
// Answer to : Is there a user transaction in play ?
// but what is a 'user transaction' ?
return false;
}
-
- public Transaction getTransaction()
- {
+
+ public Transaction getTransaction() {
Transaction transaction = (Transaction) threadTransaction.get();
if (transaction == null) {
transaction = new JDBCTransaction(this);
@@ -188,139 +182,131 @@
// We don't want to overwrite or return an existing transaction, as someone
// else should decide if we commit it or not. If we get here then we probably
// have a bug.
- if(log.isDebugEnabled()) log.debug("Possible re-used transaction: ");
+ if (log.isDebugEnabled()) log.debug("Possible re-used transaction: ");
transaction = null;
}
return transaction;
}
-
- public void removeTransaction()
- {
+
+ public void removeTransaction() {
threadTransaction.set(null);
}
/**
- * Returns the connection attached to the current transaction if exists
- * or the "common" connection.
- *
+ * Returns the connection attached to the current transaction if exists
+ * or the "common" connection.
+ *
* @return Data Base Connection
*/
- public Connection getDbConnection()
- {
+ public Connection getDbConnection() {
JDBCTransaction transaction = (JDBCTransaction) threadTransaction.get();
- if ( transaction == null )return DbConnection;
+ if (transaction == null) return DbConnection;
return transaction.getDbConnection();
}
-
- public void initStorage (AxisModule moduleDesc)
- throws SandeshaStorageException
- {
+
+ public void initStorage(AxisModule moduleDesc)
+ throws SandeshaStorageException {
log.info("init PersistentStorageManager");
- if ( DbConnectionString == null || DbDriver == null )
- throw new SandeshaStorageException ("Can't proceed. Needed properties are not set.");
+ if (DbConnectionString == null || DbDriver == null)
+ throw new SandeshaStorageException("Can't proceed. Needed properties are not set.");
- DbConnection = dbConnect();
+ DbConnection = dbConnect();
}
-
- public Connection dbConnect ()
- throws SandeshaStorageException
- {
- try {
- Class.forName(DbDriver);
- return DriverManager.getConnection(DbConnectionString,DbUser,DbPassword);
- } catch (Exception ex) {
- log.error("Unable to create DB connection ", ex);
- throw new SandeshaStorageException(ex);
- }
+
+ public Connection dbConnect()
+ throws SandeshaStorageException {
+ try {
+ Class.forName(DbDriver);
+ return DriverManager.getConnection(DbConnectionString, DbUser, DbPassword);
+ } catch (Exception ex) {
+ log.error("Unable to create DB connection ", ex);
+ throw new SandeshaStorageException(ex);
+ }
}
- public MessageContext retrieveMessageContext(String key,ConfigurationContext configContext)
- throws SandeshaStorageException
- {
+ public MessageContext retrieveMessageContext(String key, ConfigurationContext configContext)
+ throws SandeshaStorageException {
log.debug("Enter retrieveMessageContext for key " + key);
- /**/
+ /**/
if (storageMap.containsKey(key)) {
log.debug("retrieveMessageContext get from cache");
return (MessageContext) storageMap.get(key);
}
- /**/
+ /**/
try {
- Statement stmt = getDbConnection().createStatement();
- /**/
- ResultSet rs = stmt.executeQuery("select * from wsrm_msgctx where ctx_key='" +
+ Statement stmt = getDbConnection().createStatement();
+ /**/
+ ResultSet rs = stmt.executeQuery("select * from wsrm_msgctx where ctx_key='" +
key + "'");
- rs.next();
- MessageContext msgCtx = new MessageContext();
- msgCtx.readExternal(new ObjectInputStream(rs.getBinaryStream("ctx")));
- msgCtx.activate(configContext);
- msgCtx.setProperty(Sandesha2Constants.POST_FAILURE_MESSAGE,Sandesha2Constants.VALUE_TRUE);
- rs.close();
- stmt.close();
- log.debug("RetrieveMessageContext get from DB");
- return msgCtx;
- } catch ( Exception ex) {
+ rs.next();
+ MessageContext msgCtx = new MessageContext();
+ msgCtx.readExternal(new ObjectInputStream(rs.getBinaryStream("ctx")));
+ msgCtx.activate(configContext);
+ msgCtx.setProperty(Sandesha2Constants.POST_FAILURE_MESSAGE, Sandesha2Constants.VALUE_TRUE);
+ rs.close();
+ stmt.close();
+ log.debug("RetrieveMessageContext get from DB");
+ return msgCtx;
+ } catch (Exception ex) {
log.error("RetrieveMessageContext exception " + ex);
throw new SandeshaStorageException(ex);
}
}
- synchronized public void storeMessageContext(String key,MessageContext msgContext)
- throws SandeshaStorageException
- {
- if ( log.isDebugEnabled() ) log.debug("Enter storeMessageContext for key " + key + " context " + msgContext);
- storageMap.put(key,msgContext);
+ synchronized public void storeMessageContext(String key, MessageContext msgContext)
+ throws SandeshaStorageException {
+ if (log.isDebugEnabled()) log.debug("Enter storeMessageContext for key " + key + " context " + msgContext);
+ storageMap.put(key, msgContext);
try {
- PreparedStatement pstmt = getDbConnection().prepareStatement("insert into wsrm_msgctx(ctx_key,ctx)values(?,?)");
- pstmt.setString(1,key);
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- ObjectOutputStream oos = new ObjectOutputStream(baos);
- msgContext.writeExternal(oos);
- oos.close();
- ByteArrayInputStream bais = new ByteArrayInputStream(baos.toByteArray());
- pstmt.setBinaryStream(2, bais, bais.available());
- pstmt.execute();
- pstmt.close();
- } catch(Exception ex) {
+ PreparedStatement pstmt = getDbConnection().prepareStatement("insert into wsrm_msgctx(ctx_key,ctx)values(?,?)");
+ pstmt.setString(1, key);
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ ObjectOutputStream oos = new ObjectOutputStream(baos);
+ msgContext.writeExternal(oos);
+ oos.close();
+ ByteArrayInputStream bais = new ByteArrayInputStream(baos.toByteArray());
+ pstmt.setBinaryStream(2, bais, bais.available());
+ pstmt.execute();
+ pstmt.close();
+ } catch (Exception ex) {
throw new SandeshaStorageException(ex);
}
}
- synchronized public void updateMessageContext(String key,MessageContext msgContext)
- throws SandeshaStorageException
- {
- if ( log.isDebugEnabled() ) log.debug("updateMessageContext key : " + key);
- storageMap.put(key,msgContext);
+ synchronized public void updateMessageContext(String key, MessageContext msgContext)
+ throws SandeshaStorageException {
+ if (log.isDebugEnabled()) log.debug("updateMessageContext key : " + key);
+ storageMap.put(key, msgContext);
PreparedStatement pstmt = null;
try {
- pstmt = getDbConnection().prepareStatement("update wsrm_msgctx set ctx=?" +
- "where ctx_key='" + key + "'");
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- ObjectOutputStream oos = new ObjectOutputStream(baos);
- msgContext.writeExternal(oos);
- oos.close();
- ByteArrayInputStream bais = new ByteArrayInputStream(baos.toByteArray());
- pstmt.setBinaryStream(1, bais, bais.available());
- pstmt.executeQuery();
- pstmt.close();
- } catch(Exception ex) {
+ pstmt = getDbConnection().prepareStatement("update wsrm_msgctx set ctx=?" +
+ "where ctx_key='" + key + "'");
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ ObjectOutputStream oos = new ObjectOutputStream(baos);
+ msgContext.writeExternal(oos);
+ oos.close();
+ ByteArrayInputStream bais = new ByteArrayInputStream(baos.toByteArray());
+ pstmt.setBinaryStream(1, bais, bais.available());
+ pstmt.executeQuery();
+ pstmt.close();
+ } catch (Exception ex) {
throw new SandeshaStorageException("Exception in updateMessageContext", ex);
}
}
-
+
public void removeMessageContext(String key)
- throws SandeshaStorageException
- {
- if ( log.isDebugEnabled() ) log.debug("removeMessageContext key : " + key);
+ throws SandeshaStorageException {
+ if (log.isDebugEnabled()) log.debug("removeMessageContext key : " + key);
try {
- Statement stmt = getDbConnection().createStatement();
- MessageContext messageInCache = (MessageContext) storageMap.get(key);
- if (messageInCache!=null) storageMap.remove(key);
- stmt.executeUpdate("delete from wsrm_msgctx where ctx_key='" + key + "'");
- stmt.close();
- } catch (Exception ex ) {
+ Statement stmt = getDbConnection().createStatement();
+ MessageContext messageInCache = (MessageContext) storageMap.get(key);
+ if (messageInCache != null) storageMap.remove(key);
+ stmt.executeUpdate("delete from wsrm_msgctx where ctx_key='" + key + "'");
+ stmt.close();
+ } catch (Exception ex) {
throw new SandeshaStorageException("Exception in removeMessageContext", ex);
}
}
-
+
}
Modified: webservices/sandesha/trunk/java/modules/persistence/src/resources/module_config_axis2.xml
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/modules/persistence/src/resources/module_config_axis2.xml?rev=708805&r1=708804&r2=708805&view=diff
==============================================================================
--- webservices/sandesha/trunk/java/modules/persistence/src/resources/module_config_axis2.xml (original)
+++ webservices/sandesha/trunk/java/modules/persistence/src/resources/module_config_axis2.xml Tue Oct 28 23:41:11 2008
@@ -1,15 +1,15 @@
<!-- Add something like this in axis2.xml for the right DB -->
- <moduleConfig name="sandesha2">
- <!-- MySQL
- <parameter name="db.connectionstring">jdbc:mysql://localhost/wsrm</parameter>
- <parameter name="db.driver">com.mysql.jdbc.Driver</parameter>
- <parameter name="db.user"></parameter>
- <parameter name="db.password"></parameter>
- -->
- <!-- Derby -->
- <parameter name="db.connectionstring">jdbc:derby:derby/wsrm</parameter>
- <parameter name="db.driver">org.apache.derby.jdbc.EmbeddedDriver</parameter>
- <parameter name="db.user"></parameter>
- <parameter name="db.password"></parameter>
- <!-- -->
- </moduleConfig>
+<moduleConfig name="sandesha2">
+ <!-- MySQL
+ <parameter name="db.connectionstring">jdbc:mysql://localhost/wsrm</parameter>
+ <parameter name="db.driver">com.mysql.jdbc.Driver</parameter>
+ <parameter name="db.user"></parameter>
+ <parameter name="db.password"></parameter>
+ -->
+ <!-- Derby -->
+ <parameter name="db.connectionstring">jdbc:derby:derby/wsrm</parameter>
+ <parameter name="db.driver">org.apache.derby.jdbc.EmbeddedDriver</parameter>
+ <parameter name="db.user"></parameter>
+ <parameter name="db.password"></parameter>
+ <!-- -->
+</moduleConfig>
Modified: webservices/sandesha/trunk/java/modules/persistence/src/test/java/org/apache/sandesha2/RMScenariosTest.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/modules/persistence/src/test/java/org/apache/sandesha2/RMScenariosTest.java?rev=708805&r1=708804&r2=708805&view=diff
==============================================================================
--- webservices/sandesha/trunk/java/modules/persistence/src/test/java/org/apache/sandesha2/RMScenariosTest.java (original)
+++ webservices/sandesha/trunk/java/modules/persistence/src/test/java/org/apache/sandesha2/RMScenariosTest.java Tue Oct 28 23:41:11 2008
@@ -39,46 +39,46 @@
public class RMScenariosTest extends SandeshaTestCase {
- private boolean serverStarted = false;
+ private boolean serverStarted = false;
protected ConfigurationContext configContext = null;
protected String to = "http://127.0.0.1:" + serverPort + "/axis2/services/RMSampleService";
-
+
protected String repoPath = "target" + File.separator + "repos" + File.separator + "persistence-server";
protected String axis2_xml = "target" + File.separator + "repos" + File.separator + "persistence-server" + File.separator + "server_axis2.xml";
protected String repoPathClient = "target" + File.separator + "repos" + File.separator + "persistence-client";
protected String axis2_xmlClient = "target" + File.separator + "repos" + File.separator + "persistence-client" + File.separator + "client_axis2.xml";
-
- public RMScenariosTest () {
- super ("RMScenariosTest");
+
+ public RMScenariosTest() {
+ super("RMScenariosTest");
}
-
- public RMScenariosTest (String name) {
+
+ public RMScenariosTest(String name) {
super(name);
}
- public void setUp () throws Exception {
+ public void setUp() throws Exception {
super.setUp();
if (!serverStarted) {
startServer(repoPath, axis2_xml);
- configContext = ConfigurationContextFactory.createConfigurationContextFromFileSystem(repoPathClient,axis2_xmlClient);
+ configContext = ConfigurationContextFactory.createConfigurationContextFromFileSystem(repoPathClient, axis2_xmlClient);
}
serverStarted = true;
}
-
+
/**
* Override the teardown processing
*/
- public void tearDown () throws Exception {
+ public void tearDown() throws Exception {
super.tearDown();
}
- public void testPing() throws Exception {
+ public void testPing() throws Exception {
// Run a ping test with sync acks
runPing(false, false);
-
+
// Run a ping test with async acks
runPing(true, true);
}
@@ -86,25 +86,25 @@
public void testAsyncEcho() throws Exception {
// Test async echo with sync acks
Options clientOptions = new Options();
- runEcho(clientOptions, true, false, false,true,false);
-
+ runEcho(clientOptions, true, false, false, true, false);
+
// Test async echo with async acks
clientOptions = new Options();
- runEcho(clientOptions, true, true, false,true,false);
-
+ runEcho(clientOptions, true, true, false, true, false);
+
// Test async echo with async acks and offer
clientOptions = new Options();
- clientOptions.setProperty(SandeshaClientConstants.OFFERED_SEQUENCE_ID,SandeshaUtil.getUUID());
- runEcho(clientOptions, true, true, false,true,true);
+ clientOptions.setProperty(SandeshaClientConstants.OFFERED_SEQUENCE_ID, SandeshaUtil.getUUID());
+ runEcho(clientOptions, true, true, false, true, true);
}
-
+
public void testSyncEchoWithOffer() throws Exception {
// Test sync echo with an offer, and the 1.1 spec
Options clientOptions = new Options();
- clientOptions.setProperty(SandeshaClientConstants.OFFERED_SEQUENCE_ID,SandeshaUtil.getUUID());
- clientOptions.setProperty(SandeshaClientConstants.RM_SPEC_VERSION,Sandesha2Constants.SPEC_VERSIONS.v1_1);
- runEcho(clientOptions, false, false, true,true,false);
-
+ clientOptions.setProperty(SandeshaClientConstants.OFFERED_SEQUENCE_ID, SandeshaUtil.getUUID());
+ clientOptions.setProperty(SandeshaClientConstants.RM_SPEC_VERSION, Sandesha2Constants.SPEC_VERSIONS.v1_1);
+ runEcho(clientOptions, false, false, true, true, false);
+
// // Test sync echo with an offer, and the 1.0 spec. The offer is not automatic as this
// // is a client that hasn't been built from WSDL. If the user's operations had been
// // modelled properly then the offer would happen automatically.
@@ -112,40 +112,40 @@
// clientOptions.setProperty(SandeshaClientConstants.OFFERED_SEQUENCE_ID,SandeshaUtil.getUUID());
// clientOptions.setProperty(SandeshaClientConstants.RM_SPEC_VERSION,Sandesha2Constants.SPEC_VERSIONS.v1_0);
// runEcho(clientOptions, false, false, true,false,false);
- }
+ }
- public void testSyncEcho() throws Exception {
+ public void testSyncEcho() throws Exception {
// Test sync echo with no offer, and the 1.1 spec
Options clientOptions = new Options();
- clientOptions.setProperty(SandeshaClientConstants.RM_SPEC_VERSION,Sandesha2Constants.SPEC_VERSIONS.v1_1);
- runEcho(clientOptions, false, false, true,true,true);
+ clientOptions.setProperty(SandeshaClientConstants.RM_SPEC_VERSION, Sandesha2Constants.SPEC_VERSIONS.v1_1);
+ runEcho(clientOptions, false, false, true, true, true);
}
public void runPing(boolean asyncAcks, boolean stopListener) throws Exception {
-
+
Options clientOptions = new Options();
- ServiceClient serviceClient = new ServiceClient (configContext,null);
+ ServiceClient serviceClient = new ServiceClient(configContext, null);
serviceClient.setOptions(clientOptions);
String sequenceKey = SandeshaUtil.getUUID();
clientOptions.setAction(pingAction);
- clientOptions.setTo(new EndpointReference (to));
- clientOptions.setProperty(SandeshaClientConstants.SEQUENCE_KEY,sequenceKey);
+ clientOptions.setTo(new EndpointReference(to));
+ clientOptions.setProperty(SandeshaClientConstants.SEQUENCE_KEY, sequenceKey);
clientOptions.setProperty(SandeshaClientConstants.LAST_MESSAGE, Constants.VALUE_TRUE);
-
- if(asyncAcks) {
+
+ if (asyncAcks) {
clientOptions.setUseSeparateListener(true);
}
serviceClient.fireAndForget(getPingOMBlock("ping1"));
-
+
long limit = System.currentTimeMillis() + waitTime;
Error lastError = null;
- while(System.currentTimeMillis() < limit) {
+ while (System.currentTimeMillis() < limit) {
Thread.sleep(tickTime); // Try the assertions each tick interval, until they pass or we time out
-
+
try {
SequenceReport sequenceReport = SandeshaClient.getOutgoingSequenceReport(serviceClient);
System.out.println("Checking Outbound Sequence: " + sequenceReport.getSequenceID());
@@ -155,86 +155,87 @@
lastError = null;
break;
- } catch(Error e) {
- e.printStackTrace();
- System.out.println("Possible error:" + e);
+ } catch (Error e) {
+ e.printStackTrace();
+ System.out.println("Possible error:" + e);
lastError = e;
}
}
- if(lastError != null) throw lastError;
+ if (lastError != null) throw lastError;
if (stopListener)
configContext.getListenerManager().stop();
-
+
serviceClient.cleanup();
}
public void runEcho(Options clientOptions, boolean asyncReply, boolean asyncAcks, boolean explicitTermination, boolean checkInboundTermination, boolean stopListener) throws Exception {
-
+
String sequenceKey = SandeshaUtil.getUUID();
- ServiceClient serviceClient = new ServiceClient (configContext,null);
+ ServiceClient serviceClient = new ServiceClient(configContext, null);
serviceClient.setOptions(clientOptions);
clientOptions.setAction(echoAction);
- clientOptions.setTo(new EndpointReference (to));
- clientOptions.setProperty(SandeshaClientConstants.SEQUENCE_KEY,sequenceKey);
+ clientOptions.setTo(new EndpointReference(to));
+ clientOptions.setProperty(SandeshaClientConstants.SEQUENCE_KEY, sequenceKey);
clientOptions.setTransportInProtocol(Constants.TRANSPORT_HTTP);
- if(asyncReply || asyncAcks) {
+ if (asyncReply || asyncAcks) {
clientOptions.setUseSeparateListener(true);
-
- if(asyncAcks) {
+
+ if (asyncAcks) {
String acksTo = serviceClient.getMyEPR(Constants.TRANSPORT_HTTP).getAddress();
- clientOptions.setProperty(SandeshaClientConstants.AcksTo,acksTo);
+ clientOptions.setProperty(SandeshaClientConstants.AcksTo, acksTo);
} else {
String acksTo = AddressingConstants.Final.WSA_ANONYMOUS_URL;
- clientOptions.setProperty(SandeshaClientConstants.AcksTo,acksTo);
+ clientOptions.setProperty(SandeshaClientConstants.AcksTo, acksTo);
}
}
-
- if(asyncAcks) {
+
+ if (asyncAcks) {
String acksTo = serviceClient.getMyEPR(Constants.TRANSPORT_HTTP).getAddress();
- clientOptions.setProperty(SandeshaClientConstants.AcksTo,acksTo);
+ clientOptions.setProperty(SandeshaClientConstants.AcksTo, acksTo);
} else {
String acksTo = AddressingConstants.Final.WSA_ANONYMOUS_URL;
- clientOptions.setProperty(SandeshaClientConstants.AcksTo,acksTo);
+ clientOptions.setProperty(SandeshaClientConstants.AcksTo, acksTo);
}
// Establish a baseline count for inbound sequences
List oldIncomingReports = SandeshaClient.getIncomingSequenceReports(configContext);
-
- TestCallback callback1 = new TestCallback ("Callback 1");
- serviceClient.sendReceiveNonBlocking (getEchoOMBlock("echo1",sequenceKey),callback1);
-
- TestCallback callback2 = new TestCallback ("Callback 2");
- serviceClient.sendReceiveNonBlocking (getEchoOMBlock("echo2",sequenceKey),callback2);
-
- if (!explicitTermination
- &&
- !Sandesha2Constants.SPEC_VERSIONS.v1_1.equals(clientOptions.getProperty(SandeshaClientConstants.RM_SPEC_VERSION))) {
-
+
+ TestCallback callback1 = new TestCallback("Callback 1");
+ serviceClient.sendReceiveNonBlocking(getEchoOMBlock("echo1", sequenceKey), callback1);
+
+ TestCallback callback2 = new TestCallback("Callback 2");
+ serviceClient.sendReceiveNonBlocking(getEchoOMBlock("echo2", sequenceKey), callback2);
+
+ if (!explicitTermination
+ &&
+ !Sandesha2Constants.SPEC_VERSIONS.v1_1.equals(clientOptions.getProperty(SandeshaClientConstants.RM_SPEC_VERSION)))
+ {
+
clientOptions.setProperty(SandeshaClientConstants.LAST_MESSAGE, Constants.VALUE_TRUE);
}
-
- TestCallback callback3 = new TestCallback ("Callback 3");
- serviceClient.sendReceiveNonBlocking (getEchoOMBlock("echo3",sequenceKey),callback3);
-
+
+ TestCallback callback3 = new TestCallback("Callback 3");
+ serviceClient.sendReceiveNonBlocking(getEchoOMBlock("echo3", sequenceKey), callback3);
+
if (explicitTermination) {
Thread.sleep(10000);
SandeshaClient.terminateSequence(serviceClient);
}
-
+
long limit = System.currentTimeMillis() + waitTime;
Error lastError = null;
- while(System.currentTimeMillis() < limit) {
+ while (System.currentTimeMillis() < limit) {
Thread.sleep(tickTime); // Try the assertions each tick interval, until they pass or we time out
-
+
try {
-
- //assertions for the out sequence.
+
+ //assertions for the out sequence.
SequenceReport outgoingSequenceReport = SandeshaClient.getOutgoingSequenceReport(serviceClient);
System.out.println("Checking Outbound Sequence: " + outgoingSequenceReport.getSequenceID());
assertTrue("Outbound message #1", outgoingSequenceReport.getCompletedMessages().contains(new Long(1)));
@@ -242,46 +243,46 @@
assertTrue("Outbound message #3", outgoingSequenceReport.getCompletedMessages().contains(new Long(3)));
assertEquals("Outbound sequence status: TERMINATED", SequenceReport.SEQUENCE_STATUS_TERMINATED, outgoingSequenceReport.getSequenceStatus());
assertEquals("Outbound sequence direction: OUT", SequenceReport.SEQUENCE_DIRECTION_OUT, outgoingSequenceReport.getSequenceDirection());
-
+
//assertions for the inbound sequence. The one we care about is a new sequence,
//so it will not exist in the oldSequences list.
List incomingSequences = SandeshaClient.getIncomingSequenceReports(configContext);
SequenceReport incomingSequenceReport = getNewReport(incomingSequences, oldIncomingReports);
System.out.println("Checking Inbound Sequence: " + incomingSequenceReport.getSequenceID());
String offer = (String) clientOptions.getProperty(SandeshaClientConstants.OFFERED_SEQUENCE_ID);
- if(offer != null) assertEquals("Inbound seq id", offer, incomingSequenceReport.getSequenceID());
- assertEquals ("Inbound message count", 3, incomingSequenceReport.getCompletedMessages().size());
+ if (offer != null) assertEquals("Inbound seq id", offer, incomingSequenceReport.getSequenceID());
+ assertEquals("Inbound message count", 3, incomingSequenceReport.getCompletedMessages().size());
assertTrue("Inbound message #1", incomingSequenceReport.getCompletedMessages().contains(new Long(1)));
assertTrue("Inbound message #2", incomingSequenceReport.getCompletedMessages().contains(new Long(2)));
assertTrue("Inbound message #3", incomingSequenceReport.getCompletedMessages().contains(new Long(3)));
-
+
if (checkInboundTermination)
assertEquals("Inbound sequence status: TERMINATED", SequenceReport.SEQUENCE_STATUS_TERMINATED, incomingSequenceReport.getSequenceStatus());
-
+
assertEquals("Inbound sequence direction: IN", SequenceReport.SEQUENCE_DIRECTION_IN, incomingSequenceReport.getSequenceDirection());
-
+
assertTrue("Callback #1", callback1.isComplete());
assertEquals("Callback #1 data", "echo1", callback1.getResult());
-
+
assertTrue("Callback #2", callback2.isComplete());
assertEquals("Callback #2 data", "echo1echo2", callback2.getResult());
-
+
assertTrue("Callback #3", callback3.isComplete());
assertEquals("Callback #3 data", "echo1echo2echo3", callback3.getResult());
-
+
lastError = null;
break;
- } catch(Error e) {
- e.printStackTrace();
- System.out.println("Possible error:" + e);
+ } catch (Error e) {
+ e.printStackTrace();
+ System.out.println("Possible error:" + e);
lastError = e;
}
}
- if(lastError != null) throw lastError;
-
+ if (lastError != null) throw lastError;
+
if (stopListener)
configContext.getListenerManager().stop();
-
+
serviceClient.cleanup();
}
@@ -289,13 +290,13 @@
// was established by this test. Note that some of the old sequences may have timed out.
private SequenceReport getNewReport(List incomingSequences, List oldIncomingReports) {
HashSet sequenceIds = new HashSet();
- for(Iterator oldSequences = oldIncomingReports.iterator(); oldSequences.hasNext(); ) {
+ for (Iterator oldSequences = oldIncomingReports.iterator(); oldSequences.hasNext();) {
SequenceReport report = (SequenceReport) oldSequences.next();
sequenceIds.add(report.getSequenceID());
}
- for(Iterator currentSequences = incomingSequences.iterator(); currentSequences.hasNext(); ) {
+ for (Iterator currentSequences = incomingSequences.iterator(); currentSequences.hasNext();) {
SequenceReport report = (SequenceReport) currentSequences.next();
- if(!sequenceIds.contains(report.getSequenceID())) {
+ if (!sequenceIds.contains(report.getSequenceID())) {
return report;
}
}
Modified: webservices/sandesha/trunk/java/modules/persistence/src/test/java/org/apache/sandesha2/SandeshaTestCase.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/modules/persistence/src/test/java/org/apache/sandesha2/SandeshaTestCase.java?rev=708805&r1=708804&r2=708805&view=diff
==============================================================================
--- webservices/sandesha/trunk/java/modules/persistence/src/test/java/org/apache/sandesha2/SandeshaTestCase.java (original)
+++ webservices/sandesha/trunk/java/modules/persistence/src/test/java/org/apache/sandesha2/SandeshaTestCase.java Tue Oct 28 23:41:11 2008
@@ -52,16 +52,16 @@
import org.apache.commons.logging.LogFactory;
public class SandeshaTestCase extends TestCase {
-
+
String resourceDir = ""; //"test-resources";
- Properties properties = null;
- final String PROPERTY_FILE_NAME = "sandesha2-test.properties";
- public final int DEFAULT_SERVER_TEST_PORT = 8060;
- public ConfigurationContext serverConfigurationContext = null;
- private final String RMServiceName = "RMSampleService";
+ Properties properties = null;
+ final String PROPERTY_FILE_NAME = "sandesha2-test.properties";
+ public final int DEFAULT_SERVER_TEST_PORT = 8060;
+ public ConfigurationContext serverConfigurationContext = null;
+ private final String RMServiceName = "RMSampleService";
private Log log = LogFactory.getLog(getClass());
-
- private final static String applicationNamespaceName = "http://tempuri.org/";
+
+ private final static String applicationNamespaceName = "http://tempuri.org/";
private final static String echoString = "echoString";
private final static String ping = "ping";
private final static String Text = "Text";
@@ -75,111 +75,111 @@
protected int tickTime = 10000; // Each wait will check the test assertions each second
protected String pingAction = "urn:wsrm:Ping";
protected String echoAction = "urn:wsrm:EchoString";
-
- public SandeshaTestCase(String name) {
- super(name);
- String testRource = "target" + File.separator + "test-classes";
- resourceDir = new File(testRource).getPath();
-
- String propFileStr = resourceDir + File.separator + PROPERTY_FILE_NAME;
- properties = new Properties ();
-
- try {
- FileInputStream propertyFile = new FileInputStream (new File(propFileStr));
+
+ public SandeshaTestCase(String name) {
+ super(name);
+ String testRource = "target" + File.separator + "test-classes";
+ resourceDir = new File(testRource).getPath();
+
+ String propFileStr = resourceDir + File.separator + PROPERTY_FILE_NAME;
+ properties = new Properties();
+
+ try {
+ FileInputStream propertyFile = new FileInputStream(new File(propFileStr));
properties.load(propertyFile);
} catch (FileNotFoundException e) {
log.error(e);
} catch (IOException e) {
log.error(e);
}
- }
-
- public void setUp () throws Exception {
+ }
+
+ public void setUp() throws Exception {
super.setUp();
-
+
String serverPortStr = getTestProperty("test.server.port");
- if (serverPortStr!=null) {
+ if (serverPortStr != null) {
try {
serverPort = Integer.parseInt(serverPortStr);
} catch (NumberFormatException e) {
log.error(e);
}
}
- }
-
+ }
+
public ConfigurationContext startServer(String repoPath, String axis2_xml)
- throws Exception {
+ throws Exception {
ConfigurationContext configContext =
- ConfigurationContextFactory.createConfigurationContextFromFileSystem(repoPath,axis2_xml);
+ ConfigurationContextFactory.createConfigurationContextFromFileSystem(repoPath, axis2_xml);
- httpServer = new SimpleHTTPServer (configContext,serverPort);
+ httpServer = new SimpleHTTPServer(configContext, serverPort);
httpServer.start();
Thread.sleep(300);
-
+
return configContext;
}
- public void tearDown () throws Exception {
- if (httpServer!=null) {
+ public void tearDown() throws Exception {
+ if (httpServer != null) {
httpServer.stop();
}
-
+
Thread.sleep(300);
}
protected InputStreamReader getResource(String relativePath, String resourceName) {
- String resourceFile = resourceDir + relativePath + File.separator + resourceName;
- try {
- FileReader reader = new FileReader(resourceFile);
- return reader;
- } catch (FileNotFoundException e) {
- throw new RuntimeException("cannot load the test-resource", e);
- }
- }
-
- protected SOAPEnvelope getSOAPEnvelope() {
- return OMAbstractFactory.getSOAP11Factory().getDefaultEnvelope();
- }
-
- protected SOAPEnvelope getSOAPEnvelope(String relativePath, String resourceName) {
- try {
- XMLStreamReader reader = XMLInputFactory.newInstance().createXMLStreamReader(
- getResource(relativePath, resourceName));
- OMXMLParserWrapper wrapper = OMXMLBuilderFactory.createStAXSOAPModelBuilder(
- OMAbstractFactory.getSOAP11Factory(), reader);
- return (SOAPEnvelope) wrapper.getDocumentElement();
-
- } catch (XMLStreamException e) {
- throw new RuntimeException(e);
- }
- }
-
- protected SOAPEnvelope getEmptySOAPEnvelope() {
- return OMAbstractFactory.getSOAP11Factory().getDefaultEnvelope();
- }
+ String resourceFile = resourceDir + relativePath + File.separator + resourceName;
+ try {
+ FileReader reader = new FileReader(resourceFile);
+ return reader;
+ } catch (FileNotFoundException e) {
+ throw new RuntimeException("cannot load the test-resource", e);
+ }
+ }
+
+ protected SOAPEnvelope getSOAPEnvelope() {
+ return OMAbstractFactory.getSOAP11Factory().getDefaultEnvelope();
+ }
+
+ protected SOAPEnvelope getSOAPEnvelope(String relativePath, String resourceName) {
+ try {
+ XMLStreamReader reader = XMLInputFactory.newInstance().createXMLStreamReader(
+ getResource(relativePath, resourceName));
+ OMXMLParserWrapper wrapper = OMXMLBuilderFactory.createStAXSOAPModelBuilder(
+ OMAbstractFactory.getSOAP11Factory(), reader);
+ return (SOAPEnvelope) wrapper.getDocumentElement();
+
+ } catch (XMLStreamException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ protected SOAPEnvelope getEmptySOAPEnvelope() {
+ return OMAbstractFactory.getSOAP11Factory().getDefaultEnvelope();
+ }
- protected static OMElement getEchoOMBlock(String text, String sequenceKey) {
+ protected static OMElement getEchoOMBlock(String text, String sequenceKey) {
OMFactory fac = OMAbstractFactory.getOMFactory();
- OMNamespace applicationNamespace = fac.createOMNamespace(applicationNamespaceName,"ns1");
+ OMNamespace applicationNamespace = fac.createOMNamespace(applicationNamespaceName, "ns1");
OMElement echoStringElement = fac.createOMElement(echoString, applicationNamespace);
- OMElement textElem = fac.createOMElement(Text,applicationNamespace);
- OMElement sequenceElem = fac.createOMElement(Sequence,applicationNamespace);
-
+ OMElement textElem = fac.createOMElement(Text, applicationNamespace);
+ OMElement sequenceElem = fac.createOMElement(Sequence, applicationNamespace);
+
textElem.setText(text);
sequenceElem.setText(sequenceKey);
echoStringElement.addChild(textElem);
echoStringElement.addChild(sequenceElem);
-
+
return echoStringElement;
}
-
+
protected OMElement getPingOMBlock(String text) {
OMFactory fac = OMAbstractFactory.getOMFactory();
- OMNamespace namespace = fac.createOMNamespace(applicationNamespaceName,"ns1");
+ OMNamespace namespace = fac.createOMNamespace(applicationNamespaceName, "ns1");
OMElement pingElem = fac.createOMElement(ping, namespace);
OMElement textElem = fac.createOMElement(Text, namespace);
-
+
textElem.setText(text);
pingElem.addChild(textElem);
@@ -189,39 +189,39 @@
protected String checkEchoOMBlock(OMElement response) {
assertEquals("Response namespace", applicationNamespaceName, response.getNamespace().getNamespaceURI());
assertEquals("Response local name", echoStringResponse, response.getLocalName());
-
- OMElement echoStringReturnElem = response.getFirstChildWithName(new QName (applicationNamespaceName,EchoStringReturn));
+
+ OMElement echoStringReturnElem = response.getFirstChildWithName(new QName(applicationNamespaceName, EchoStringReturn));
assertNotNull("Echo String Return", echoStringReturnElem);
-
+
String resultStr = echoStringReturnElem.getText();
return resultStr;
}
- public String getTestProperty (String key) {
- if (properties!=null)
- return properties.getProperty(key);
-
- return null;
- }
-
- public void overrideConfigurationContext (ConfigurationContext context,MessageReceiver messageReceiver, String operationName, boolean newOperation, int mep) throws Exception {
-
-
- AxisService rmService = context.getAxisConfiguration().getService(RMServiceName);
-
- AxisOperation operation = null;
-
- if (newOperation) {
- operation = rmService.getOperation(new QName (operationName));
- if (operation==null)
- throw new Exception ("Given operation not found");
- } else {
- operation = AxisOperationFactory.getAxisOperation(mep);
- rmService.addOperation(operation);
- }
-
- operation.setMessageReceiver(messageReceiver);
- }
+ public String getTestProperty(String key) {
+ if (properties != null)
+ return properties.getProperty(key);
+
+ return null;
+ }
+
+ public void overrideConfigurationContext(ConfigurationContext context, MessageReceiver messageReceiver, String operationName, boolean newOperation, int mep) throws Exception {
+
+
+ AxisService rmService = context.getAxisConfiguration().getService(RMServiceName);
+
+ AxisOperation operation = null;
+
+ if (newOperation) {
+ operation = rmService.getOperation(new QName(operationName));
+ if (operation == null)
+ throw new Exception("Given operation not found");
+ } else {
+ operation = AxisOperationFactory.getAxisOperation(mep);
+ rmService.addOperation(operation);
+ }
+
+ operation.setMessageReceiver(messageReceiver);
+ }
protected class TestCallback implements AxisCallback {
@@ -229,23 +229,23 @@
boolean completed = false;
boolean errorReported = false;
String resultStr;
-
+
public boolean isComplete() {
return completed;
}
-
+
public boolean isErrorReported() {
return errorReported;
}
- public String getResult () {
+ public String getResult() {
return resultStr;
}
-
- public TestCallback (String name) {
+
+ public TestCallback(String name) {
this.name = name;
}
-
+
public void onComplete() {
completed = true;
}
@@ -256,13 +256,13 @@
this.resultStr = checkEchoOMBlock(contents);
System.out.println("TestCallback got text: '" + resultStr + "'");
}
-
+
public void onFault(MessageContext result) {
errorReported = true;
System.out.println("TestCallback got fault: " + result.getEnvelope());
}
- public void onError (Exception e) {
+ public void onError(Exception e) {
errorReported = true;
System.out.println("TestCallback got exception");
e.printStackTrace();
lmpx.com only provides a reader for public news (NNTP) servers. It is not
affiliated with the servers or forums shown here and is not responsible for
the content of articles, which is written by their respective authors.