[hornetq-commits] JBoss hornetq SVN: r9686 - in trunk/tests/src/org/hornetq/tests/soak: client and 1 other directory.
do-not-reply at jboss.org
do-not-reply at jboss.org
Tue Sep 14 19:36:33 EDT 2010
Author: clebert.suconic at jboss.com
Date: 2010-09-14 19:36:32 -0400 (Tue, 14 Sep 2010)
New Revision: 9686
Added:
trunk/tests/src/org/hornetq/tests/soak/client/
trunk/tests/src/org/hornetq/tests/soak/client/ClientAbstract.java
trunk/tests/src/org/hornetq/tests/soak/client/ClientSoakTest.java
trunk/tests/src/org/hornetq/tests/soak/client/Receiver.java
trunk/tests/src/org/hornetq/tests/soak/client/Sender.java
Log:
Adding test to replicate leakage
Added: trunk/tests/src/org/hornetq/tests/soak/client/ClientAbstract.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/soak/client/ClientAbstract.java (rev 0)
+++ trunk/tests/src/org/hornetq/tests/soak/client/ClientAbstract.java 2010-09-14 23:36:32 UTC (rev 9686)
@@ -0,0 +1,227 @@
+/*
+ * Copyright 2010 Red Hat, Inc.
+ * Red Hat licenses this file to you under the Apache License, version
+ * 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package org.hornetq.tests.soak.client;
+
+import java.util.logging.Logger;
+
+import javax.transaction.xa.XAResource;
+import javax.transaction.xa.Xid;
+
+import org.hornetq.api.core.client.ClientSession;
+import org.hornetq.api.core.client.ClientSessionFactory;
+import org.hornetq.core.transaction.impl.XidImpl;
+import org.hornetq.utils.UUIDGenerator;
+
+/**
+ * WARNING: This is not a sample on how you should handle XA.
+ * You are supposed to use a TransactionManager.
+ * This class is doing the job of a TransactionManager that fits for the purpose of this test only,
+ * however there are many more pitfalls to deal with Transactions.
+ *
+ * This is just to stress and soak test Transactions with HornetQ.
+ *
+ * And this is dealing with XA directly for the purpose testing only.
+ *
+ * @author <a href="mailto:clebert.suconic at jboss.org">Clebert Suconic</a>
+ *
+ *
+ */
+public abstract class ClientAbstract extends Thread
+{
+
+ // Constants -----------------------------------------------------
+ private static final Logger log = Logger.getLogger(ClientAbstract.class.getName());
+
+ // Attributes ----------------------------------------------------
+
+ protected ClientSession session;
+
+ protected final ClientSessionFactory sf;
+
+ protected Xid activeXid;
+
+ protected volatile boolean running = true;
+
+ protected volatile int errors = 0;
+
+ /**
+ * A commit was called
+ * case we don't find the Xid, means it was accepted
+ */
+ protected volatile boolean pendingCommit = false;
+
+ // Static --------------------------------------------------------
+
+ // Constructors --------------------------------------------------
+
+ public ClientAbstract(ClientSessionFactory sf)
+ {
+ this.sf = sf;
+ }
+
+ // Public --------------------------------------------------------
+
+ public ClientSession getConnection()
+ {
+ return session;
+ }
+
+ public int getErrorsCount()
+ {
+ return errors;
+ }
+
+ public final void connect()
+ {
+ while (running)
+ {
+ try
+ {
+ disconnect();
+
+ session = sf.createXASession();
+
+ if (activeXid != null)
+ {
+ synchronized (ClientAbstract.class)
+ {
+ Xid[] xids = session.recover(XAResource.TMSTARTRSCAN);
+ boolean found = false;
+ for (Xid recXid : xids)
+ {
+ if (recXid.equals(activeXid))
+ {
+ // System.out.println("Calling commit after a prepare on " + this);
+ found = true;
+ callCommit();
+ }
+ }
+
+ if (!found)
+ {
+ if (pendingCommit)
+ {
+ onCommit();
+ }
+ else
+ {
+ onRollback();
+ }
+
+ activeXid = null;
+ pendingCommit = false;
+ }
+ }
+ }
+
+ connectClients();
+
+ break;
+ }
+ catch (Exception e)
+ {
+ ClientAbstract.log.warning("Can't connect to server, retrying");
+ disconnect();
+ try
+ {
+ Thread.sleep(1000);
+ }
+ catch (InterruptedException ignored)
+ {
+ // if an interruption was sent, we will respect it and leave the loop
+ break;
+ }
+ }
+ }
+ }
+
+ @Override
+ public void run()
+ {
+ connect();
+ }
+
+ protected void callCommit() throws Exception
+ {
+ pendingCommit = true;
+ session.commit(activeXid, false);
+ pendingCommit = false;
+ activeXid = null;
+ onCommit();
+ }
+
+ protected void callPrepare() throws Exception
+ {
+ session.prepare(activeXid);
+ }
+
+ public void beginTX() throws Exception
+ {
+ activeXid = newXID();
+
+ session.start(activeXid, XAResource.TMNOFLAGS);
+ }
+
+ public void endTX() throws Exception
+ {
+ session.end(activeXid, XAResource.TMSUCCESS);
+ callPrepare();
+ callCommit();
+ }
+
+ public void setRunning(final boolean running)
+ {
+ this.running = running;
+ }
+
+ /**
+ * @return
+ */
+ private XidImpl newXID()
+ {
+ return new XidImpl("tst".getBytes(), 1, UUIDGenerator.getInstance().generateStringUUID().getBytes());
+ }
+
+ protected abstract void connectClients() throws Exception;
+
+ protected abstract void onCommit();
+
+ protected abstract void onRollback();
+
+ public void disconnect()
+ {
+ try
+ {
+ if (session != null)
+ {
+ session.close();
+ }
+ }
+ catch (Exception ignored)
+ {
+ ignored.printStackTrace();
+ }
+
+ session = null;
+ }
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+
+}
Added: trunk/tests/src/org/hornetq/tests/soak/client/ClientSoakTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/soak/client/ClientSoakTest.java (rev 0)
+++ trunk/tests/src/org/hornetq/tests/soak/client/ClientSoakTest.java 2010-09-14 23:36:32 UTC (rev 9686)
@@ -0,0 +1,192 @@
+/*
+ * Copyright 2010 Red Hat, Inc.
+ * Red Hat licenses this file to you under the Apache License, version
+ * 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package org.hornetq.tests.soak.client;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.concurrent.TimeUnit;
+
+import org.hornetq.api.core.SimpleString;
+import org.hornetq.api.core.client.ClientMessage;
+import org.hornetq.api.core.client.ClientProducer;
+import org.hornetq.api.core.client.ClientSession;
+import org.hornetq.api.core.client.ClientSessionFactory;
+import org.hornetq.core.config.Configuration;
+import org.hornetq.core.config.DivertConfiguration;
+import org.hornetq.core.server.HornetQServer;
+import org.hornetq.core.settings.impl.AddressSettings;
+import org.hornetq.tests.util.ServiceTestBase;
+
+/**
+ * A ClientSoakTest
+ *
+ * @author <a href="mailto:clebert.suconic at jboss.org">Clebert Suconic</a>
+ *
+ *
+ */
+public class ClientSoakTest extends ServiceTestBase
+{
+
+ // Constants -----------------------------------------------------
+
+ // Attributes ----------------------------------------------------
+
+ private static final SimpleString ADDRESS = new SimpleString("ADD");
+
+ private static final SimpleString DIVERTED_AD1 = ClientSoakTest.ADDRESS.concat("-1");
+
+ private static final SimpleString DIVERTED_AD2 = ClientSoakTest.ADDRESS.concat("-2");
+
+ private static final boolean IS_NETTY = true;
+
+ private static final boolean IS_JOURNAL = true;
+
+ public static final int MIN_MESSAGES_ON_QUEUE = 5000;
+
+ // Static --------------------------------------------------------
+
+ // Constructors --------------------------------------------------
+
+ // Public --------------------------------------------------------
+
+ private HornetQServer server;
+
+ @Override
+ protected void setUp() throws Exception
+ {
+ clearData();
+
+ Configuration config = createDefaultConfig(ClientSoakTest.IS_NETTY);
+
+ config.setJournalFileSize(10 * 1024 * 1024);
+
+ server = createServer(IS_JOURNAL, config, -1, -1, new HashMap<String, AddressSettings>());
+
+ DivertConfiguration divert1 = new DivertConfiguration("dv1",
+ "nm1",
+ ClientSoakTest.ADDRESS.toString(),
+ ClientSoakTest.DIVERTED_AD1.toString(),
+ true,
+ null,
+ null);
+
+ DivertConfiguration divert2 = new DivertConfiguration("dv2",
+ "nm2",
+ ClientSoakTest.ADDRESS.toString(),
+ ClientSoakTest.DIVERTED_AD2.toString(),
+ true,
+ null,
+ null);
+
+ ArrayList<DivertConfiguration> divertList = new ArrayList<DivertConfiguration>();
+ divertList.add(divert1);
+ divertList.add(divert2);
+
+ config.setDivertConfigurations(divertList);
+
+ server.start();
+
+ ClientSessionFactory sf = createFactory(ClientSoakTest.IS_NETTY);
+
+ ClientSession session = sf.createSession();
+
+ session.createQueue(ClientSoakTest.ADDRESS, ClientSoakTest.ADDRESS, true);
+
+ session.createQueue(ClientSoakTest.DIVERTED_AD1, ClientSoakTest.DIVERTED_AD1, true);
+
+ session.createQueue(ClientSoakTest.DIVERTED_AD2, ClientSoakTest.DIVERTED_AD2, true);
+
+ session.close();
+
+ sf.close();
+
+ }
+
+ @Override
+ protected void tearDown() throws Exception
+ {
+ server.stop();
+ server = null;
+ }
+
+ public void testSoakClient() throws Exception
+ {
+ final ClientSessionFactory sf = createFactory(IS_NETTY);
+
+ ClientSession session = sf.createSession(false, false);
+
+ ClientProducer producer = session.createProducer(ADDRESS);
+
+ for (int i = 0; i < MIN_MESSAGES_ON_QUEUE; i++)
+ {
+ ClientMessage msg = session.createMessage(true);
+ msg.putLongProperty("count", i);
+ msg.getBodyBuffer().writeBytes(new byte[10 * 1024]);
+ producer.send(msg);
+
+ if (i % 1000 == 0)
+ {
+ System.out.println("Sent " + i + " messages");
+ session.commit();
+ }
+ }
+
+ session.commit();
+
+ session.close();
+ sf.close();
+
+ Receiver rec1 = new Receiver(createFactory(IS_NETTY), DIVERTED_AD1.toString());
+ Receiver rec2 = new Receiver(createFactory(IS_NETTY), DIVERTED_AD2.toString());
+
+ Sender send = new Sender(createFactory(IS_NETTY), ADDRESS.toString(), new Receiver[] { rec1, rec2 });
+
+ send.start();
+ rec1.start();
+ rec2.start();
+
+ long timeEnd = System.currentTimeMillis() + TimeUnit.HOURS.toMillis(1);
+ while (timeEnd > System.currentTimeMillis())
+ {
+ if (send.getErrorsCount() != 0 || rec1.getErrorsCount() != 0 || rec2.getErrorsCount() != 0)
+ {
+ System.out.println("There are sequence errors in some of the clients, please look at the logs");
+ break;
+ }
+ Thread.sleep(10000);
+ }
+
+ send.setRunning(false);
+ rec1.setRunning(false);
+ rec2.setRunning(false);
+
+ send.join();
+ rec1.join();
+ rec2.join();
+
+ assertEquals(0, send.getErrorsCount());
+ assertEquals(0, rec1.getErrorsCount());
+ assertEquals(0, rec2.getErrorsCount());
+
+ }
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+
+}
Added: trunk/tests/src/org/hornetq/tests/soak/client/Receiver.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/soak/client/Receiver.java (rev 0)
+++ trunk/tests/src/org/hornetq/tests/soak/client/Receiver.java 2010-09-14 23:36:32 UTC (rev 9686)
@@ -0,0 +1,185 @@
+/*
+ * Copyright 2010 Red Hat, Inc.
+ * Red Hat licenses this file to you under the Apache License, version
+ * 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package org.hornetq.tests.soak.client;
+
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.hornetq.api.core.client.ClientConsumer;
+import org.hornetq.api.core.client.ClientMessage;
+import org.hornetq.api.core.client.ClientSessionFactory;
+import org.hornetq.utils.ReusableLatch;
+
+/**
+ * A Receiver
+ *
+ * @author <a href="mailto:clebert.suconic at jboss.org">Clebert Suconic</a>
+ *
+ *
+ */
+public class Receiver extends ClientAbstract
+{
+
+ // Constants -----------------------------------------------------
+
+ // Attributes ----------------------------------------------------
+
+ // We should leave some messages on paging. We don't want to consume all for this test
+ private final Semaphore minConsume = new Semaphore(0);
+
+ private final ReusableLatch latchMax = new ReusableLatch(0);
+
+ private static final int MAX_DIFF = 10000;
+
+ // The difference between producer and consuming
+ private final AtomicInteger currentDiff = new AtomicInteger(0);
+
+ private final String queue;
+
+ protected long msgs = 0;
+
+ protected int pendingMsgs = 0;
+
+ protected int pendingSemaphores = 0;
+
+ protected ClientConsumer cons;
+
+
+ // Static --------------------------------------------------------
+
+ // Constructors --------------------------------------------------
+
+ public Receiver(ClientSessionFactory sf, String queue)
+ {
+ super(sf);
+ this.queue = queue;
+ }
+
+ // Public --------------------------------------------------------
+
+ public void run()
+ {
+ super.run();
+
+ while (running)
+ {
+ try
+ {
+ beginTX();
+
+ for (int i = 0 ; i < 1000; i++)
+ {
+ ClientMessage msg = cons.receive(5000);
+ if (msg == null)
+ {
+ break;
+ }
+
+ msg.acknowledge();
+
+ if (msg.getLongProperty("count") != msgs + pendingMsgs)
+ {
+ errors++;
+ System.out.println("count should be " + (msgs + pendingMsgs) + " when it was " + msg.getLongProperty("count") + " on " + queue);
+ }
+
+ pendingMsgs++;
+ if (!minConsume.tryAcquire(1, 5, TimeUnit.SECONDS))
+ {
+ break;
+ }
+
+ }
+
+ endTX();
+ }
+ catch (Exception e)
+ {
+ connect();
+ }
+
+
+ }
+ }
+
+ /* (non-Javadoc)
+ * @see org.hornetq.jms.example.ClientAbstract#connectClients()
+ */
+ @Override
+ protected void connectClients() throws Exception
+ {
+
+ cons = session.createConsumer(queue);
+
+ session.start();
+ }
+
+ /* (non-Javadoc)
+ * @see org.hornetq.jms.example.ClientAbstract#onCommit()
+ */
+ @Override
+ protected void onCommit()
+ {
+ msgs += pendingMsgs;
+ this.currentDiff.addAndGet(-pendingMsgs);
+ latchMax.countDown(pendingMsgs);
+ pendingMsgs = 0;
+ }
+
+ /* (non-Javadoc)
+ * @see org.hornetq.jms.example.ClientAbstract#onRollback()
+ */
+ @Override
+ protected void onRollback()
+ {
+ minConsume.release(pendingMsgs);
+ pendingMsgs = 0;
+ }
+
+ public String toString()
+ {
+ return "Receiver::" + this.queue + ", msgs=" + msgs + ", pending=" + pendingMsgs;
+ }
+
+ /**
+ * @param pendingMsgs2
+ */
+ public void messageProduced(int producedMessages)
+ {
+ minConsume.release(producedMessages);
+ currentDiff.addAndGet(producedMessages);
+ if (currentDiff.get() > MAX_DIFF)
+ {
+ latchMax.setCount(currentDiff.get() - MAX_DIFF);
+ try
+ {
+ latchMax.await(5, TimeUnit.SECONDS);
+ }
+ catch (InterruptedException e)
+ {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+
+}
Added: trunk/tests/src/org/hornetq/tests/soak/client/Sender.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/soak/client/Sender.java (rev 0)
+++ trunk/tests/src/org/hornetq/tests/soak/client/Sender.java 2010-09-14 23:36:32 UTC (rev 9686)
@@ -0,0 +1,126 @@
+/*
+ * Copyright 2010 Red Hat, Inc.
+ * Red Hat licenses this file to you under the Apache License, version
+ * 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package org.hornetq.tests.soak.client;
+
+import org.hornetq.api.core.client.ClientMessage;
+import org.hornetq.api.core.client.ClientProducer;
+import org.hornetq.api.core.client.ClientSessionFactory;
+
+/**
+ * A Sender
+ *
+ * @author <a href="mailto:clebert.suconic at jboss.org">Clebert Suconic</a>
+ *
+ *
+ */
+public class Sender extends ClientAbstract
+{
+
+ // Constants -----------------------------------------------------
+
+ // Attributes ----------------------------------------------------
+
+ protected ClientProducer producer;
+
+ protected String queue;
+
+ protected long msgs = ClientSoakTest.MIN_MESSAGES_ON_QUEUE;
+
+ protected int pendingMsgs = 0;
+
+ protected final Receiver[] receivers;
+
+ // Static --------------------------------------------------------
+
+ // Constructors --------------------------------------------------
+
+ // Public --------------------------------------------------------
+
+ public Sender(final ClientSessionFactory sf, String queue, final Receiver[] receivers)
+ {
+ super(sf);
+ this.receivers = receivers;
+ this.queue = queue;
+ }
+
+ @Override
+ protected void connectClients() throws Exception
+ {
+ producer = session.createProducer(queue);
+ }
+
+ public void run()
+ {
+ super.run();
+ while (running)
+ {
+ try
+ {
+ beginTX();
+ for (int i = 0; i < 1000; i++)
+ {
+ ClientMessage msg = session.createMessage(true);
+ msg.putLongProperty("count", pendingMsgs + msgs);
+ msg.getBodyBuffer().writeBytes(new byte[10 * 1024]);
+ producer.send(msg);
+ pendingMsgs++;
+ }
+ endTX();
+ }
+ catch (Exception e)
+ {
+ connect();
+ }
+ }
+ }
+
+ /* (non-Javadoc)
+ * @see org.hornetq.jms.example.ClientAbstract#onCommit()
+ */
+ @Override
+ protected void onCommit()
+ {
+ this.msgs += pendingMsgs;
+ for (Receiver rec : receivers)
+ {
+ rec.messageProduced(pendingMsgs);
+ }
+
+ pendingMsgs = 0;
+ }
+
+ /* (non-Javadoc)
+ * @see org.hornetq.jms.example.ClientAbstract#onRollback()
+ */
+ @Override
+ protected void onRollback()
+ {
+ pendingMsgs = 0;
+ }
+
+ public String toString()
+ {
+ return "Sender, msgs=" + msgs + ", pending=" + pendingMsgs;
+
+ }
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+
+}
More information about the hornetq-commits
mailing list