[hornetq-commits] JBoss hornetq SVN: r10501 - in branches/Branch_2_2_EAP: src/main/org/hornetq/core/client/impl and 5 other directories.

Author: clebert.suconic(a)jboss.com Date: 2011-04-13 14:00:12 -0400 (Wed, 13 Apr 2011) New Revision: 10501 Added: branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionAddMetaDataMessageV2.java Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/api/core/client/ClientSession.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/DelegatingSession.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/ServerSessionPacketHandler.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/PacketDecoder.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/PacketImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionAddMetaDataMessage.java branches/Branch_2_2_EAP/tests/jms-tests/src/org/hornetq/jms/tests/message/MessageHeaderTest.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadRandomReattachTestBase.java Log: https://issues.jboss.org/browse/JBPAPP-6300 - Fixing possible dead lock over Failover with metadata Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/api/core/client/ClientSession.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/api/core/client/ClientSession.java 2011-04-13 15:30:27 UTC (rev 10500) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/api/core/client/ClientSession.java 2011-04-13 18:00:12 UTC (rev 10501) @@ -567,4 +567,12 @@ * @throws HornetQException */ void addMetaData(String key, String data) throws HornetQException; + + /** + * Attach any metadata to the session. + * Sends a Metadata using the older version + * @deprecated Use {@link ClientSession#addMetaData(String, String)} + * @throws HornetQException + */ + void addMetaDataV1(String key, String data) throws HornetQException; } Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2011-04-13 15:30:27 UTC (rev 10500) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2011-04-13 18:00:12 UTC (rev 10501) @@ -452,6 +452,11 @@ { stopPingingAfterOne = true; } + + public void resumePinging() + { + stopPingingAfterOne = false; + } // Protected // ------------------------------------------------------------------------------------ Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2011-04-13 15:30:27 UTC (rev 10500) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2011-04-13 18:00:12 UTC (rev 10501) @@ -47,6 +47,7 @@ import org.hornetq.core.protocol.core.impl.wireformat.RollbackMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionAcknowledgeMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionAddMetaDataMessage; +import org.hornetq.core.protocol.core.impl.wireformat.SessionAddMetaDataMessageV2; import org.hornetq.core.protocol.core.impl.wireformat.SessionBindingQueryMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionBindingQueryResponseMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionCloseMessage; @@ -1085,16 +1086,22 @@ // Resetting the metadata after failover for (Map.Entry<String, String> entries : metadata.entrySet()) { - sendPacketWithoutLock(new SessionAddMetaDataMessage(entries.getKey(), entries.getValue())); + sendPacketWithoutLock(new SessionAddMetaDataMessageV2(entries.getKey(), entries.getValue(), false)); } } - public void addMetaData(String key, String data) throws HornetQException + public void addMetaDataV1(String key, String data) throws HornetQException { metadata.put(key, data); channel.sendBlocking(new SessionAddMetaDataMessage(key, data)); } + public void addMetaData(String key, String data) throws HornetQException + { + metadata.put(key, data); + channel.send(new SessionAddMetaDataMessageV2(key, data)); + } + public ClientSessionFactoryInternal getSessionFactory() { return sessionFactory; Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/DelegatingSession.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/DelegatingSession.java 2011-04-13 15:30:27 UTC (rev 10500) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/DelegatingSession.java 2011-04-13 18:00:12 UTC (rev 10501) @@ -566,7 +566,14 @@ { session.addMetaData(key, data); } + + @Deprecated + public void addMetaDataV1(String key, String data) throws HornetQException + { + session.addMetaDataV1(key, data); + } + public boolean isCompressLargeMessages() { return session.isCompressLargeMessages(); Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/ServerSessionPacketHandler.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/ServerSessionPacketHandler.java 2011-04-13 15:30:27 UTC (rev 10500) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/ServerSessionPacketHandler.java 2011-04-13 18:00:12 UTC (rev 10501) @@ -63,6 +63,7 @@ import org.hornetq.core.protocol.core.impl.wireformat.RollbackMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionAcknowledgeMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionAddMetaDataMessage; +import org.hornetq.core.protocol.core.impl.wireformat.SessionAddMetaDataMessageV2; import org.hornetq.core.protocol.core.impl.wireformat.SessionBindingQueryMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionBindingQueryResponseMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionConsumerCloseMessage; @@ -479,6 +480,12 @@ session.addMetaData(message.getKey(), message.getData()); break; } + case PacketImpl.SESS_ADD_METADATA2: + { + SessionAddMetaDataMessageV2 message = (SessionAddMetaDataMessageV2)packet; + session.addMetaData(message.getKey(), message.getData()); + break; + } } } catch (HornetQXAException e) Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/PacketDecoder.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/PacketDecoder.java 2011-04-13 15:30:27 UTC (rev 10500) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/PacketDecoder.java 2011-04-13 18:00:12 UTC (rev 10501) @@ -41,6 +41,8 @@ import static org.hornetq.core.protocol.core.impl.PacketImpl.REPLICATION_PREPARE; import static org.hornetq.core.protocol.core.impl.PacketImpl.REPLICATION_RESPONSE; import static org.hornetq.core.protocol.core.impl.PacketImpl.SESS_ACKNOWLEDGE; +import static org.hornetq.core.protocol.core.impl.PacketImpl.SESS_ADD_METADATA; +import static org.hornetq.core.protocol.core.impl.PacketImpl.SESS_ADD_METADATA2; import static org.hornetq.core.protocol.core.impl.PacketImpl.SESS_BINDINGQUERY; import static org.hornetq.core.protocol.core.impl.PacketImpl.SESS_BINDINGQUERY_RESP; import static org.hornetq.core.protocol.core.impl.PacketImpl.SESS_CLOSE; @@ -81,12 +83,9 @@ import static org.hornetq.core.protocol.core.impl.PacketImpl.SESS_XA_START; import static org.hornetq.core.protocol.core.impl.PacketImpl.SESS_XA_SUSPEND; import static org.hornetq.core.protocol.core.impl.PacketImpl.SUBSCRIBE_TOPOLOGY; -import static org.hornetq.core.protocol.core.impl.PacketImpl.SESS_ADD_METADATA; import org.hornetq.api.core.HornetQBuffer; -import org.hornetq.core.client.impl.ClientMessageImpl; import org.hornetq.core.logging.Logger; -import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.protocol.core.Packet; import org.hornetq.core.protocol.core.impl.wireformat.ClusterTopologyChangeMessage; import org.hornetq.core.protocol.core.impl.wireformat.CreateQueueMessage; @@ -117,6 +116,7 @@ import org.hornetq.core.protocol.core.impl.wireformat.RollbackMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionAcknowledgeMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionAddMetaDataMessage; +import org.hornetq.core.protocol.core.impl.wireformat.SessionAddMetaDataMessageV2; import org.hornetq.core.protocol.core.impl.wireformat.SessionBindingQueryMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionBindingQueryResponseMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionCloseMessage; @@ -519,6 +519,11 @@ packet = new SessionAddMetaDataMessage(); break; } + case SESS_ADD_METADATA2: + { + packet = new SessionAddMetaDataMessageV2(); + break; + } default: { throw new IllegalArgumentException("Invalid type: " + packetType); Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/PacketImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/PacketImpl.java 2011-04-13 15:30:27 UTC (rev 10500) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/PacketImpl.java 2011-04-13 18:00:12 UTC (rev 10501) @@ -186,6 +186,8 @@ public static final byte SESS_ADD_METADATA = 104; + public static final byte SESS_ADD_METADATA2 = 105; + public static final byte CLUSTER_TOPOLOGY = 110; public static final byte NODE_ANNOUNCE = 111; Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionAddMetaDataMessage.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionAddMetaDataMessage.java 2011-04-13 15:30:27 UTC (rev 10500) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionAddMetaDataMessage.java 2011-04-13 18:00:12 UTC (rev 10501) @@ -18,6 +18,8 @@ /** * A SessionAddMetaDataMessage + * + * Packet deprecated: It exists only to support old formats * * @author <a href="mailto:hgao@redhat.com>Howard Gao</a> * Added: branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionAddMetaDataMessageV2.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionAddMetaDataMessageV2.java (rev 0) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionAddMetaDataMessageV2.java 2011-04-13 18:00:12 UTC (rev 10501) @@ -0,0 +1,91 @@ +/* + * 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.core.protocol.core.impl.wireformat; + +import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.core.protocol.core.impl.PacketImpl; + +/** + * A SessionAddMetaDataMessage + * + * This packet replaces {@link SessionAddMetaDataMessage} + * + * @author Clebert Suconic</a> + * + * + */ +public class SessionAddMetaDataMessageV2 extends PacketImpl +{ + private String key; + private String data; + /** + * It won require confirmation during failover / reconnect + */ + private boolean requiresConfirmation = true; + + public SessionAddMetaDataMessageV2() + { + super(PacketImpl.SESS_ADD_METADATA2); + } + + public SessionAddMetaDataMessageV2(String k, String d) + { + this(); + key = k; + data = d; + } + + public SessionAddMetaDataMessageV2(String k, String d, boolean requiresConfirmation) + { + this(); + key = k; + data = d; + this.requiresConfirmation = requiresConfirmation; + } + + @Override + public void encodeRest(final HornetQBuffer buffer) + { + buffer.writeString(key); + buffer.writeString(data); + buffer.writeBoolean(requiresConfirmation); + } + + @Override + public void decodeRest(final HornetQBuffer buffer) + { + key = buffer.readString(); + data = buffer.readString(); + requiresConfirmation = buffer.readBoolean(); + } + + @Override + public final boolean isRequiresConfirmations() + { + return requiresConfirmation; + } + + + + public String getKey() + { + return key; + } + + public String getData() + { + return data; + } + +} Modified: branches/Branch_2_2_EAP/tests/jms-tests/src/org/hornetq/jms/tests/message/MessageHeaderTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/jms-tests/src/org/hornetq/jms/tests/message/MessageHeaderTest.java 2011-04-13 15:30:27 UTC (rev 10500) +++ branches/Branch_2_2_EAP/tests/jms-tests/src/org/hornetq/jms/tests/message/MessageHeaderTest.java 2011-04-13 18:00:12 UTC (rev 10501) @@ -1408,6 +1408,15 @@ // TODO Auto-generated method stub } + + /* (non-Javadoc) + * @see org.hornetq.api.core.client.ClientSession#addMetaDataV1(java.lang.String, java.lang.String) + */ + public void addMetaDataV1(String key, String data) throws HornetQException + { + // TODO Auto-generated method stub + + } } } Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadRandomReattachTestBase.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadRandomReattachTestBase.java 2011-04-13 15:30:27 UTC (rev 10500) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadRandomReattachTestBase.java 2011-04-13 18:00:12 UTC (rev 10501) @@ -38,6 +38,7 @@ import org.hornetq.core.server.HornetQServer; import org.hornetq.jms.client.HornetQBytesMessage; import org.hornetq.jms.client.HornetQTextMessage; +import org.hornetq.tests.util.RandomUtil; /** * A MultiThreadRandomReattachTestBase @@ -264,12 +265,19 @@ protected ClientSession createAutoCommitSession(final ClientSessionFactory sf) throws Exception { - return sf.createSession(false, true, true); + ClientSession session = sf.createSession(false, true, true); + session.addMetaData("someData", RandomUtil.randomString()); + session.addMetaData("someData2", RandomUtil.randomString()); + return session; } protected ClientSession createTransactionalSession(final ClientSessionFactory sf) throws Exception { - return sf.createSession(false, false, false); + ClientSession session = sf.createSession(false, false, false); + session.addMetaData("someData", RandomUtil.randomString()); + session.addMetaData("someData2", RandomUtil.randomString()); + + return session; } protected void doTestA(final ClientSessionFactory sf, final int threadNum, final ClientSession session2) throws Exception @@ -432,6 +440,7 @@ } ClientSession sessSend = sf.createSession(false, true, true); + sessSend.addMetaData("some-data", RandomUtil.randomString()); ClientProducer producer = sessSend.createProducer(MultiThreadRandomReattachTestBase.ADDRESS); @@ -497,7 +506,9 @@ long start = System.currentTimeMillis(); ClientSession s = sf.createSession(false, false, false); + s.addMetaData("some-data", RandomUtil.randomString()); + final int numMessages = 100; final int numSessions = 10; @@ -523,7 +534,9 @@ } ClientSession sessSend = sf.createSession(false, false, false); + sessSend.addMetaData("some-data", RandomUtil.randomString()); + ClientProducer producer = sessSend.createProducer(MultiThreadRandomReattachTestBase.ADDRESS); sendMessages(sessSend, producer, numMessages, threadNum); @@ -606,7 +619,9 @@ long start = System.currentTimeMillis(); ClientSession s = sf.createSession(false, false, false); + s.addMetaData("some-data", RandomUtil.randomString()); + final int numMessages = 100; final int numSessions = 10; @@ -619,6 +634,7 @@ SimpleString subName = new SimpleString(threadNum + " sub" + i); ClientSession sessConsume = sf.createSession(false, false, false); + sessConsume.addMetaData("data", RandomUtil.randomString()); sessConsume.createQueue(MultiThreadRandomReattachTestBase.ADDRESS, subName, null, false); @@ -630,7 +646,9 @@ } ClientSession sessSend = sf.createSession(false, false, false); + sessSend.addMetaData("some-data", RandomUtil.randomString()); + ClientProducer producer = sessSend.createProducer(MultiThreadRandomReattachTestBase.ADDRESS); sendMessages(sessSend, producer, numMessages, threadNum); @@ -746,7 +764,9 @@ long start = System.currentTimeMillis(); ClientSession s = sf.createSession(false, false, false); + s.addMetaData("some-data", RandomUtil.randomString()); + final int numMessages = 100; final int numSessions = 10; @@ -759,7 +779,9 @@ SimpleString subName = new SimpleString(threadNum + "sub" + i); ClientSession sessConsume = sf.createSession(false, true, true); + sessConsume.addMetaData("some-data", RandomUtil.randomString()); + sessConsume.start(); sessConsume.createQueue(MultiThreadRandomReattachTestBase.ADDRESS, subName, null, false); @@ -772,6 +794,7 @@ } ClientSession sessSend = sf.createSession(false, true, true); + sessSend.addMetaData("some-data", RandomUtil.randomString()); ClientProducer producer = sessSend.createProducer(MultiThreadRandomReattachTestBase.ADDRESS); @@ -804,6 +827,8 @@ long start = System.currentTimeMillis(); ClientSession s = sf.createSession(false, false, false); + s.addMetaData("data", RandomUtil.randomString()); + final int numMessages = 100; @@ -817,6 +842,7 @@ SimpleString subName = new SimpleString(threadNum + "sub" + i); ClientSession sessConsume = sf.createSession(false, true, true); + sessConsume.addMetaData("data", RandomUtil.randomString()); sessConsume.createQueue(MultiThreadRandomReattachTestBase.ADDRESS, subName, null, false); @@ -828,6 +854,7 @@ } ClientSession sessSend = sf.createSession(false, true, true); + sessSend.addMetaData("data", RandomUtil.randomString()); ClientProducer producer = sessSend.createProducer(MultiThreadRandomReattachTestBase.ADDRESS); @@ -865,6 +892,7 @@ long start = System.currentTimeMillis(); ClientSession s = sf.createSession(false, false, false); + s.addMetaData("data", RandomUtil.randomString()); final int numMessages = 100; @@ -878,6 +906,8 @@ SimpleString subName = new SimpleString(threadNum + "sub" + i); ClientSession sessConsume = sf.createSession(false, false, false); + sessConsume.addMetaData("data", RandomUtil.randomString()); + sessConsume.start(); @@ -891,7 +921,9 @@ } ClientSession sessSend = sf.createSession(false, false, false); + sessSend.addMetaData("data", RandomUtil.randomString()); + ClientProducer producer = sessSend.createProducer(MultiThreadRandomReattachTestBase.ADDRESS); sendMessages(sessSend, producer, numMessages, threadNum); @@ -941,7 +973,9 @@ long start = System.currentTimeMillis(); ClientSession s = sf.createSession(false, false, false); + s.addMetaData("data", RandomUtil.randomString()); + final int numMessages = 100; final int numSessions = 10; @@ -954,7 +988,9 @@ SimpleString subName = new SimpleString(threadNum + "sub" + i); ClientSession sessConsume = sf.createSession(false, false, false); + sessConsume.addMetaData("data", RandomUtil.randomString()); + sessConsume.createQueue(MultiThreadRandomReattachTestBase.ADDRESS, subName, null, false); ClientConsumer consumer = sessConsume.createConsumer(subName); @@ -965,6 +1001,7 @@ } ClientSession sessSend = sf.createSession(false, false, false); + sessSend.addMetaData("data", RandomUtil.randomString()); ClientProducer producer = sessSend.createProducer(MultiThreadRandomReattachTestBase.ADDRESS); @@ -1018,14 +1055,18 @@ protected void doTestI(final ClientSessionFactory sf, final int threadNum) throws Exception { ClientSession sessCreate = sf.createSession(false, true, true); + sessCreate.addMetaData("data", RandomUtil.randomString()); + sessCreate.createQueue(MultiThreadRandomReattachTestBase.ADDRESS, new SimpleString(threadNum + MultiThreadRandomReattachTestBase.ADDRESS.toString()), null, false); ClientSession sess = sf.createSession(false, true, true); + sess.addMetaData("data", RandomUtil.randomString()); + sess.start(); ClientConsumer consumer = sess.createConsumer(new SimpleString(threadNum + MultiThreadRandomReattachTestBase.ADDRESS.toString())); @@ -1051,13 +1092,16 @@ protected void doTestJ(final ClientSessionFactory sf, final int threadNum) throws Exception { ClientSession sessCreate = sf.createSession(false, true, true); + sessCreate.addMetaData("data", RandomUtil.randomString()); + sessCreate.createQueue(MultiThreadRandomReattachTestBase.ADDRESS, new SimpleString(threadNum + MultiThreadRandomReattachTestBase.ADDRESS.toString()), null, false); ClientSession sess = sf.createSession(false, true, true); + sess.addMetaData("data", RandomUtil.randomString()); sess.start(); @@ -1084,6 +1128,7 @@ protected void doTestK(final ClientSessionFactory sf, final int threadNum) throws Exception { ClientSession s = sf.createSession(false, false, false); + s.addMetaData("data", RandomUtil.randomString()); s.createQueue(MultiThreadRandomReattachTestBase.ADDRESS, new SimpleString(threadNum + MultiThreadRandomReattachTestBase.ADDRESS.toString()), @@ -1114,6 +1159,8 @@ for (int i = 0; i < numSessions; i++) { ClientSession session = sf.createSession(false, false, false); + + session.addMetaData("data", RandomUtil.randomString()); session.close(); } @@ -1129,6 +1176,7 @@ false); ClientSession sess = sf.createSession(false, true, true); + sess.addMetaData("data", RandomUtil.randomString()); sess.stop();