[Ubermq-commits] jms/src/com/ubermq/jms/server ServerTestCase.java,NONE,1.1 ServerConfig.java,1.9,1.
Brought to you by:
jimmyp
Menu
▾
▴
[Ubermq-commits] jms/src/com/ubermq/jms/server ServerTestCase.java,NONE,1.1 ServerConfig.java,1.9,1.10 MessageServer.java,1.46,1.47 ClusteredServer.java,1.4,1.5
|
From: <ji...@us...> - 2004-01-21 02:05:29
|
Update of /cvsroot/ubermq/jms/src/com/ubermq/jms/server
In directory sc8-pr-cvs1:/tmp/cvs-serv28341/src/com/ubermq/jms/server
Modified Files:
ServerConfig.java MessageServer.java ClusteredServer.java
Added Files:
ServerTestCase.java
Log Message:
bug fixes related to read/write initialization race conditions
--- NEW FILE: ServerTestCase.java ---
/*
* Contents copyright 2002-03 Rhombus Technologies.
*/
package com.ubermq.jms.server;
import java.util.*;
import javax.jms.*;
import junit.framework.*;
import com.ubermq.jms.client.*;
import com.ubermq.jms.client.test.*;
/**
* Tests various aspects of the server, such as clustering and direct pipe connections.<P>
* @author jp
* @since 2.4
*/
public class ServerTestCase extends TestCase
{
public ServerTestCase(String arg0)
{
super(arg0);
}
/**
* Tests Clustering.
*/
public void testClustering() throws JMSException
{
Properties p = new Properties();
p.put("server.port", "5001");
MessageServer ms = new MessageServer(p);
ms.addStandardProtocols();
ms.run();
p.put("server.port", "5002");
p.put("clustering.enable", "true");
p.put("clustering.forward", "ubermq://localhost:5001");
MessageServer ms2 = new MessageServer(p);
ms2.addStandardProtocols();
ms2.run();
TopicConnectionFactory localFactory = new URLTopicConnectionFactory("ubermq://localhost:5001"),
remoteFactory = new URLTopicConnectionFactory("ubermq://localhost:5002");
RegressionTestCase.sendAndReceive(localFactory, remoteFactory, "A", "A", null, 20, 20);
RegressionTestCase.sendAndReceive(localFactory, "hello", "hello", null, 20, 20);
RegressionTestCase.sendAndReceive(localFactory, "B", "B", "where ordinal != 2", 10, 9);
RegressionTestCase.sendAndReceive(remoteFactory, localFactory, "B", "B", null, 20, 20);
}
/**
* Tests pipe connections and makes sure they work w/ themselves, and others.
*/
public void testPipes() throws JMSException
{
Properties p = new Properties();
p.put("server.port", "5003");
MessageServer ms = new MessageServer(p);
ms.addStandardProtocols();
ms.run();
TopicConnectionFactory localFactory = new PipeConnectionFactory(ms),
remoteFactory = new URLTopicConnectionFactory(ms.getServiceUrl());
System.out.println("testPipes");
RegressionTestCase.sendAndReceive(localFactory, remoteFactory, "A", "A", null, 20, 20);
RegressionTestCase.sendAndReceive(localFactory, "hello", "hello", null, 20, 20);
RegressionTestCase.sendAndReceive(localFactory, "B", "B", "where ordinal != 2", 10, 9);
RegressionTestCase.sendAndReceive(remoteFactory, localFactory, "B", "B", null, 20, 20);
}
}
Index: ServerConfig.java
===================================================================
RCS file: /cvsroot/ubermq/jms/src/com/ubermq/jms/server/ServerConfig.java,v
retrieving revision 1.9
retrieving revision 1.10
diff -C2 -d -r1.9 -r1.10
*** ServerConfig.java 24 Jan 2003 15:34:49 -0000 1.9
--- ServerConfig.java 21 Jan 2004 02:05:26 -0000 1.10
***************
*** 29,33 ****
* to retrieve an instance of the object. It must also implement
* IDatagramFactory and IAckDatagramFactory.
! * @see com.ubermq.jms.server.datagram.IDatagramFactory,
* com.ubermq.jms.server.datagram.IAckDatagramFactory
*/
--- 29,33 ----
* to retrieve an instance of the object. It must also implement
* IDatagramFactory and IAckDatagramFactory.
! * @see com.ubermq.jms.common.datagram.IDatagramFactory,
* com.ubermq.jms.server.datagram.IAckDatagramFactory
*/
***************
*** 119,122 ****
--- 119,123 ----
public static final String DURABLE_LOG_SIZE = "server.durable.logsize";
+
///////////// DATAGRAM PROCESSOR
***************
*** 138,148 ****
/**
- * the size of the buffer allocated for custom properties. this can be small
- * if you are not actively using properties, but make it large if you
- * are putting objects,etc. in there
- */
- public static final String DGP_MAXIMUM_PROPERTY_LENGTH = "server.dgram.maximumprops";
-
- /**
* whether our clients care about ACK packets or not. This should usually be true
* to throttle publishers to a reasonable pace.
--- 139,142 ----
***************
*** 161,164 ****
--- 155,159 ----
public static final String DGP_OVERFLOW_HANDLER_INIT = "server.dgram.overflowhandler.init";
+
////////////////////////// SSL
Index: MessageServer.java
===================================================================
RCS file: /cvsroot/ubermq/jms/src/com/ubermq/jms/server/MessageServer.java,v
retrieving revision 1.46
retrieving revision 1.47
diff -C2 -d -r1.46 -r1.47
*** MessageServer.java 12 Jan 2004 19:24:25 -0000 1.46
--- MessageServer.java 21 Jan 2004 02:05:26 -0000 1.47
***************
*** 4,9 ****
import com.ubermq.jms.server.admin.*;
import com.ubermq.jms.server.cluster.*;
! import com.ubermq.jms.server.datagram.*;
! import com.ubermq.jms.server.datagram.impl.*;
import com.ubermq.jms.server.journal.*;
import com.ubermq.jms.server.journal.impl.*;
--- 4,10 ----
import com.ubermq.jms.server.admin.*;
import com.ubermq.jms.server.cluster.*;
! import com.ubermq.jms.common.*;
! import com.ubermq.jms.common.datagram.*;
! import com.ubermq.jms.common.datagram.impl.*;
import com.ubermq.jms.server.journal.*;
import com.ubermq.jms.server.journal.impl.*;
***************
*** 55,59 ****
private ReadWriteTransformThread[] read, write;
- public static final int DEFAULT_PORT = 3999;
private static final int CLUSTER_MAX_CXNS = 2;
private static final String ALL_TOPICS = "#";
--- 56,59 ----
***************
*** 167,171 ****
read = new ReadWriteTransformThread[RW_THREAD_COUNT];
write = new ReadWriteTransformThread[RW_THREAD_COUNT];
! com.ubermq.Utility.getLogger().debug("Creating read/write pool of size " + RW_THREAD_COUNT);
for (int i = 0; i < RW_THREAD_COUNT; i++)
{
--- 167,171 ----
read = new ReadWriteTransformThread[RW_THREAD_COUNT];
write = new ReadWriteTransformThread[RW_THREAD_COUNT];
! com.ubermq.util.Utility.getLogger().debug("Creating read/write pool of size " + RW_THREAD_COUNT);
for (int i = 0; i < RW_THREAD_COUNT; i++)
{
***************
*** 192,200 ****
{
int i = new Random().nextInt(read.length);
! read[i].register((ConnectionInfo)incoming);
! write[i].register((ConnectionInfo)incoming);
}
});
! com.ubermq.Utility.getLogger().info("Protocol " + p.toString() + " started");
if (serviceURI == null)
--- 192,200 ----
{
int i = new Random().nextInt(read.length);
! write[i].register((ConnectionInfo)incoming, true);
! read[i].register((ConnectionInfo)incoming, true);
}
});
! com.ubermq.util.Utility.getLogger().info("Protocol " + p.toString() + " started");
if (serviceURI == null)
***************
*** 208,212 ****
catch(Exception x)
{
! com.ubermq.Utility.getLogger().error("", x);;
}
--- 208,212 ----
catch(Exception x)
{
! com.ubermq.util.Utility.getLogger().error("", x);;
}
***************
*** 251,255 ****
protected void onClusterAnnouncement(String url)
{
! com.ubermq.Utility.getLogger().info("joining cluster member at " + url);
if (clusterCxnCount < CLUSTER_MAX_CXNS)
{
--- 251,255 ----
protected void onClusterAnnouncement(String url)
{
! com.ubermq.util.Utility.getLogger().info("joining cluster member at " + url);
if (clusterCxnCount < CLUSTER_MAX_CXNS)
{
***************
*** 289,293 ****
* implementation.<p>
*
! * @see com.ubermq.jms.server.datagram.impl.DatagramFactory
* @return a datagram factory holder
*/
--- 289,293 ----
* implementation.<p>
*
! * @see com.ubermq.jms.common.datagram.impl.DatagramFactory
* @return a datagram factory holder
*/
***************
*** 350,355 ****
dp);
dp.accept(ci);
! read[0].register((ConnectionInfo)ci);
! write[0].register((ConnectionInfo)ci);
return ci;
}
--- 350,355 ----
dp);
dp.accept(ci);
! write[0].register((ConnectionInfo)ci, true);
! read[0].register((ConnectionInfo)ci, true);
return ci;
}
***************
*** 365,369 ****
{
return Integer.valueOf(Configurator.getProperty(ConfigConstants.SERVER_PORT,
! String.valueOf(DEFAULT_PORT))).intValue();
}
--- 365,369 ----
{
return Integer.valueOf(Configurator.getProperty(ConfigConstants.SERVER_PORT,
! String.valueOf(MessageConstants.DEFAULT_PORT))).intValue();
}
***************
*** 415,419 ****
public String toString()
{
! return port == DEFAULT_PORT ? "UberMQ" : ("UberMQ (" + port + ")");
}
--- 415,419 ----
public String toString()
{
! return port == MessageConstants.DEFAULT_PORT ? "UberMQ" : ("UberMQ (" + port + ")");
}
***************
*** 428,432 ****
try
{
! if (port == DEFAULT_PORT)
return URI.create("ubermq://" + InetAddress.getLocalHost().getHostName());
else
--- 428,432 ----
try
{
! if (port == MessageConstants.DEFAULT_PORT)
return URI.create("ubermq://" + InetAddress.getLocalHost().getHostName());
else
***************
*** 476,480 ****
serviceURI = URI.create("//" + InetAddress.getLocalHost().getHostName() +
":" + ADMIN_PORT + "/" + ADMIN_SERVICE_NAME);
! com.ubermq.Utility.getLogger().info("Administrative service running at " + serviceURI.toString());
}
catch (java.rmi.RemoteException e)
--- 476,480 ----
serviceURI = URI.create("//" + InetAddress.getLocalHost().getHostName() +
":" + ADMIN_PORT + "/" + ADMIN_SERVICE_NAME);
! com.ubermq.util.Utility.getLogger().info("Administrative service running at " + serviceURI.toString());
}
catch (java.rmi.RemoteException e)
***************
*** 535,539 ****
// run it
s.run();
! com.ubermq.Utility.getLogger().info("UberMQ " + com.ubermq.jms.client.impl.Connection.UBERMQ_PROVIDER_VERSION + " running at " + s.getServiceUrl());
}
--- 535,539 ----
// run it
s.run();
! com.ubermq.util.Utility.getLogger().info("UberMQ " + com.ubermq.jms.client.impl.Connection.UBERMQ_PROVIDER_VERSION + " running at " + s.getServiceUrl());
}
***************
*** 574,578 ****
{
// fall through
! com.ubermq.Utility.getLogger().error("", e2);
}
}
--- 574,578 ----
{
// fall through
! com.ubermq.util.Utility.getLogger().error("", e2);
}
}
Index: ClusteredServer.java
===================================================================
RCS file: /cvsroot/ubermq/jms/src/com/ubermq/jms/server/ClusteredServer.java,v
retrieving revision 1.4
retrieving revision 1.5
diff -C2 -d -r1.4 -r1.5
*** ClusteredServer.java 12 Jan 2004 19:24:25 -0000 1.4
--- ClusteredServer.java 21 Jan 2004 02:05:26 -0000 1.5
***************
*** 107,111 ****
catch (Exception e)
{
! com.ubermq.Utility.getLogger().error("", e);
}
finally
--- 107,111 ----
catch (Exception e)
{
! com.ubermq.util.Utility.getLogger().error("", e);
}
finally
***************
*** 148,152 ****
catch (Exception e)
{
! com.ubermq.Utility.getLogger().error("", e);
}
}
--- 148,152 ----
catch (Exception e)
{
! com.ubermq.util.Utility.getLogger().error("", e);
}
}
|