[hornetq-commits] JBoss hornetq SVN: r8811 - in trunk: src/main/org/hornetq/core/client/impl and 7 other directories.
do-not-reply at jboss.org
do-not-reply at jboss.org
Wed Jan 20 09:38:53 EST 2010
Author: timfox
Date: 2010-01-20 09:38:53 -0500 (Wed, 20 Jan 2010)
New Revision: 8811
Added:
trunk/src/main/org/hornetq/core/server/BindingQueryResult.java
trunk/src/main/org/hornetq/core/server/QueueQueryResult.java
trunk/src/main/org/hornetq/core/server/SessionCallback.java
Modified:
trunk/src/main/org/hornetq/api/core/client/ClientSession.java
trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java
trunk/src/main/org/hornetq/core/remoting/Channel.java
trunk/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java
trunk/src/main/org/hornetq/core/remoting/impl/wireformat/SessionQueueQueryResponseMessage.java
trunk/src/main/org/hornetq/core/server/ServerSession.java
trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java
trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java
trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java
trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java
trunk/src/main/org/hornetq/core/server/impl/ServerSessionPacketHandler.java
trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java
trunk/tests/src/org/hornetq/tests/integration/xa/BasicXaTest.java
Log:
partial refactoring to remove remoting knowledge from session
Modified: trunk/src/main/org/hornetq/api/core/client/ClientSession.java
===================================================================
--- trunk/src/main/org/hornetq/api/core/client/ClientSession.java 2010-01-20 13:24:55 UTC (rev 8810)
+++ trunk/src/main/org/hornetq/api/core/client/ClientSession.java 2010-01-20 14:38:53 UTC (rev 8811)
@@ -455,6 +455,7 @@
/**
* Creates a ClientMessage.
*
+ * @param type type of the message
* @param durable whether the created message is durable or not
* @param expiration the message expiration
* @param timestamp the message timestamp
Modified: trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2010-01-20 13:24:55 UTC (rev 8810)
+++ trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2010-01-20 14:38:53 UTC (rev 8811)
@@ -866,7 +866,7 @@
{
// The session was found on the server - we reattached transparently ok
- channel.replayCommands(response.getLastConfirmedCommandID(), channel.getID());
+ channel.replayCommands(response.getLastConfirmedCommandID());
}
else
{
Modified: trunk/src/main/org/hornetq/core/remoting/Channel.java
===================================================================
--- trunk/src/main/org/hornetq/core/remoting/Channel.java 2010-01-20 13:24:55 UTC (rev 8810)
+++ trunk/src/main/org/hornetq/core/remoting/Channel.java 2010-01-20 14:38:53 UTC (rev 8811)
@@ -89,9 +89,8 @@
* Typically called after a connection has been transferred.
*
* @param lastConfirmedCommandID the last confirmed packet
- * @param newID the new id to use
*/
- void replayCommands(int lastConfirmedCommandID, final long newID);
+ void replayCommands(int lastConfirmedCommandID);
/**
* returns the last confirmed packet command id
Modified: trunk/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java 2010-01-20 13:24:55 UTC (rev 8810)
+++ trunk/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java 2010-01-20 14:38:53 UTC (rev 8811)
@@ -336,7 +336,7 @@
}
}
- public void replayCommands(final int otherLastConfirmedCommandID, final long newChannelID)
+ public void replayCommands(final int otherLastConfirmedCommandID)
{
if (resendCache != null)
{
@@ -344,8 +344,6 @@
for (final Packet packet : resendCache)
{
- packet.setChannelID(newChannelID);
-
doWrite(packet);
}
}
Modified: trunk/src/main/org/hornetq/core/remoting/impl/wireformat/SessionQueueQueryResponseMessage.java
===================================================================
--- trunk/src/main/org/hornetq/core/remoting/impl/wireformat/SessionQueueQueryResponseMessage.java 2010-01-20 13:24:55 UTC (rev 8810)
+++ trunk/src/main/org/hornetq/core/remoting/impl/wireformat/SessionQueueQueryResponseMessage.java 2010-01-20 14:38:53 UTC (rev 8811)
@@ -15,6 +15,7 @@
import org.hornetq.api.core.HornetQBuffer;
import org.hornetq.api.core.SimpleString;
+import org.hornetq.core.server.QueueQueryResult;
/**
*
@@ -41,15 +42,10 @@
private boolean temporary;
- public SessionQueueQueryResponseMessage(final SimpleString name,
- final SimpleString address,
- final boolean durable,
- final boolean temporary,
- final SimpleString filterString,
- final int consumerCount,
- final int messageCount)
+ public SessionQueueQueryResponseMessage(final QueueQueryResult result)
{
- this(name, address, durable, temporary, filterString, consumerCount, messageCount, true);
+ this(result.getName(), result.getAddress(), result.isDurable(), result.isTemporary(),
+ result.getFilterString(), result.getConsumerCount(), result.getMessageCount(), result.isExists());
}
public SessionQueueQueryResponseMessage()
Added: trunk/src/main/org/hornetq/core/server/BindingQueryResult.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/BindingQueryResult.java (rev 0)
+++ trunk/src/main/org/hornetq/core/server/BindingQueryResult.java 2010-01-20 14:38:53 UTC (rev 8811)
@@ -0,0 +1,49 @@
+/*
+ * Copyright 2009 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.server;
+
+import java.util.List;
+
+import org.hornetq.api.core.SimpleString;
+
+/**
+ *
+ * A BindingQueryResult
+ *
+ * @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
+ *
+ */
+public class BindingQueryResult
+{
+ private boolean exists;
+
+ private List<SimpleString> queueNames;
+
+ public BindingQueryResult(final boolean exists, final List<SimpleString> queueNames)
+ {
+ this.exists = exists;
+
+ this.queueNames = queueNames;
+ }
+
+ public boolean isExists()
+ {
+ return exists;
+ }
+
+ public List<SimpleString> getQueueNames()
+ {
+ return queueNames;
+ }
+}
Added: trunk/src/main/org/hornetq/core/server/QueueQueryResult.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/QueueQueryResult.java (rev 0)
+++ trunk/src/main/org/hornetq/core/server/QueueQueryResult.java 2010-01-20 14:38:53 UTC (rev 8811)
@@ -0,0 +1,125 @@
+/*
+ * Copyright 2009 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.server;
+
+import org.hornetq.api.core.SimpleString;
+
+/**
+ *
+ * A QueueQueryResult
+ *
+ * @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
+ *
+ */
+public class QueueQueryResult
+{
+ private SimpleString name;
+
+ private boolean exists;
+
+ private boolean durable;
+
+ private int consumerCount;
+
+ private int messageCount;
+
+ private SimpleString filterString;
+
+ private SimpleString address;
+
+ private boolean temporary;
+
+ public QueueQueryResult(final SimpleString name,
+ final SimpleString address,
+ final boolean durable,
+ final boolean temporary,
+ final SimpleString filterString,
+ final int consumerCount,
+ final int messageCount)
+ {
+ this(name, address, durable, temporary, filterString, consumerCount, messageCount, true);
+ }
+
+ public QueueQueryResult()
+ {
+ this(null, null, false, false, null, 0, 0, false);
+ }
+
+ private QueueQueryResult(final SimpleString name,
+ final SimpleString address,
+ final boolean durable,
+ final boolean temporary,
+ final SimpleString filterString,
+ final int consumerCount,
+ final int messageCount,
+ final boolean exists)
+ {
+ this.durable = durable;
+
+ this.temporary = temporary;
+
+ this.consumerCount = consumerCount;
+
+ this.messageCount = messageCount;
+
+ this.filterString = filterString;
+
+ this.address = address;
+
+ this.name = name;
+
+ this.exists = exists;
+ }
+
+ public boolean isExists()
+ {
+ return exists;
+ }
+
+ public boolean isDurable()
+ {
+ return durable;
+ }
+
+ public int getConsumerCount()
+ {
+ return consumerCount;
+ }
+
+ public int getMessageCount()
+ {
+ return messageCount;
+ }
+
+ public SimpleString getFilterString()
+ {
+ return filterString;
+ }
+
+ public SimpleString getAddress()
+ {
+ return address;
+ }
+
+ public SimpleString getName()
+ {
+ return name;
+ }
+
+ public boolean isTemporary()
+ {
+ return temporary;
+ }
+
+}
Modified: trunk/src/main/org/hornetq/core/server/ServerSession.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/ServerSession.java 2010-01-20 13:24:55 UTC (rev 8810)
+++ trunk/src/main/org/hornetq/core/server/ServerSession.java 2010-01-20 14:38:53 UTC (rev 8811)
@@ -13,34 +13,13 @@
package org.hornetq.core.server;
+import java.util.List;
+
+import javax.transaction.xa.Xid;
+
+import org.hornetq.api.core.SimpleString;
import org.hornetq.core.remoting.Channel;
-import org.hornetq.core.remoting.Packet;
import org.hornetq.core.remoting.RemotingConnection;
-import org.hornetq.core.remoting.impl.wireformat.CreateQueueMessage;
-import org.hornetq.core.remoting.impl.wireformat.RollbackMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionAcknowledgeMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionConsumerCloseMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionConsumerFlowCreditMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionCreateConsumerMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionDeleteQueueMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionExpiredMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionForceConsumerDelivery;
-import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionRequestProducerCreditsMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionSendContinuationMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionSendLargeMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionSendMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXACommitMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXAEndMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXAForgetMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXAJoinMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXAPrepareMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXAResumeMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXARollbackMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXASetTimeoutMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXAStartMessage;
-import org.hornetq.core.server.impl.ServerSessionPacketHandler;
/**
*
@@ -54,8 +33,6 @@
{
String getName();
- long getID();
-
String getUsername();
String getPassword();
@@ -68,73 +45,79 @@
void close() throws Exception;
- void handleAcknowledge(final SessionAcknowledgeMessage packet);
+ void handleAcknowledge(long consumerID, long messageID) throws Exception;
- void handleExpired(final SessionExpiredMessage packet);
+ void handleExpired(long consumerID, long messageID) throws Exception;
- void handleRollback(RollbackMessage packet);
+ void handleRollback(boolean considerLastMessageAsDelivered) throws Exception;
- void handleCommit(Packet packet);
+ void handleCommit() throws Exception;
- void handleXACommit(SessionXACommitMessage packet);
+ void handleXACommit(Xid xid, boolean onePhase) throws Exception;
- void handleXAEnd(SessionXAEndMessage packet);
+ void handleXAEnd(Xid xid) throws Exception;
- void handleXAForget(SessionXAForgetMessage packet);
+ void handleXAForget(Xid xid) throws Exception;
- void handleXAJoin(SessionXAJoinMessage packet);
+ void handleXAJoin(Xid xid) throws Exception;
- void handleXAPrepare(SessionXAPrepareMessage packet);
+ void handleXAPrepare(Xid xid) throws Exception;
- void handleXAResume(SessionXAResumeMessage packet);
+ void handleXAResume(Xid xid) throws Exception;
- void handleXARollback(SessionXARollbackMessage packet);
+ void handleXARollback(Xid xid) throws Exception;
- void handleXAStart(SessionXAStartMessage packet);
+ void handleXAStart(Xid xid) throws Exception;
- void handleXASuspend(Packet packet);
+ void handleXASuspend() throws Exception;
- void handleGetInDoubtXids(Packet packet);
+ List<Xid> handleGetInDoubtXids();
- void handleGetXATimeout(Packet packet);
+ int handleGetXATimeout();
- void handleSetXATimeout(SessionXASetTimeoutMessage packet);
+ void handleSetXATimeout(int timeout);
- void handleStart(Packet packet);
+ void handleStart();
- void handleStop(Packet packet);
+ void handleStop();
- void handleCreateQueue(CreateQueueMessage packet);
+ void handleCreateQueue(SimpleString address,
+ SimpleString name,
+ SimpleString filterString,
+ boolean temporary,
+ boolean durable) throws Exception;
- void handleDeleteQueue(SessionDeleteQueueMessage packet);
+ void handleDeleteQueue(SimpleString name) throws Exception;
- void handleCreateConsumer(SessionCreateConsumerMessage packet);
+ void handleCreateConsumer(long consumerID, SimpleString name, SimpleString filterString, boolean browseOnly) throws Exception;
- void handleExecuteQueueQuery(SessionQueueQueryMessage packet);
+ QueueQueryResult handleExecuteQueueQuery(SimpleString name) throws Exception;
- void handleExecuteBindingQuery(SessionBindingQueryMessage packet);
+ BindingQueryResult handleExecuteBindingQuery(SimpleString address);
- void handleCloseConsumer(SessionConsumerCloseMessage packet);
+ void handleCloseConsumer(long consumerID) throws Exception;
- void handleReceiveConsumerCredits(SessionConsumerFlowCreditMessage packet);
+ void handleReceiveConsumerCredits(long consumerID, int credits) throws Exception;
- void handleSendContinuations(SessionSendContinuationMessage packet);
+ void handleSendContinuations(int packetSize, byte[] body, boolean continues) throws Exception;
- void handleSend(SessionSendMessage packet);
+ void handleSend(ServerMessage message) throws Exception;
- void handleSendLargeMessage(SessionSendLargeMessage packet);
+ void handleSendLargeMessage(byte[] largeMessageHeader) throws Exception;
- void handleForceConsumerDelivery(SessionForceConsumerDelivery message);
+ void handleForceConsumerDelivery(long consumerID, long sequence) throws Exception;
- void handleRequestProducerCredits(SessionRequestProducerCreditsMessage message) throws Exception;
+ void handleRequestProducerCredits(SimpleString address, int credits) throws Exception;
- void handleClose(Packet packet);
+ void handleClose() throws Exception;
int transferConnection(RemotingConnection newConnection, int lastReceivedCommandID);
Channel getChannel();
- ServerSessionPacketHandler getHandler();
-
- void setHandler(ServerSessionPacketHandler handler);
+ void setTransferring(boolean transferring);
+
+ void runConnectionFailureRunners();
+
+ void setCallback(SessionCallback callback);
}
Added: trunk/src/main/org/hornetq/core/server/SessionCallback.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/SessionCallback.java (rev 0)
+++ trunk/src/main/org/hornetq/core/server/SessionCallback.java 2010-01-20 14:38:53 UTC (rev 8811)
@@ -0,0 +1,35 @@
+/*
+ * Copyright 2009 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.server;
+
+import org.hornetq.api.core.SimpleString;
+
+/**
+ * A SessionCallback
+ *
+ * @author Tim Fox
+ *
+ *
+ */
+public interface SessionCallback
+{
+ public void sendProducerCreditsMessage(int credits, SimpleString address, int offset);
+
+ public int sendMessage(ServerMessage message, long consumerID, int deliveryCount);
+
+ public int sendLargeMessage(long consumerID, byte[] headerBuffer, long bodySize, int deliveryCount);
+
+ public int sendLargeMessageContinuation(long consumerID, byte[] body, boolean continues, boolean requiresResponse);
+
+}
Modified: trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2010-01-20 13:24:55 UTC (rev 8810)
+++ trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2010-01-20 14:38:53 UTC (rev 8811)
@@ -22,7 +22,6 @@
import java.util.Set;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@@ -643,11 +642,11 @@
configuration.isPersistDeliveryCountBeforeDelivery(),
xa,
connection,
+ channel,
storageManager,
postOffice,
resourceManager,
- securityStore,
- channel,
+ securityStore,
managementService,
this,
configuration.getManagementAddress());
@@ -656,10 +655,11 @@
ServerSessionPacketHandler handler = new ServerSessionPacketHandler(session,
storageManager.newContext(executorFactory.getExecutor()),
- storageManager);
+ storageManager,
+ channel);
+
+ session.setCallback(handler);
- session.setHandler(handler);
-
channel.setHandler(handler);
return new CreateSessionResponseMessage(version.getIncrementingVersion());
Modified: trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java 2010-01-20 13:24:55 UTC (rev 8810)
+++ trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java 2010-01-20 14:38:53 UTC (rev 8811)
@@ -245,12 +245,12 @@
executor.execute(deliverRunner);
}
}
-
+
public Executor getExecutor()
{
return executor;
}
-
+
public synchronized void deliverNow()
{
deliverRunner.run();
@@ -743,7 +743,7 @@
}
return false;
}
-
+
public int sendMessagesToDeadLetterAddress(Filter filter) throws Exception
{
int count = 0;
@@ -757,7 +757,7 @@
deliveringCount.incrementAndGet();
sendToDeadLetterAddress(ref);
iter.remove();
- count ++;
+ count++;
}
}
return count;
@@ -832,7 +832,7 @@
return false;
}
-
+
public int changeReferencesPriority(final Filter filter, final byte newPriority) throws Exception
{
Iterator<MessageReference> iter = messageReferences.iterator();
@@ -843,7 +843,7 @@
MessageReference ref = iter.next();
if (filter == null || filter.match(ref.getMessage()))
{
- count ++;
+ count++;
iter.remove();
ref.getMessage().setPriority(newPriority);
addLast(ref);
@@ -1404,7 +1404,8 @@
}
catch (Exception e)
{
- QueueImpl.log.warn("Unable to remove message id = " + message.getMessageID() + " please remove manually");
+ QueueImpl.log.warn("Unable to remove message id = " + message.getMessageID() + " please remove manually",
+ e);
}
}
}
Modified: trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java 2010-01-20 13:24:55 UTC (rev 8810)
+++ trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java 2010-01-20 14:38:53 UTC (rev 8811)
@@ -23,7 +23,6 @@
import org.hornetq.api.core.HornetQBuffer;
import org.hornetq.api.core.HornetQBuffers;
-import org.hornetq.api.core.HornetQException;
import org.hornetq.api.core.management.ManagementHelper;
import org.hornetq.api.core.management.NotificationType;
import org.hornetq.core.client.impl.ClientConsumerImpl;
@@ -33,10 +32,6 @@
import org.hornetq.core.persistence.StorageManager;
import org.hornetq.core.postoffice.Binding;
import org.hornetq.core.postoffice.QueueBinding;
-import org.hornetq.core.remoting.Channel;
-import org.hornetq.core.remoting.impl.wireformat.SessionReceiveContinuationMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionReceiveLargeMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionReceiveMessage;
import org.hornetq.core.server.HandleStatus;
import org.hornetq.core.server.LargeServerMessage;
import org.hornetq.core.server.MessageReference;
@@ -44,6 +39,7 @@
import org.hornetq.core.server.ServerConsumer;
import org.hornetq.core.server.ServerMessage;
import org.hornetq.core.server.ServerSession;
+import org.hornetq.core.server.SessionCallback;
import org.hornetq.core.server.management.ManagementService;
import org.hornetq.core.server.management.Notification;
import org.hornetq.core.transaction.Transaction;
@@ -112,7 +108,7 @@
private final java.util.Queue<MessageReference> deliveringRefs = new ConcurrentLinkedQueue<MessageReference>();
- private final Channel channel;
+ private final SessionCallback callback;
private volatile boolean closed;
@@ -133,12 +129,11 @@
final boolean started,
final boolean browseOnly,
final StorageManager storageManager,
- final Channel channel,
+ final SessionCallback callback,
final boolean preAcknowledge,
final boolean strictUpdateDeliveryCount,
final ManagementService managementService) throws Exception
{
-
this.id = id;
this.filter = filter;
@@ -157,7 +152,7 @@
this.storageManager = storageManager;
- this.channel = channel;
+ this.callback = callback;
this.preAcknowledge = preAcknowledge;
@@ -361,9 +356,7 @@
forcedDeliveryMessage.putLongProperty(ClientConsumerImpl.FORCED_DELIVERY_MESSAGE, sequence);
forcedDeliveryMessage.setAddress(messageQueue.getName());
- final SessionReceiveMessage packet = new SessionReceiveMessage(id, forcedDeliveryMessage, 0);
-
- channel.send(packet);
+ callback.sendMessage(forcedDeliveryMessage, id, 0);
}
catch (Exception e)
{
@@ -634,15 +627,12 @@
*/
private void deliverStandardMessage(final MessageReference ref, final ServerMessage message)
{
- final SessionReceiveMessage packet = new SessionReceiveMessage(id, message, ref.getDeliveryCount());
+ int packetSize = callback.sendMessage(message, id, ref.getDeliveryCount());
- channel.send(packet);
-
if (availableCredits != null)
{
- availableCredits.addAndGet(-packet.getPacketSize());
+ availableCredits.addAndGet(-packetSize);
}
-
}
// Inner classes
@@ -732,21 +722,18 @@
sizePendingLargeMessage = context.getLargeBodySize();
- SessionReceiveLargeMessage initialPacket = new SessionReceiveLargeMessage(id,
- headerBuffer.toByteBuffer()
- .array(),
- context.getLargeBodySize(),
- ref.getDeliveryCount());
-
context.open();
sentInitialPacket = true;
- channel.send(initialPacket);
+ int packetSize = callback.sendLargeMessage(id,
+ headerBuffer.toByteBuffer().array(),
+ context.getLargeBodySize(),
+ ref.getDeliveryCount());
if (availableCredits != null)
{
- availableCredits.addAndGet(-initialPacket.getPacketSize());
+ availableCredits.addAndGet(-packetSize);
}
// Execute the rest of the large message on a different thread so as not to tie up the delivery thread
@@ -768,22 +755,33 @@
return false;
}
- SessionReceiveContinuationMessage chunk = createChunkSend(context);
+ int localChunkLen = 0;
- int chunkLen = chunk.getBody().length;
+ localChunkLen = (int)Math.min(sizePendingLargeMessage - positionPendingLargeMessage, minLargeMessageSize);
- channel.send(chunk);
+ HornetQBuffer bodyBuffer = HornetQBuffers.fixedBuffer(localChunkLen);
+ context.encode(bodyBuffer, localChunkLen);
+
+ byte[] body = bodyBuffer.toByteBuffer().array();
+
+ int packetSize = callback.sendLargeMessageContinuation(id,
+ body,
+ positionPendingLargeMessage + localChunkLen < sizePendingLargeMessage,
+ false);
+
+ int chunkLen = body.length;
+
if (ServerConsumerImpl.trace)
{
- ServerConsumerImpl.trace("deliverLargeMessage: Sending " + chunk.getPacketSize() +
+ ServerConsumerImpl.trace("deliverLargeMessage: Sending " + packetSize +
" availableCredits now is " +
availableCredits);
}
if (availableCredits != null)
{
- availableCredits.addAndGet(-chunk.getPacketSize());
+ availableCredits.addAndGet(-packetSize);
}
positionPendingLargeMessage += chunkLen;
@@ -846,26 +844,6 @@
lock.unlock();
}
}
-
- private SessionReceiveContinuationMessage createChunkSend(final BodyEncoder context) throws HornetQException
- {
- SessionReceiveContinuationMessage chunk;
-
- int localChunkLen = 0;
-
- localChunkLen = (int)Math.min(sizePendingLargeMessage - positionPendingLargeMessage, minLargeMessageSize);
-
- HornetQBuffer bodyBuffer = HornetQBuffers.fixedBuffer(localChunkLen);
-
- context.encode(bodyBuffer, localChunkLen);
-
- chunk = new SessionReceiveContinuationMessage(id,
- bodyBuffer.toByteBuffer().array(),
- positionPendingLargeMessage + localChunkLen < sizePendingLargeMessage,
- false);
-
- return chunk;
- }
}
private class BrowserDeliverer implements Runnable
Modified: trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2010-01-20 13:24:55 UTC (rev 8810)
+++ trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2010-01-20 14:38:53 UTC (rev 8811)
@@ -24,13 +24,13 @@
import java.util.concurrent.ConcurrentHashMap;
import javax.transaction.xa.XAException;
-import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;
import org.hornetq.api.core.HornetQException;
import org.hornetq.api.core.SimpleString;
import org.hornetq.api.core.management.ManagementHelper;
import org.hornetq.core.client.impl.ClientMessageImpl;
+import org.hornetq.core.exception.HornetQXAException;
import org.hornetq.core.filter.Filter;
import org.hornetq.core.filter.impl.FilterImpl;
import org.hornetq.core.journal.IOAsyncTask;
@@ -45,52 +45,20 @@
import org.hornetq.core.remoting.Channel;
import org.hornetq.core.remoting.CloseListener;
import org.hornetq.core.remoting.FailureListener;
-import org.hornetq.core.remoting.Packet;
import org.hornetq.core.remoting.RemotingConnection;
-import org.hornetq.core.remoting.impl.wireformat.CreateQueueMessage;
-import org.hornetq.core.remoting.impl.wireformat.HornetQExceptionMessage;
-import org.hornetq.core.remoting.impl.wireformat.NullResponseMessage;
-import org.hornetq.core.remoting.impl.wireformat.PacketImpl;
-import org.hornetq.core.remoting.impl.wireformat.RollbackMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionAcknowledgeMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionConsumerCloseMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionConsumerFlowCreditMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionCreateConsumerMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionDeleteQueueMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionExpiredMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionForceConsumerDelivery;
-import org.hornetq.core.remoting.impl.wireformat.SessionProducerCreditsMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionRequestProducerCreditsMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionSendContinuationMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionSendLargeMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionSendMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXACommitMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXAEndMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXAForgetMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXAGetInDoubtXidsResponseMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXAGetTimeoutResponseMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXAJoinMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXAPrepareMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXAResponseMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXAResumeMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXARollbackMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXASetTimeoutMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXASetTimeoutResponseMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionXAStartMessage;
import org.hornetq.core.security.CheckType;
import org.hornetq.core.security.SecurityStore;
+import org.hornetq.core.server.BindingQueryResult;
import org.hornetq.core.server.HornetQServer;
import org.hornetq.core.server.LargeServerMessage;
import org.hornetq.core.server.MessageReference;
import org.hornetq.core.server.Queue;
+import org.hornetq.core.server.QueueQueryResult;
import org.hornetq.core.server.RoutingContext;
import org.hornetq.core.server.ServerConsumer;
import org.hornetq.core.server.ServerMessage;
import org.hornetq.core.server.ServerSession;
+import org.hornetq.core.server.SessionCallback;
import org.hornetq.core.server.management.ManagementService;
import org.hornetq.core.server.management.Notification;
import org.hornetq.core.transaction.ResourceManager;
@@ -113,10 +81,10 @@
private static final Logger log = Logger.getLogger(ServerSessionImpl.class);
// Static -------------------------------------------------------------------------------
-
+
// Attributes ----------------------------------------------------------------------------
- private final long id;
+ //private final long id;
private final String username;
@@ -133,6 +101,8 @@
private final boolean strictUpdateDeliveryCount;
private RemotingConnection remotingConnection;
+
+ private Channel channel;
private final Map<Long, ServerConsumer> consumers = new ConcurrentHashMap<Long, ServerConsumer>();
@@ -146,8 +116,6 @@
private final SecurityStore securityStore;
- private final Channel channel;
-
private final ManagementService managementService;
private volatile boolean started = false;
@@ -163,13 +131,15 @@
// The current currentLargeMessage being processed
private volatile LargeServerMessage currentLargeMessage;
- private ServerSessionPacketHandler handler;
+ // private ServerSessionPacketHandler handler;
private boolean closed;
private final Map<SimpleString, CreditManagerHolder> creditManagerHolders = new HashMap<SimpleString, CreditManagerHolder>();
private final RoutingContext routingContext = new RoutingContextImpl(null);
+
+ private SessionCallback callback;
// Constructors ---------------------------------------------------------------------------------
@@ -183,17 +153,15 @@
final boolean strictUpdateDeliveryCount,
final boolean xa,
final RemotingConnection remotingConnection,
+ final Channel channel,
final StorageManager storageManager,
final PostOffice postOffice,
final ResourceManager resourceManager,
- final SecurityStore securityStore,
- final Channel channel,
+ final SecurityStore securityStore,
final ManagementService managementService,
final HornetQServer server,
final SimpleString managementAddress) throws Exception
{
- id = channel.getID();
-
this.username = username;
this.password = password;
@@ -207,6 +175,8 @@
this.preAcknowledge = preAcknowledge;
this.remotingConnection = remotingConnection;
+
+ this.channel = channel;
this.storageManager = storageManager;
@@ -223,8 +193,6 @@
this.strictUpdateDeliveryCount = strictUpdateDeliveryCount;
- this.channel = channel;
-
this.managementService = managementService;
this.name = name;
@@ -232,7 +200,7 @@
this.server = server;
this.managementAddress = managementAddress;
-
+
remotingConnection.addFailureListener(this);
remotingConnection.addCloseListener(this);
@@ -240,16 +208,11 @@
// ServerSession implementation ----------------------------------------------------------------------------
- public ServerSessionPacketHandler getHandler()
+ public void setCallback(final SessionCallback callback)
{
- return handler;
+ this.callback = callback;
}
-
- public void setHandler(final ServerSessionPacketHandler handler)
- {
- this.handler = handler;
- }
-
+
public String getUsername()
{
return username;
@@ -265,11 +228,6 @@
return minLargeMessageSize;
}
- public long getID()
- {
- return id;
- }
-
public String getName()
{
return name;
@@ -332,577 +290,333 @@
}
}
- public void handleCreateConsumer(final SessionCreateConsumerMessage packet)
+ public void handleCreateConsumer(final long consumerID,
+ final SimpleString name,
+ final SimpleString filterString,
+ final boolean browseOnly) throws Exception
{
- SimpleString name = packet.getQueueName();
-
- SimpleString filterString = packet.getFilterString();
+ Binding binding = postOffice.getBinding(name);
- boolean browseOnly = packet.isBrowseOnly();
-
- Packet response = null;
-
- try
+ if (binding == null || binding.getType() != BindingType.LOCAL_QUEUE)
{
- Binding binding = postOffice.getBinding(name);
+ throw new HornetQException(HornetQException.QUEUE_DOES_NOT_EXIST, "Queue " + name + " does not exist");
+ }
- if (binding == null || binding.getType() != BindingType.LOCAL_QUEUE)
- {
- throw new HornetQException(HornetQException.QUEUE_DOES_NOT_EXIST, "Queue " + name + " does not exist");
- }
+ securityStore.check(binding.getAddress(), CheckType.CONSUME, this);
- securityStore.check(binding.getAddress(), CheckType.CONSUME, this);
+ Filter filter = FilterImpl.createFilter(filterString);
- Filter filter = FilterImpl.createFilter(filterString);;
+ ServerConsumer consumer = new ServerConsumerImpl(consumerID,
+ this,
+ (QueueBinding)binding,
+ filter,
+ started,
+ browseOnly,
+ storageManager,
+ callback,
+ preAcknowledge,
+ strictUpdateDeliveryCount,
+ managementService);
- ServerConsumer consumer = new ServerConsumerImpl(packet.getID(),
- this,
- (QueueBinding)binding,
- filter,
- started,
- browseOnly,
- storageManager,
- channel,
- preAcknowledge,
- strictUpdateDeliveryCount,
- managementService);
+ consumers.put(consumer.getID(), consumer);
- consumers.put(consumer.getID(), consumer);
+ if (!browseOnly)
+ {
+ TypedProperties props = new TypedProperties();
- if (!browseOnly)
- {
- TypedProperties props = new TypedProperties();
+ props.putSimpleStringProperty(ManagementHelper.HDR_ADDRESS, binding.getAddress());
- props.putSimpleStringProperty(ManagementHelper.HDR_ADDRESS, binding.getAddress());
+ props.putSimpleStringProperty(ManagementHelper.HDR_CLUSTER_NAME, binding.getClusterName());
- props.putSimpleStringProperty(ManagementHelper.HDR_CLUSTER_NAME, binding.getClusterName());
+ props.putSimpleStringProperty(ManagementHelper.HDR_ROUTING_NAME, binding.getRoutingName());
- props.putSimpleStringProperty(ManagementHelper.HDR_ROUTING_NAME, binding.getRoutingName());
+ props.putIntProperty(ManagementHelper.HDR_DISTANCE, binding.getDistance());
- props.putIntProperty(ManagementHelper.HDR_DISTANCE, binding.getDistance());
+ Queue theQueue = (Queue)binding.getBindable();
- Queue theQueue = (Queue)binding.getBindable();
+ props.putIntProperty(ManagementHelper.HDR_CONSUMER_COUNT, theQueue.getConsumerCount());
- props.putIntProperty(ManagementHelper.HDR_CONSUMER_COUNT, theQueue.getConsumerCount());
+ if (filterString != null)
+ {
+ props.putSimpleStringProperty(ManagementHelper.HDR_FILTERSTRING, filterString);
+ }
- if (filterString != null)
- {
- props.putSimpleStringProperty(ManagementHelper.HDR_FILTERSTRING, filterString);
- }
+ Notification notification = new Notification(null, CONSUMER_CREATED, props);
- Notification notification = new Notification(null, CONSUMER_CREATED, props);
+ managementService.sendNotification(notification);
+ }
+ }
- managementService.sendNotification(notification);
- }
-
- //We send back queue information on the queue as a response- this allows the queue to
- //be automaticall recreated on failover
-
- if (packet.isRequiresResponse())
- {
- response = doExecuteQueueQuery(name);
- }
- else
- {
- response = null;
- }
+ public void handleCreateQueue(final SimpleString address,
+ final SimpleString name,
+ final SimpleString filterString,
+ final boolean temporary,
+ final boolean durable) throws Exception
+ {
+ if (durable)
+ {
+ // make sure the user has privileges to create this queue
+ securityStore.check(address, CheckType.CREATE_DURABLE_QUEUE, this);
}
- catch (Exception e)
+ else
{
- if (e instanceof HornetQException)
- {
- response = new HornetQExceptionMessage((HornetQException)e);
- }
- else
- {
- ServerSessionImpl.log.error("Failed to create consumer", e);
- response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR));
- }
+ securityStore.check(address, CheckType.CREATE_NON_DURABLE_QUEUE, this);
}
-
- sendResponse(packet, response, false, false);
- }
- public void handleCreateQueue(final CreateQueueMessage packet)
- {
- SimpleString address = packet.getAddress();
+ server.createQueue(address, name, filterString, durable, temporary);
- final SimpleString name = packet.getQueueName();
-
- SimpleString filterString = packet.getFilterString();
-
- boolean temporary = packet.isTemporary();
-
- boolean durable = packet.isDurable();
-
- Packet response = null;
-
- try
+ if (temporary)
{
- if (durable)
- {
- // make sure the user has privileges to create this queue
- securityStore.check(address, CheckType.CREATE_DURABLE_QUEUE, this);
- }
- else
- {
- securityStore.check(address, CheckType.CREATE_NON_DURABLE_QUEUE, this);
- }
+ // Temporary queue in core simply means the queue will be deleted if
+ // the remoting connection
+ // dies. It does not mean it will get deleted automatically when the
+ // session is closed.
+ // It is up to the user to delete the queue when finished with it
- server.createQueue(address, name, filterString, durable, temporary);
-
- if (temporary)
+ failureRunners.put(name, new Runnable()
{
- // Temporary queue in core simply means the queue will be deleted if
- // the remoting connection
- // dies. It does not mean it will get deleted automatically when the
- // session is closed.
- // It is up to the user to delete the queue when finished with it
-
- failureRunners.put(name, new Runnable()
+ public void run()
{
- public void run()
+ try
{
- try
+ if (postOffice.getBinding(name) != null)
{
- if (postOffice.getBinding(name) != null)
+ postOffice.removeBinding(name);
+
+ if (postOffice.getBindingsForAddress(name).getBindings().size() == 0)
{
- postOffice.removeBinding(name);
-
- if (postOffice.getBindingsForAddress(name).getBindings().size() == 0)
- {
- creditManagerHolders.remove(name);
- }
+ creditManagerHolders.remove(name);
}
}
- catch (Exception e)
- {
- ServerSessionImpl.log.error("Failed to remove temporary queue " + name);
- }
}
- });
- }
-
- if (packet.isRequiresResponse())
- {
- response = new NullResponseMessage();
- }
- else
- {
- response = null;
- }
+ catch (Exception e)
+ {
+ ServerSessionImpl.log.error("Failed to remove temporary queue " + name);
+ }
+ }
+ });
}
- catch (Exception e)
- {
- if (e instanceof HornetQException)
- {
- response = new HornetQExceptionMessage((HornetQException)e);
- }
- else
- {
- ServerSessionImpl.log.error("Failed to create queue", e);
- response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR));
- }
- }
-
- sendResponse(packet, response, false, false);
}
- public void handleDeleteQueue(final SessionDeleteQueueMessage packet)
+ public void handleDeleteQueue(final SimpleString name) throws Exception
{
- SimpleString name = packet.getQueueName();
+ Binding binding = postOffice.getBinding(name);
- Packet response = null;
-
- try
+ if (binding == null || binding.getType() != BindingType.LOCAL_QUEUE)
{
- Binding binding = postOffice.getBinding(name);
+ throw new HornetQException(HornetQException.QUEUE_DOES_NOT_EXIST);
+ }
- if (binding == null || binding.getType() != BindingType.LOCAL_QUEUE)
- {
- throw new HornetQException(HornetQException.QUEUE_DOES_NOT_EXIST);
- }
+ server.destroyQueue(name, this);
- server.destroyQueue(name, this);
+ failureRunners.remove(name);
- failureRunners.remove(name);
-
- if (postOffice.getBindingsForAddress(name).getBindings().size() == 0)
- {
- creditManagerHolders.remove(name);
- }
-
- response = new NullResponseMessage();
- }
- catch (Exception e)
+ if (postOffice.getBindingsForAddress(name).getBindings().size() == 0)
{
- if (e instanceof HornetQException)
- {
- response = new HornetQExceptionMessage((HornetQException)e);
- }
- else
- {
- ServerSessionImpl.log.error("Failed to delete queue", e);
- response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR));
- }
+ creditManagerHolders.remove(name);
}
-
- sendResponse(packet, response, false, false);
}
-
- public void handleExecuteQueueQuery(final SessionQueueQueryMessage packet)
- {
- SimpleString name = packet.getQueueName();
- Packet response = null;
-
- try
+ public QueueQueryResult handleExecuteQueueQuery(final SimpleString name) throws Exception
+ {
+ if (name == null)
{
- response = doExecuteQueueQuery(name);
+ throw new IllegalArgumentException("Queue name is null");
}
- catch (Exception e)
- {
- if (e instanceof HornetQException)
- {
- response = new HornetQExceptionMessage((HornetQException)e);
- }
- else
- {
- ServerSessionImpl.log.error("Failed to execute queue query", e);
- response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR));
- }
- }
+ QueueQueryResult response;
- sendResponse(packet, response, false, false);
- }
+ Binding binding = postOffice.getBinding(name);
- public void handleExecuteBindingQuery(final SessionBindingQueryMessage packet)
- {
- SimpleString address = packet.getAddress();
-
- Packet response = null;
-
- try
+ if (binding != null && binding.getType() == BindingType.LOCAL_QUEUE)
{
- if (address == null)
- {
- throw new IllegalArgumentException("Address is null");
- }
+ Queue queue = (Queue)binding.getBindable();
- List<SimpleString> names = new ArrayList<SimpleString>();
+ Filter filter = queue.getFilter();
- Bindings bindings = postOffice.getMatchingBindings(address);
+ SimpleString filterString = filter == null ? null : filter.getFilterString();
- for (Binding binding : bindings.getBindings())
- {
- if (binding.getType() == BindingType.LOCAL_QUEUE)
- {
- names.add(binding.getUniqueName());
- }
- }
-
- response = new SessionBindingQueryResponseMessage(!names.isEmpty(), names);
+ response = new QueueQueryResult(name,
+ binding.getAddress(),
+ queue.isDurable(),
+ queue.isTemporary(),
+ filterString,
+ queue.getConsumerCount(),
+ queue.getMessageCount());
}
- catch (Exception e)
+ // make an exception for the management address (see HORNETQ-29)
+ else if (name.equals(managementAddress))
{
- ServerSessionImpl.log.error("Failed to execute binding query", e);
-
- if (e instanceof HornetQException)
- {
- response = new HornetQExceptionMessage((HornetQException)e);
- }
- else
- {
- response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR));
- }
+ response = new QueueQueryResult(name, managementAddress, true, false, null, -1, -1);
}
+ else
+ {
+ response = new QueueQueryResult();
+ }
- sendResponse(packet, response, false, false);
+ return response;
}
- public void handleForceConsumerDelivery(final SessionForceConsumerDelivery message)
+ public BindingQueryResult handleExecuteBindingQuery(final SimpleString address)
{
- try
+ if (address == null)
{
- ServerConsumer consumer = consumers.get(message.getConsumerID());
-
- consumer.forceDelivery(message.getSequence());
+ throw new IllegalArgumentException("Address is null");
}
- catch (Exception e)
- {
- ServerSessionImpl.log.error("Failed to query consumer deliveries", e);
- }
- sendResponse(message, null, false, false);
- }
+ List<SimpleString> names = new ArrayList<SimpleString>();
- public void handleAcknowledge(final SessionAcknowledgeMessage packet)
- {
- Packet response = null;
+ Bindings bindings = postOffice.getMatchingBindings(address);
- try
+ for (Binding binding : bindings.getBindings())
{
- ServerConsumer consumer = consumers.get(packet.getConsumerID());
-
- consumer.acknowledge(autoCommitAcks, tx, packet.getMessageID());
-
- if (packet.isRequiresResponse())
+ if (binding.getType() == BindingType.LOCAL_QUEUE)
{
- response = new NullResponseMessage();
+ names.add(binding.getUniqueName());
}
}
- catch (Exception e)
- {
- ServerSessionImpl.log.error("Failed to acknowledge", e);
- if (packet.isRequiresResponse())
- {
- if (e instanceof HornetQException)
- {
- response = new HornetQExceptionMessage((HornetQException)e);
- }
- else
- {
- response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR));
- }
- }
- }
+ return new BindingQueryResult(!names.isEmpty(), names);
+ }
- sendResponse(packet, response, false, false);
+ public void handleForceConsumerDelivery(final long consumerID, final long sequence) throws Exception
+ {
+ ServerConsumer consumer = consumers.get(consumerID);
+
+ consumer.forceDelivery(sequence);
}
- public void handleExpired(final SessionExpiredMessage packet)
+ public void handleAcknowledge(final long consumerID, final long messageID) throws Exception
{
- try
- {
- MessageReference ref = consumers.get(packet.getConsumerID()).getExpired(packet.getMessageID());
+ ServerConsumer consumer = consumers.get(consumerID);
- if (ref != null)
- {
- ref.getQueue().expire(ref);
- }
- }
- catch (Exception e)
+ consumer.acknowledge(autoCommitAcks, tx, messageID);
+ }
+
+ public void handleExpired(final long consumerID, final long messageID) throws Exception
+ {
+ MessageReference ref = consumers.get(consumerID).getExpired(messageID);
+
+ if (ref != null)
{
- ServerSessionImpl.log.error("Failed to acknowledge", e);
+ ref.getQueue().expire(ref);
}
-
- sendResponse(packet, null, false, false);
}
- public void handleCommit(final Packet packet)
+ public void handleCommit() throws Exception
{
- Packet response = null;
-
try
{
tx.commit();
-
- response = new NullResponseMessage();
}
- catch (Exception e)
- {
- ServerSessionImpl.log.error("Failed to commit", e);
-
- if (e instanceof HornetQException)
- {
- response = new HornetQExceptionMessage((HornetQException)e);
- }
- else
- {
- response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR));
- }
- }
finally
{
tx = new TransactionImpl(storageManager);
}
-
- sendResponse(packet, response, false, false);
}
- public void handleRollback(final RollbackMessage packet)
+ public void handleRollback(final boolean considerLastMessageAsDelivered) throws Exception
{
- Packet response = null;
+ rollback(considerLastMessageAsDelivered);
+ }
- try
+ public void handleXACommit(final Xid xid, final boolean onePhase) throws Exception
+ {
+ if (tx != null)
{
- rollback(packet.isConsiderLastMessageAsDelivered());
+ final String msg = "Cannot commit, session is currently doing work in transaction " + tx.getXid();
- response = new NullResponseMessage();
+ throw new HornetQXAException(XAException.XAER_PROTO, msg);
}
- catch (Exception e)
+ else
{
- ServerSessionImpl.log.error("Failed to rollback", e);
+ Transaction theTx = resourceManager.removeTransaction(xid);
- if (e instanceof HornetQException)
+ if (theTx == null)
{
- response = new HornetQExceptionMessage((HornetQException)e);
+ // checked heuristic committed transactions
+ if (resourceManager.getHeuristicCommittedTransactions().contains(xid))
+ {
+ throw new HornetQXAException(XAException.XA_HEURCOM,
+ "transaction has been heuristically committed: " + xid);
+ }
+ // checked heuristic rolled back transactions
+ else if (resourceManager.getHeuristicRolledbackTransactions().contains(xid))
+ {
+ throw new HornetQXAException(XAException.XA_HEURRB,
+ "transaction has been heuristically rolled back: " + xid);
+ }
+ else
+ {
+ throw new HornetQXAException(XAException.XAER_NOTA, "Cannot find xid in resource manager: " + xid);
+ }
}
else
{
- response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR));
+ if (theTx.getState() == Transaction.State.SUSPENDED)
+ {
+ // Put it back
+ resourceManager.putTransaction(xid, tx);
+
+ throw new HornetQXAException(XAException.XAER_PROTO, "Cannot commit transaction, it is suspended " + xid);
+ }
+ else
+ {
+ theTx.commit(onePhase);
+ }
}
}
-
- sendResponse(packet, response, false, false);
}
- public void handleXACommit(final SessionXACommitMessage packet)
+ public void handleXAEnd(final Xid xid) throws Exception
{
- Packet response = null;
-
- Xid xid = packet.getXid();
-
- try
+ if (tx != null && tx.getXid().equals(xid))
{
- if (tx != null)
+ if (tx.getState() == Transaction.State.SUSPENDED)
{
- final String msg = "Cannot commit, session is currently doing work in transaction " + tx.getXid();
+ final String msg = "Cannot end, transaction is suspended";
- response = new SessionXAResponseMessage(true, XAException.XAER_PROTO, msg);
+ throw new HornetQXAException(XAException.XAER_PROTO, msg);
}
else
{
- Transaction theTx = resourceManager.removeTransaction(xid);
-
- if (theTx == null)
- {
- // checked heuristic committed transactions
- if (resourceManager.getHeuristicCommittedTransactions().contains(xid))
- {
- response = new SessionXAResponseMessage(true,
- XAException.XA_HEURCOM,
- "transaction has been heuristically committed: " + xid);
- }
- // checked heuristic rolled back transactions
- else if (resourceManager.getHeuristicRolledbackTransactions().contains(xid))
- {
- response = new SessionXAResponseMessage(true,
- XAException.XA_HEURRB,
- "transaction has been heuristically rolled back: " + xid);
- }
- else
- {
- response = new SessionXAResponseMessage(true,
- XAException.XAER_NOTA,
- "Cannot find xid in resource manager: " + xid);
- }
- }
- else
- {
- if (theTx.getState() == Transaction.State.SUSPENDED)
- {
- // Put it back
- resourceManager.putTransaction(xid, tx);
-
- response = new SessionXAResponseMessage(true,
- XAException.XAER_PROTO,
- "Cannot commit transaction, it is suspended " + xid);
- }
- else
- {
- theTx.commit(packet.isOnePhase());
-
- response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
- }
- }
+ tx = null;
}
}
- catch (Exception e)
+ else
{
- ServerSessionImpl.log.error("Failed to xa commit", e);
+ // It's also legal for the TM to call end for a Xid in the suspended
+ // state
+ // See JTA 1.1 spec 3.4.4 - state diagram
+ // Although in practice TMs rarely do this.
+ Transaction theTx = resourceManager.getTransaction(xid);
- if (e instanceof HornetQException)
+ if (theTx == null)
{
- response = new HornetQExceptionMessage((HornetQException)e);
- }
- else
- {
- response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR));
- }
- }
+ final String msg = "Cannot find suspended transaction to end " + xid;
- sendResponse(packet, response, false, false);
- }
-
- public void handleXAEnd(final SessionXAEndMessage packet)
- {
- Packet response = null;
-
- Xid xid = packet.getXid();
-
- try
- {
- if (tx != null && tx.getXid().equals(xid))
- {
- if (tx.getState() == Transaction.State.SUSPENDED)
- {
- final String msg = "Cannot end, transaction is suspended";
-
- response = new SessionXAResponseMessage(true, XAException.XAER_PROTO, msg);
- }
- else
- {
- tx = null;
- }
+ throw new HornetQXAException(XAException.XAER_NOTA, msg);
}
else
{
- // It's also legal for the TM to call end for a Xid in the suspended
- // state
- // See JTA 1.1 spec 3.4.4 - state diagram
- // Although in practice TMs rarely do this.
- Transaction theTx = resourceManager.getTransaction(xid);
-
- if (theTx == null)
+ if (theTx.getState() != Transaction.State.SUSPENDED)
{
- final String msg = "Cannot find suspended transaction to end " + xid;
+ final String msg = "Transaction is not suspended " + xid;
- response = new SessionXAResponseMessage(true, XAException.XAER_NOTA, msg);
+ throw new HornetQXAException(XAException.XAER_PROTO, msg);
}
else
{
- if (theTx.getState() != Transaction.State.SUSPENDED)
- {
- final String msg = "Transaction is not suspended " + xid;
-
- response = new SessionXAResponseMessage(true, XAException.XAER_PROTO, msg);
- }
- else
- {
- theTx.resume();
- }
+ theTx.resume();
}
}
-
- if (response == null)
- {
- response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
- }
}
- catch (Exception e)
- {
- ServerSessionImpl.log.error("Failed to xa end", e);
-
- if (e instanceof HornetQException)
- {
- response = new HornetQExceptionMessage((HornetQException)e);
- }
- else
- {
- response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR));
- }
- }
-
- sendResponse(packet, response, false, false);
}
- public void handleXAForget(final SessionXAForgetMessage packet)
+ public void handleXAForget(final Xid xid) throws Exception
{
- long id = resourceManager.removeHeuristicCompletion(packet.getXid());
- int code = XAResource.XA_OK;
+ long id = resourceManager.removeHeuristicCompletion(xid);
+
if (id != -1)
{
try
@@ -912,402 +626,236 @@
catch (Exception e)
{
e.printStackTrace();
- code = XAException.XAER_RMERR;
+
+ throw new HornetQXAException(XAException.XAER_RMERR);
}
}
else
{
- code = XAException.XAER_NOTA;
+ throw new HornetQXAException(XAException.XAER_NOTA);
}
-
- Packet response = new SessionXAResponseMessage((code != XAResource.XA_OK), code, null);
-
- sendResponse(packet, response, false, false);
}
- public void handleXAJoin(final SessionXAJoinMessage packet)
+ public void handleXAJoin(final Xid xid) throws Exception
{
- Packet response = null;
+ Transaction theTx = resourceManager.getTransaction(xid);
- Xid xid = packet.getXid();
-
- try
+ if (theTx == null)
{
- Transaction theTx = resourceManager.getTransaction(xid);
+ final String msg = "Cannot find xid in resource manager: " + xid;
- if (theTx == null)
- {
- final String msg = "Cannot find xid in resource manager: " + xid;
-
- response = new SessionXAResponseMessage(true, XAException.XAER_NOTA, msg);
- }
- else
- {
- if (theTx.getState() == Transaction.State.SUSPENDED)
- {
- response = new SessionXAResponseMessage(true,
- XAException.XAER_PROTO,
- "Cannot join tx, it is suspended " + xid);
- }
- else
- {
- tx = theTx;
-
- response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
- }
- }
+ throw new HornetQXAException(XAException.XAER_NOTA, msg);
}
- catch (Exception e)
+ else
{
- ServerSessionImpl.log.error("Failed to xa join", e);
-
- if (e instanceof HornetQException)
+ if (theTx.getState() == Transaction.State.SUSPENDED)
{
- response = new HornetQExceptionMessage((HornetQException)e);
+ throw new HornetQXAException(XAException.XAER_PROTO, "Cannot join tx, it is suspended " + xid);
}
else
{
- response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR));
+ tx = theTx;
}
}
-
- sendResponse(packet, response, false, false);
}
- public void handleXAResume(final SessionXAResumeMessage packet)
+ public void handleXAResume(final Xid xid) throws Exception
{
- Packet response = null;
+ if (tx != null)
+ {
+ final String msg = "Cannot resume, session is currently doing work in a transaction " + tx.getXid();
- Xid xid = packet.getXid();
-
- try
+ throw new HornetQXAException(XAException.XAER_PROTO, msg);
+ }
+ else
{
- if (tx != null)
+ Transaction theTx = resourceManager.getTransaction(xid);
+
+ if (theTx == null)
{
- final String msg = "Cannot resume, session is currently doing work in a transaction " + tx.getXid();
+ final String msg = "Cannot find xid in resource manager: " + xid;
- response = new SessionXAResponseMessage(true, XAException.XAER_PROTO, msg);
+ throw new HornetQXAException(XAException.XAER_NOTA, msg);
}
else
{
- Transaction theTx = resourceManager.getTransaction(xid);
-
- if (theTx == null)
+ if (theTx.getState() != Transaction.State.SUSPENDED)
{
- final String msg = "Cannot find xid in resource manager: " + xid;
-
- response = new SessionXAResponseMessage(true, XAException.XAER_NOTA, msg);
+ throw new HornetQXAException(XAException.XAER_PROTO,
+ "Cannot resume transaction, it is not suspended " + xid);
}
else
{
- if (theTx.getState() != Transaction.State.SUSPENDED)
- {
- response = new SessionXAResponseMessage(true,
- XAException.XAER_PROTO,
- "Cannot resume transaction, it is not suspended " + xid);
- }
- else
- {
- tx = theTx;
+ tx = theTx;
- tx.resume();
-
- response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
- }
+ tx.resume();
}
}
}
- catch (Exception e)
- {
- ServerSessionImpl.log.error("Failed to xa resume", e);
-
- if (e instanceof HornetQException)
- {
- response = new HornetQExceptionMessage((HornetQException)e);
- }
- else
- {
- response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR));
- }
- }
-
- sendResponse(packet, response, false, false);
}
- public void handleXARollback(final SessionXARollbackMessage packet)
+ public void handleXARollback(final Xid xid) throws Exception
{
- Packet response = null;
+ if (tx != null)
+ {
+ final String msg = "Cannot roll back, session is currently doing work in a transaction " + tx.getXid();
- Xid xid = packet.getXid();
-
- try
+ throw new HornetQXAException(XAException.XAER_PROTO, msg);
+ }
+ else
{
- if (tx != null)
- {
- final String msg = "Cannot roll back, session is currently doing work in a transaction " + tx.getXid();
+ Transaction theTx = resourceManager.removeTransaction(xid);
- response = new SessionXAResponseMessage(true, XAException.XAER_PROTO, msg);
- }
- else
+ if (theTx == null)
{
- Transaction theTx = resourceManager.removeTransaction(xid);
-
- if (theTx == null)
+ // checked heuristic committed transactions
+ if (resourceManager.getHeuristicCommittedTransactions().contains(xid))
{
- // checked heuristic committed transactions
- if (resourceManager.getHeuristicCommittedTransactions().contains(xid))
- {
- response = new SessionXAResponseMessage(true,
- XAException.XA_HEURCOM,
- "transaction has ben heuristically committed: " + xid);
- }
- // checked heuristic rolled back transactions
- else if (resourceManager.getHeuristicRolledbackTransactions().contains(xid))
- {
- response = new SessionXAResponseMessage(true,
- XAException.XA_HEURRB,
- "transaction has ben heuristically rolled back: " + xid);
- }
- else
- {
- response = new SessionXAResponseMessage(true,
- XAException.XAER_NOTA,
- "Cannot find xid in resource manager: " + xid);
- }
+ throw new HornetQXAException(XAException.XA_HEURCOM,
+ "transaction has ben heuristically committed: " + xid);
}
+ // checked heuristic rolled back transactions
+ else if (resourceManager.getHeuristicRolledbackTransactions().contains(xid))
+ {
+ throw new HornetQXAException(XAException.XA_HEURRB,
+ "transaction has ben heuristically rolled back: " + xid);
+ }
else
{
- if (theTx.getState() == Transaction.State.SUSPENDED)
- {
- // Put it back
- resourceManager.putTransaction(xid, tx);
-
- response = new SessionXAResponseMessage(true,
- XAException.XAER_PROTO,
- "Cannot rollback transaction, it is suspended " + xid);
- }
- else
- {
- doRollback(false, theTx);
-
- response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
- }
+ throw new HornetQXAException(XAException.XAER_NOTA, "Cannot find xid in resource manager: " + xid);
}
}
- }
- catch (Exception e)
- {
- ServerSessionImpl.log.error("Failed to xa rollback", e);
-
- if (e instanceof HornetQException)
- {
- response = new HornetQExceptionMessage((HornetQException)e);
- }
else
{
- response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR));
- }
- }
-
- sendResponse(packet, response, false, false);
- }
-
- public void handleXAStart(final SessionXAStartMessage packet)
- {
- Packet response = null;
-
- Xid xid = packet.getXid();
-
- try
- {
- if (tx != null)
- {
- final String msg = "Cannot start, session is already doing work in a transaction " + tx.getXid();
-
- response = new SessionXAResponseMessage(true, XAException.XAER_PROTO, msg);
- }
- else
- {
- tx = new TransactionImpl(xid, storageManager, postOffice);
-
- boolean added = resourceManager.putTransaction(xid, tx);
-
- if (!added)
+ if (theTx.getState() == Transaction.State.SUSPENDED)
{
- final String msg = "Cannot start, there is already a xid " + tx.getXid();
+ // Put it back
+ resourceManager.putTransaction(xid, tx);
- response = new SessionXAResponseMessage(true, XAException.XAER_DUPID, msg);
+ throw new HornetQXAException(XAException.XAER_PROTO,
+ "Cannot rollback transaction, it is suspended " + xid);
}
else
{
- response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
+ doRollback(false, theTx);
}
}
}
- catch (Exception e)
+ }
+
+ public void handleXAStart(final Xid xid) throws Exception
+ {
+ if (tx != null)
{
- ServerSessionImpl.log.error("Failed to xa start", e);
+ final String msg = "Cannot start, session is already doing work in a transaction " + tx.getXid();
- if (e instanceof HornetQException)
+ throw new HornetQXAException(XAException.XAER_PROTO, msg);
+ }
+ else
+ {
+ tx = new TransactionImpl(xid, storageManager, postOffice);
+
+ boolean added = resourceManager.putTransaction(xid, tx);
+
+ if (!added)
{
- response = new HornetQExceptionMessage((HornetQException)e);
+ final String msg = "Cannot start, there is already a xid " + tx.getXid();
+
+ throw new HornetQXAException(XAException.XAER_DUPID, msg);
}
- else
- {
- response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR));
- }
}
-
- sendResponse(packet, response, false, false);
}
- public void handleXASuspend(final Packet packet)
+ public void handleXASuspend() throws Exception
{
- Packet response = null;
+ if (tx == null)
+ {
+ final String msg = "Cannot suspend, session is not doing work in a transaction ";
- try
+ throw new HornetQXAException(XAException.XAER_PROTO, msg);
+ }
+ else
{
- if (tx == null)
+ if (tx.getState() == Transaction.State.SUSPENDED)
{
- final String msg = "Cannot suspend, session is not doing work in a transaction ";
+ final String msg = "Cannot suspend, transaction is already suspended " + tx.getXid();
- response = new SessionXAResponseMessage(true, XAException.XAER_PROTO, msg);
+ throw new HornetQXAException(XAException.XAER_PROTO, msg);
}
else
{
- if (tx.getState() == Transaction.State.SUSPENDED)
- {
- final String msg = "Cannot suspend, transaction is already suspended " + tx.getXid();
+ tx.suspend();
- response = new SessionXAResponseMessage(true, XAException.XAER_PROTO, msg);
- }
- else
- {
- tx.suspend();
-
- tx = null;
-
- response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
- }
+ tx = null;
}
}
- catch (Exception e)
- {
- ServerSessionImpl.log.error("Failed to xa suspend", e);
-
- if (e instanceof HornetQException)
- {
- response = new HornetQExceptionMessage((HornetQException)e);
- }
- else
- {
- response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR));
- }
- }
-
- sendResponse(packet, response, false, false);
}
- public void handleXAPrepare(final SessionXAPrepareMessage packet)
+ public void handleXAPrepare(final Xid xid) throws Exception
{
- Packet response = null;
+ if (tx != null)
+ {
+ final String msg = "Cannot commit, session is currently doing work in a transaction " + tx.getXid();
- Xid xid = packet.getXid();
-
- try
+ throw new HornetQXAException(XAException.XAER_PROTO, msg);
+ }
+ else
{
- if (tx != null)
+ Transaction theTx = resourceManager.getTransaction(xid);
+
+ if (theTx == null)
{
- final String msg = "Cannot commit, session is currently doing work in a transaction " + tx.getXid();
+ final String msg = "Cannot find xid in resource manager: " + xid;
- response = new SessionXAResponseMessage(true, XAException.XAER_PROTO, msg);
+ throw new HornetQXAException(XAException.XAER_NOTA, msg);
}
else
{
- Transaction theTx = resourceManager.getTransaction(xid);
-
- if (theTx == null)
+ if (theTx.getState() == Transaction.State.SUSPENDED)
{
- final String msg = "Cannot find xid in resource manager: " + xid;
-
- response = new SessionXAResponseMessage(true, XAException.XAER_NOTA, msg);
+ throw new HornetQXAException(XAException.XAER_PROTO,
+ "Cannot prepare transaction, it is suspended " + xid);
}
else
{
- if (theTx.getState() == Transaction.State.SUSPENDED)
- {
- response = new SessionXAResponseMessage(true,
- XAException.XAER_PROTO,
- "Cannot prepare transaction, it is suspended " + xid);
- }
- else
- {
- theTx.prepare();
-
- response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
- }
+ theTx.prepare();
}
}
}
- catch (Exception e)
- {
- ServerSessionImpl.log.error("Failed to xa prepare", e);
-
- if (e instanceof HornetQException)
- {
- response = new HornetQExceptionMessage((HornetQException)e);
- }
- else
- {
- response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR));
- }
- }
-
- sendResponse(packet, response, false, false);
}
- public void handleGetInDoubtXids(final Packet packet)
+ public List<Xid> handleGetInDoubtXids()
{
- List<Xid> indoubtsXids = new ArrayList<Xid>();
- indoubtsXids.addAll(resourceManager.getPreparedTransactions());
- indoubtsXids.addAll(resourceManager.getHeuristicCommittedTransactions());
- indoubtsXids.addAll(resourceManager.getHeuristicRolledbackTransactions());
- Packet response = new SessionXAGetInDoubtXidsResponseMessage(indoubtsXids);
+ List<Xid> xids = new ArrayList<Xid>();
- sendResponse(packet, response, false, false);
+ xids.addAll(resourceManager.getPreparedTransactions());
+ xids.addAll(resourceManager.getHeuristicCommittedTransactions());
+ xids.addAll(resourceManager.getHeuristicRolledbackTransactions());
+
+ return xids;
}
- public void handleGetXATimeout(final Packet packet)
+ public int handleGetXATimeout()
{
- Packet response = new SessionXAGetTimeoutResponseMessage(resourceManager.getTimeoutSeconds());
-
- sendResponse(packet, response, false, false);
+ return resourceManager.getTimeoutSeconds();
}
- public void handleSetXATimeout(final SessionXASetTimeoutMessage packet)
+ public void handleSetXATimeout(final int timeout)
{
- Packet response = new SessionXASetTimeoutResponseMessage(resourceManager.setTimeoutSeconds(packet.getTimeoutSeconds()));
-
- sendResponse(packet, response, false, false);
+ resourceManager.setTimeoutSeconds(timeout);
}
- public void handleStart(final Packet packet)
+ public void handleStart()
{
setStarted(true);
-
- sendResponse(packet, null, false, false);
}
- public void handleStop(final Packet packet)
+ public void handleStop()
{
- final Packet response = new NullResponseMessage();
-
setStarted(false);
-
- sendResponse(packet, response, false, false);
}
- public void handleClose(final Packet packet)
+ public void handleClose()
{
storageManager.afterCompleteOperations(new IOAsyncTask()
{
@@ -1317,97 +865,63 @@
public void done()
{
- doClose(packet);
+ try
+ {
+ close();
+ }
+ catch (Exception e)
+ {
+ log.error("Failed to close session", e);
+ }
}
});
}
- public void handleCloseConsumer(final SessionConsumerCloseMessage packet)
+ public void handleCloseConsumer(final long consumerID) throws Exception
{
- final ServerConsumer consumer = consumers.get(packet.getConsumerID());
+ final ServerConsumer consumer = consumers.get(consumerID);
- Packet response;
-
- try
+ if (consumer != null)
{
- if (consumer != null)
- {
- consumer.close();
- }
- else
- {
- ServerSessionImpl.log.error("Cannot find consumer with id " + packet.getConsumerID());
- }
-
- response = new NullResponseMessage();
+ consumer.close();
}
- catch (Exception e)
+ else
{
- ServerSessionImpl.log.error("Failed to close consumer", e);
-
- if (e instanceof HornetQException)
- {
- response = new HornetQExceptionMessage((HornetQException)e);
- }
- else
- {
- response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR));
- }
+ ServerSessionImpl.log.error("Cannot find consumer with id " + consumerID);
}
-
- sendResponse(packet, response, false, false);
}
- public void handleReceiveConsumerCredits(final SessionConsumerFlowCreditMessage packet)
+ public void handleReceiveConsumerCredits(final long consumerID, final int credits) throws Exception
{
- ServerConsumer consumer = consumers.get(packet.getConsumerID());
+ ServerConsumer consumer = consumers.get(consumerID);
if (consumer == null)
{
- ServerSessionImpl.log.error("There is no consumer with id " + packet.getConsumerID());
-
+ ServerSessionImpl.log.error("There is no consumer with id " + consumerID);
+
return;
}
- try
- {
- consumer.receiveCredits(packet.getCredits());
- }
- catch (Exception e)
- {
- ServerSessionImpl.log.error("Failed to receive credits " + server.getConfiguration().isBackup(), e);
- }
-
- sendResponse(packet, null, false, false);
+ consumer.receiveCredits(credits);
}
- public void handleSendLargeMessage(final SessionSendLargeMessage packet)
+ public void handleSendLargeMessage(final byte[] largeMessageHeader) throws Exception
{
-
// need to create the LargeMessage before continue
long id = storageManager.generateUniqueID();
- LargeServerMessage msg = doCreateLargeMessage(id, packet);
+ LargeServerMessage msg = storageManager.createLargeMessage(id, largeMessageHeader);
- if (msg != null)
+ if (currentLargeMessage != null)
{
- if (currentLargeMessage != null)
- {
- ServerSessionImpl.log.warn("Replacing incomplete LargeMessage with ID=" + currentLargeMessage.getMessageID());
- }
-
- currentLargeMessage = msg;
-
- sendResponse(packet, null, false, false);
+ ServerSessionImpl.log.warn("Replacing incomplete LargeMessage with ID=" + currentLargeMessage.getMessageID());
}
+
+ currentLargeMessage = msg;
}
- public void handleSend(final SessionSendMessage packet)
+ public void handleSend(final ServerMessage message) throws Exception
{
- Packet response = null;
-
- ServerMessage message = (ServerMessage)packet.getMessage();
-
try
{
long id = storageManager.generateUniqueID();
@@ -1425,28 +939,7 @@
{
send(message);
}
-
- if (packet.isRequiresResponse())
- {
- response = new NullResponseMessage();
- }
}
- catch (Exception e)
- {
- ServerSessionImpl.log.error("Failed to send message", e);
-
- if (packet.isRequiresResponse())
- {
- if (e instanceof HornetQException)
- {
- response = new HornetQExceptionMessage((HornetQException)e);
- }
- else
- {
- response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR));
- }
- }
- }
finally
{
try
@@ -1458,74 +951,40 @@
ServerSessionImpl.log.error("Failed to release outstanding credits", e);
}
}
-
- sendResponse(packet, response, false, false);
}
- public void handleSendContinuations(final SessionSendContinuationMessage packet)
+ public void handleSendContinuations(final int packetSize, final byte[] body, final boolean continues) throws Exception
{
- Packet response = null;
-
- try
+ if (currentLargeMessage == null)
{
- if (currentLargeMessage == null)
- {
- throw new HornetQException(HornetQException.ILLEGAL_STATE, "large-message not initialized on server");
- }
+ throw new HornetQException(HornetQException.ILLEGAL_STATE, "large-message not initialized on server");
+ }
- // Immediately release the credits for the continuations- these don't contribute to the in-memory size
- // of the message
+ // Immediately release the credits for the continuations- these don't contribute to the in-memory size
+ // of the message
- releaseOutStanding(currentLargeMessage, packet.getPacketSize());
+ releaseOutStanding(currentLargeMessage, packetSize);
- currentLargeMessage.addBytes(packet.getBody());
+ currentLargeMessage.addBytes(body);
- if (!packet.isContinues())
- {
- currentLargeMessage.releaseResources();
+ if (!continues)
+ {
+ currentLargeMessage.releaseResources();
- send(currentLargeMessage);
+ send(currentLargeMessage);
- releaseOutStanding(currentLargeMessage, currentLargeMessage.getEncodeSize());
+ releaseOutStanding(currentLargeMessage, currentLargeMessage.getEncodeSize());
- currentLargeMessage = null;
- }
-
- if (packet.isRequiresResponse())
- {
- response = new NullResponseMessage();
- }
+ currentLargeMessage = null;
}
- catch (Exception e)
- {
- if (packet.isRequiresResponse())
- {
- if (e instanceof HornetQException)
- {
- response = new HornetQExceptionMessage((HornetQException)e);
- }
- else
- {
- ServerSessionImpl.log.error("Failed to send message", e);
-
- response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR));
- }
- }
- }
-
- sendResponse(packet, response, false, false);
}
- public void handleRequestProducerCredits(final SessionRequestProducerCreditsMessage packet) throws Exception
+ public void handleRequestProducerCredits(final SimpleString address, final int credits) throws Exception
{
- final SimpleString address = packet.getAddress();
-
final CreditManagerHolder holder = getCreditManagerHolder(address);
- int credits = packet.getCredits();
-
- //Requesting -ve credits means returning them
-
+ // Requesting -ve credits means returning them
+
if (credits < 0)
{
releaseOutStanding(address, -credits);
@@ -1541,7 +1000,7 @@
if (!closed)
{
sendProducerCredits(holder, credits, address);
-
+
return true;
}
else
@@ -1551,23 +1010,33 @@
}
}
});
-
+
if (gotCredits > 0)
{
sendProducerCredits(holder, gotCredits, address);
}
}
+ }
+
+ public void setTransferring(final boolean transferring)
+ {
+ Set<ServerConsumer> consumersClone = new HashSet<ServerConsumer>(consumers.values());
- sendResponse(packet, null, false, false);
+ for (ServerConsumer consumer : consumersClone)
+ {
+ consumer.setTransferring(transferring);
+ }
}
public int transferConnection(final RemotingConnection newConnection, final int lastReceivedCommandID)
- {
- //We need to disable delivery on all the consumers while the transfer is occurring- otherwise packets might get delivered
- //after the channel has transferred but *before* packets have been replayed - this will give the client the wrong
- //sequence of packets.
- //It is not sufficient to just stop the session, since right after stopping the session, another session start might be executed
- //before we have transferred the connection, leaving it in a started state
+ {
+ // We need to disable delivery on all the consumers while the transfer is occurring- otherwise packets might get
+ // delivered
+ // after the channel has transferred but *before* packets have been replayed - this will give the client the wrong
+ // sequence of packets.
+ // It is not sufficient to just stop the session, since right after stopping the session, another session start
+ // might be executed
+ // before we have transferred the connection, leaving it in a started state
setTransferring(true);
remotingConnection.removeFailureListener(this);
@@ -1579,9 +1048,9 @@
// the replicating connection will cause the outstanding responses to be be replayed on the live server,
// if these reach the client who then subsequently fails over, on reconnection to backup, it will have
// received responses that the backup did not know about.
-
+
channel.transferConnection(newConnection);
-
+
newConnection.syncIDGeneratorSequence(remotingConnection.getIDGeneratorSequence());
remotingConnection = newConnection;
@@ -1591,10 +1060,10 @@
int serverLastReceivedCommandID = channel.getLastConfirmedCommandID();
- channel.replayCommands(lastReceivedCommandID, id);
-
+ channel.replayCommands(lastReceivedCommandID);
+
channel.setTransferring(false);
-
+
setTransferring(false);
return serverLastReceivedCommandID;
@@ -1604,6 +1073,21 @@
{
return channel;
}
+
+ public void runConnectionFailureRunners()
+ {
+ for (Runnable runner : failureRunners.values())
+ {
+ try
+ {
+ runner.run();
+ }
+ catch (Throwable t)
+ {
+ ServerSessionImpl.log.error("Failed to execute failure runner", t);
+ }
+ }
+ }
// FailureListener implementation
// --------------------------------------------------------------------
@@ -1626,7 +1110,7 @@
}
}
- handleClose(new PacketImpl(PacketImpl.SESS_CLOSE));
+ handleClose();
ServerSessionImpl.log.warn("Cleared up resources for session " + name);
}
@@ -1654,7 +1138,7 @@
}
catch (Throwable t)
{
- ServerSessionImpl.log.error("Failed fire listeners " + this);
+ ServerSessionImpl.log.error("Failed to fire listeners " + this);
}
}
@@ -1665,129 +1149,6 @@
// Private
// ----------------------------------------------------------------------------
- private SessionQueueQueryResponseMessage doExecuteQueueQuery(final SimpleString name) throws Exception
- {
- if (name == null)
- {
- throw new IllegalArgumentException("Queue name is null");
- }
-
- SessionQueueQueryResponseMessage response;
-
- Binding binding = postOffice.getBinding(name);
-
- if (binding != null && binding.getType() == BindingType.LOCAL_QUEUE)
- {
- Queue queue = (Queue)binding.getBindable();
-
- Filter filter = queue.getFilter();
-
- SimpleString filterString = filter == null ? null : filter.getFilterString();
-
- response = new SessionQueueQueryResponseMessage(name,
- binding.getAddress(),
- queue.isDurable(),
- queue.isTemporary(),
- filterString,
- queue.getConsumerCount(),
- queue.getMessageCount());
- }
- // make an exception for the management address (see HORNETQ-29)
- else if (name.equals(managementAddress))
- {
- response = new SessionQueueQueryResponseMessage(name, managementAddress, true, false, null, -1, -1);
- }
- else
- {
- response = new SessionQueueQueryResponseMessage();
- }
-
- return response;
- }
-
- private void sendResponse(final Packet confirmPacket,
- final Packet response,
- final boolean flush,
- final boolean closeChannel)
- {
- storageManager.afterCompleteOperations(new IOAsyncTask()
- {
- public void onError(final int errorCode, final String errorMessage)
- {
- ServerSessionImpl.log.warn("Error processing IOCallback code = " + errorCode + " message = " + errorMessage);
-
- HornetQExceptionMessage exceptionMessage = new HornetQExceptionMessage(new HornetQException(errorCode,
- errorMessage));
-
- doConfirmAndResponse(confirmPacket, exceptionMessage, flush, closeChannel);
- }
-
- public void done()
- {
- doConfirmAndResponse(confirmPacket, response, flush, closeChannel);
- }
- });
- }
-
- /**
- * @param confirmPacket
- * @param response
- * @param flush
- * @param closeChannel
- */
- private void doConfirmAndResponse(final Packet confirmPacket,
- final Packet response,
- final boolean flush,
- final boolean closeChannel)
- {
- if (confirmPacket != null)
- {
- channel.confirm(confirmPacket);
-
- if (flush)
- {
- channel.flushConfirmations();
- }
- }
-
- if (response != null)
- {
- channel.send(response);
- }
-
- if (closeChannel)
- {
- channel.close();
- }
- }
-
- private void doClose(final Packet packet)
- {
- Packet response = null;
-
- try
- {
- close();
-
- response = new NullResponseMessage();
- }
- catch (Exception e)
- {
- ServerSessionImpl.log.error("Failed to close", e);
-
- if (e instanceof HornetQException)
- {
- response = new HornetQExceptionMessage((HornetQException)e);
- }
- else
- {
- response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR));
- }
- }
-
- sendResponse(packet, response, true, true);
- }
-
private void setStarted(final boolean s)
{
Set<ServerConsumer> consumersClone = new HashSet<ServerConsumer>(consumers.values());
@@ -1800,44 +1161,6 @@
started = s;
}
- private void setTransferring(final boolean transferring)
- {
- Set<ServerConsumer> consumersClone = new HashSet<ServerConsumer>(consumers.values());
-
- for (ServerConsumer consumer : consumersClone)
- {
- consumer.setTransferring(transferring);
- }
- }
-
- /**
- * We need to create the LargeMessage before replicating the packet, or else we won't know how to extract the destination,
- * which is stored on the header
- * @param packet
- * @throws Exception
- */
- private LargeServerMessage doCreateLargeMessage(final long id, final SessionSendLargeMessage packet)
- {
- try
- {
- LargeServerMessage msg = createLargeMessageStorage(id, packet.getLargeMessageHeader());
-
- return msg;
- }
- catch (Exception e)
- {
- ServerSessionImpl.log.error("Failed to create large message", e);
- Packet response = null;
-
- channel.confirm(packet);
- if (response != null)
- {
- channel.send(response);
- }
- return null;
- }
- }
-
private void handleManagementMessage(final ServerMessage message) throws Exception
{
try
@@ -1865,11 +1188,6 @@
}
}
- private LargeServerMessage createLargeMessageStorage(final long id, final byte[] header) throws Exception
- {
- return storageManager.createLargeMessage(id, header);
- }
-
private void doRollback(final boolean lastMessageAsDelived, final Transaction theTx) throws Exception
{
boolean wasStarted = started;
@@ -1931,7 +1249,7 @@
{
releaseOutStanding(message.getAddress(), credits);
}
-
+
private void releaseOutStanding(final SimpleString address, final int credits) throws Exception
{
CreditManagerHolder holder = getCreditManagerHolder(address);
@@ -1961,9 +1279,7 @@
{
holder.outstandingCredits += credits;
- Packet packet = new SessionProducerCreditsMessage(credits, address, -1);
-
- channel.send(packet);
+ callback.sendProducerCreditsMessage(credits, address, -1);
}
private void send(final ServerMessage msg) throws Exception
Modified: trunk/src/main/org/hornetq/core/server/impl/ServerSessionPacketHandler.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/impl/ServerSessionPacketHandler.java 2010-01-20 13:24:55 UTC (rev 8810)
+++ trunk/src/main/org/hornetq/core/server/impl/ServerSessionPacketHandler.java 2010-01-20 14:38:53 UTC (rev 8811)
@@ -44,23 +44,44 @@
import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_XA_START;
import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_XA_SUSPEND;
+import java.util.List;
+
+import javax.transaction.xa.XAResource;
+import javax.transaction.xa.Xid;
+
+import org.hornetq.api.core.HornetQException;
+import org.hornetq.api.core.SimpleString;
+import org.hornetq.core.exception.HornetQXAException;
+import org.hornetq.core.journal.IOAsyncTask;
import org.hornetq.core.logging.Logger;
import org.hornetq.core.persistence.OperationContext;
import org.hornetq.core.persistence.StorageManager;
+import org.hornetq.core.remoting.Channel;
import org.hornetq.core.remoting.ChannelHandler;
+import org.hornetq.core.remoting.CloseListener;
+import org.hornetq.core.remoting.FailureListener;
import org.hornetq.core.remoting.Packet;
+import org.hornetq.core.remoting.RemotingConnection;
import org.hornetq.core.remoting.impl.wireformat.CreateQueueMessage;
+import org.hornetq.core.remoting.impl.wireformat.HornetQExceptionMessage;
+import org.hornetq.core.remoting.impl.wireformat.NullResponseMessage;
import org.hornetq.core.remoting.impl.wireformat.PacketImpl;
import org.hornetq.core.remoting.impl.wireformat.RollbackMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionAcknowledgeMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryMessage;
+import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionConsumerCloseMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionConsumerFlowCreditMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionCreateConsumerMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionDeleteQueueMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionExpiredMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionForceConsumerDelivery;
+import org.hornetq.core.remoting.impl.wireformat.SessionProducerCreditsMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryMessage;
+import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage;
+import org.hornetq.core.remoting.impl.wireformat.SessionReceiveContinuationMessage;
+import org.hornetq.core.remoting.impl.wireformat.SessionReceiveLargeMessage;
+import org.hornetq.core.remoting.impl.wireformat.SessionReceiveMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionRequestProducerCreditsMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionSendContinuationMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionSendLargeMessage;
@@ -68,13 +89,21 @@
import org.hornetq.core.remoting.impl.wireformat.SessionXACommitMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionXAEndMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionXAForgetMessage;
+import org.hornetq.core.remoting.impl.wireformat.SessionXAGetInDoubtXidsResponseMessage;
+import org.hornetq.core.remoting.impl.wireformat.SessionXAGetTimeoutResponseMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionXAJoinMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionXAPrepareMessage;
+import org.hornetq.core.remoting.impl.wireformat.SessionXAResponseMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionXAResumeMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionXARollbackMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionXASetTimeoutMessage;
+import org.hornetq.core.remoting.impl.wireformat.SessionXASetTimeoutResponseMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionXAStartMessage;
+import org.hornetq.core.server.BindingQueryResult;
+import org.hornetq.core.server.QueueQueryResult;
+import org.hornetq.core.server.ServerMessage;
import org.hornetq.core.server.ServerSession;
+import org.hornetq.core.server.SessionCallback;
/**
* A ServerSessionPacketHandler
@@ -84,7 +113,7 @@
* @author <a href="mailto:andy.taylor at jboss.org>Andy Taylor</a>
* @author <a href="mailto:clebert.suconic at jboss.org>Clebert Suconic</a>
*/
-public class ServerSessionPacketHandler implements ChannelHandler
+public class ServerSessionPacketHandler implements ChannelHandler, CloseListener, FailureListener, SessionCallback
{
private static final Logger log = Logger.getLogger(ServerSessionPacketHandler.class);
@@ -95,220 +124,486 @@
// Storagemanager here is used to set the Context
private final StorageManager storageManager;
+ private final Channel channel;
+
+ private volatile RemotingConnection remotingConnection;
+
public ServerSessionPacketHandler(final ServerSession session,
final OperationContext sessionContext,
- final StorageManager storageManager)
+ final StorageManager storageManager,
+ final Channel channel)
{
this.session = session;
this.storageManager = storageManager;
this.sessionContext = sessionContext;
+
+ this.channel = channel;
+
+ this.remotingConnection = channel.getConnection();
+
+ addConnectionListeners();
}
public long getID()
{
- return session.getID();
+ return channel.getID();
}
+ public void connectionFailed(final HornetQException exception)
+ {
+ log.warn("Client connection failed, clearing up resources for session " + session.getName());
+
+ session.runConnectionFailureRunners();
+
+ handleCloseSession();
+
+ log.warn("Cleared up resources for session " + session.getName());
+ }
+
+ public void close()
+ {
+ channel.flushConfirmations();
+
+ handleCloseSession();
+ }
+
+ public void connectionClosed()
+ {
+ session.runConnectionFailureRunners();
+ }
+
+ private void addConnectionListeners()
+ {
+ remotingConnection.addFailureListener(this);
+ remotingConnection.addCloseListener(this);
+ }
+
+ private void removeConnectionListeners()
+ {
+ remotingConnection.removeFailureListener(this);
+ remotingConnection.removeCloseListener(this);
+ }
+
+ public Channel getChannel()
+ {
+ return channel;
+ }
+
+ public int transferConnection(final RemotingConnection newConnection, final int lastReceivedCommandID)
+ {
+ // We need to disable delivery on all the consumers while the transfer is occurring- otherwise packets might get
+ // delivered
+ // after the channel has transferred but *before* packets have been replayed - this will give the client the wrong
+ // sequence of packets.
+ // It is not sufficient to just stop the session, since right after stopping the session, another session start
+ // might be executed
+ // before we have transferred the connection, leaving it in a started state
+ session.setTransferring(true);
+
+ removeConnectionListeners();
+
+ // Note. We do not destroy the replicating connection here. In the case the live server has really crashed
+ // then the connection will get cleaned up anyway when the server ping timeout kicks in.
+ // In the case the live server is really still up, i.e. a split brain situation (or in tests), then closing
+ // the replicating connection will cause the outstanding responses to be be replayed on the live server,
+ // if these reach the client who then subsequently fails over, on reconnection to backup, it will have
+ // received responses that the backup did not know about.
+
+ channel.transferConnection(newConnection);
+
+ newConnection.syncIDGeneratorSequence(remotingConnection.getIDGeneratorSequence());
+
+ remotingConnection = newConnection;
+
+ addConnectionListeners();
+
+ int serverLastReceivedCommandID = channel.getLastConfirmedCommandID();
+
+ channel.replayCommands(lastReceivedCommandID);
+
+ channel.setTransferring(false);
+
+ session.setTransferring(false);
+
+ return serverLastReceivedCommandID;
+ }
+
public void handlePacket(final Packet packet)
{
byte type = packet.getType();
storageManager.setContext(sessionContext);
+ Packet response = null;
+ boolean flush = false;
+ boolean closeChannel = false;
+
try
{
- switch (type)
+ try
{
- case SESS_CREATECONSUMER:
+ switch (type)
{
- SessionCreateConsumerMessage request = (SessionCreateConsumerMessage)packet;
- session.handleCreateConsumer(request);
- break;
+ case SESS_CREATECONSUMER:
+ {
+ SessionCreateConsumerMessage request = (SessionCreateConsumerMessage)packet;
+ session.handleCreateConsumer(request.getID(),
+ request.getQueueName(),
+ request.getFilterString(),
+ request.isBrowseOnly());
+ if (request.isRequiresResponse())
+ {
+ // We send back queue information on the queue as a response- this allows the queue to
+ // be automaticall recreated on failover
+ response = new SessionQueueQueryResponseMessage(session.handleExecuteQueueQuery(request.getQueueName()));
+ }
+
+ break;
+ }
+ case CREATE_QUEUE:
+ {
+ CreateQueueMessage request = (CreateQueueMessage)packet;
+ session.handleCreateQueue(request.getAddress(),
+ request.getQueueName(),
+ request.getFilterString(),
+ request.isTemporary(),
+ request.isDurable());
+ if (request.isRequiresResponse())
+ {
+ response = new NullResponseMessage();
+ }
+ break;
+ }
+ case DELETE_QUEUE:
+ {
+ SessionDeleteQueueMessage request = (SessionDeleteQueueMessage)packet;
+ session.handleDeleteQueue(request.getQueueName());
+ response = new NullResponseMessage();
+ break;
+ }
+ case SESS_QUEUEQUERY:
+ {
+ SessionQueueQueryMessage request = (SessionQueueQueryMessage)packet;
+ QueueQueryResult result = session.handleExecuteQueueQuery(request.getQueueName());
+ response = new SessionQueueQueryResponseMessage(result);
+ break;
+ }
+ case SESS_BINDINGQUERY:
+ {
+ SessionBindingQueryMessage request = (SessionBindingQueryMessage)packet;
+ BindingQueryResult result = session.handleExecuteBindingQuery(request.getAddress());
+ response = new SessionBindingQueryResponseMessage(result.isExists(), result.getQueueNames());
+ break;
+ }
+ case SESS_ACKNOWLEDGE:
+ {
+ SessionAcknowledgeMessage message = (SessionAcknowledgeMessage)packet;
+ session.handleAcknowledge(message.getConsumerID(), message.getMessageID());
+ if (message.isRequiresResponse())
+ {
+ response = new NullResponseMessage();
+ }
+ break;
+ }
+ case SESS_EXPIRED:
+ {
+ SessionExpiredMessage message = (SessionExpiredMessage)packet;
+ session.handleExpired(message.getConsumerID(), message.getMessageID());
+ break;
+ }
+ case SESS_COMMIT:
+ {
+ session.handleCommit();
+ response = new NullResponseMessage();
+ break;
+ }
+ case SESS_ROLLBACK:
+ {
+ session.handleRollback(((RollbackMessage)packet).isConsiderLastMessageAsDelivered());
+ response = new NullResponseMessage();
+ break;
+ }
+ case SESS_XA_COMMIT:
+ {
+ SessionXACommitMessage message = (SessionXACommitMessage)packet;
+ session.handleXACommit(message.getXid(), message.isOnePhase());
+ response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
+ break;
+ }
+ case SESS_XA_END:
+ {
+ SessionXAEndMessage message = (SessionXAEndMessage)packet;
+ session.handleXAEnd(message.getXid());
+ response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
+ break;
+ }
+ case SESS_XA_FORGET:
+ {
+ SessionXAForgetMessage message = (SessionXAForgetMessage)packet;
+ session.handleXAForget(message.getXid());
+ response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
+ break;
+ }
+ case SESS_XA_JOIN:
+ {
+ SessionXAJoinMessage message = (SessionXAJoinMessage)packet;
+ session.handleXAJoin(message.getXid());
+ response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
+ break;
+ }
+ case SESS_XA_RESUME:
+ {
+ SessionXAResumeMessage message = (SessionXAResumeMessage)packet;
+ session.handleXAResume(message.getXid());
+ response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
+ break;
+ }
+ case SESS_XA_ROLLBACK:
+ {
+ SessionXARollbackMessage message = (SessionXARollbackMessage)packet;
+ session.handleXARollback(message.getXid());
+ response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
+ break;
+ }
+ case SESS_XA_START:
+ {
+ SessionXAStartMessage message = (SessionXAStartMessage)packet;
+ session.handleXAStart(message.getXid());
+ response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
+ break;
+ }
+ case SESS_XA_SUSPEND:
+ {
+ session.handleXASuspend();
+ response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
+ break;
+ }
+ case SESS_XA_PREPARE:
+ {
+ SessionXAPrepareMessage message = (SessionXAPrepareMessage)packet;
+ session.handleXAPrepare(message.getXid());
+ response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
+ break;
+ }
+ case SESS_XA_INDOUBT_XIDS:
+ {
+ List<Xid> xids = session.handleGetInDoubtXids();
+ response = new SessionXAGetInDoubtXidsResponseMessage(xids);
+ break;
+ }
+ case SESS_XA_GET_TIMEOUT:
+ {
+ int timeout = session.handleGetXATimeout();
+ response = new SessionXAGetTimeoutResponseMessage(timeout);
+ break;
+ }
+ case SESS_XA_SET_TIMEOUT:
+ {
+ SessionXASetTimeoutMessage message = (SessionXASetTimeoutMessage)packet;
+ session.handleSetXATimeout(message.getTimeoutSeconds());
+ response = new SessionXASetTimeoutResponseMessage(true);
+ break;
+ }
+ case SESS_START:
+ {
+ session.handleStart();
+ break;
+ }
+ case SESS_STOP:
+ {
+ session.handleStop();
+ response = new NullResponseMessage();
+ break;
+ }
+ case SESS_CLOSE:
+ {
+ handleCloseSession();
+ removeConnectionListeners();
+ response = new NullResponseMessage();
+ flush = true;
+ closeChannel = true;
+ break;
+ }
+ case SESS_CONSUMER_CLOSE:
+ {
+ SessionConsumerCloseMessage message = (SessionConsumerCloseMessage)packet;
+ session.handleCloseConsumer(message.getConsumerID());
+ response = new NullResponseMessage();
+ break;
+ }
+ case SESS_FLOWTOKEN:
+ {
+ SessionConsumerFlowCreditMessage message = (SessionConsumerFlowCreditMessage)packet;
+ session.handleReceiveConsumerCredits(message.getConsumerID(), message.getCredits());
+ break;
+ }
+ case SESS_SEND:
+ {
+ SessionSendMessage message = (SessionSendMessage)packet;
+ session.handleSend((ServerMessage)message.getMessage());
+ if (message.isRequiresResponse())
+ {
+ response = new NullResponseMessage();
+ }
+ break;
+ }
+ case SESS_SEND_LARGE:
+ {
+ SessionSendLargeMessage message = (SessionSendLargeMessage)packet;
+ session.handleSendLargeMessage(message.getLargeMessageHeader());
+ break;
+ }
+ case SESS_SEND_CONTINUATION:
+ {
+ SessionSendContinuationMessage message = (SessionSendContinuationMessage)packet;
+ session.handleSendContinuations(message.getPacketSize(), message.getBody(), message.isContinues());
+ if (message.isRequiresResponse())
+ {
+ response = new NullResponseMessage();
+ }
+ break;
+ }
+ case SESS_FORCE_CONSUMER_DELIVERY:
+ {
+ SessionForceConsumerDelivery message = (SessionForceConsumerDelivery)packet;
+ session.handleForceConsumerDelivery(message.getConsumerID(), message.getSequence());
+ break;
+ }
+ case PacketImpl.SESS_PRODUCER_REQUEST_CREDITS:
+ {
+ SessionRequestProducerCreditsMessage message = (SessionRequestProducerCreditsMessage)packet;
+ session.handleRequestProducerCredits(message.getAddress(), message.getCredits());
+ break;
+ }
}
- case CREATE_QUEUE:
- {
- CreateQueueMessage request = (CreateQueueMessage)packet;
- session.handleCreateQueue(request);
- break;
- }
- case DELETE_QUEUE:
- {
- SessionDeleteQueueMessage request = (SessionDeleteQueueMessage)packet;
- session.handleDeleteQueue(request);
- break;
- }
- case SESS_QUEUEQUERY:
- {
- SessionQueueQueryMessage request = (SessionQueueQueryMessage)packet;
- session.handleExecuteQueueQuery(request);
- break;
- }
- case SESS_BINDINGQUERY:
- {
- SessionBindingQueryMessage request = (SessionBindingQueryMessage)packet;
- session.handleExecuteBindingQuery(request);
- break;
- }
- case SESS_ACKNOWLEDGE:
- {
- SessionAcknowledgeMessage message = (SessionAcknowledgeMessage)packet;
- session.handleAcknowledge(message);
- break;
- }
- case SESS_EXPIRED:
- {
- SessionExpiredMessage message = (SessionExpiredMessage)packet;
- session.handleExpired(message);
- break;
- }
- case SESS_COMMIT:
- {
- session.handleCommit(packet);
- break;
- }
- case SESS_ROLLBACK:
- {
- session.handleRollback((RollbackMessage)packet);
- break;
- }
- case SESS_XA_COMMIT:
- {
- SessionXACommitMessage message = (SessionXACommitMessage)packet;
- session.handleXACommit(message);
- break;
- }
- case SESS_XA_END:
- {
- SessionXAEndMessage message = (SessionXAEndMessage)packet;
- session.handleXAEnd(message);
- break;
- }
- case SESS_XA_FORGET:
- {
- SessionXAForgetMessage message = (SessionXAForgetMessage)packet;
- session.handleXAForget(message);
- break;
- }
- case SESS_XA_JOIN:
- {
- SessionXAJoinMessage message = (SessionXAJoinMessage)packet;
- session.handleXAJoin(message);
- break;
- }
- case SESS_XA_RESUME:
- {
- SessionXAResumeMessage message = (SessionXAResumeMessage)packet;
- session.handleXAResume(message);
- break;
- }
- case SESS_XA_ROLLBACK:
- {
- SessionXARollbackMessage message = (SessionXARollbackMessage)packet;
- session.handleXARollback(message);
- break;
- }
- case SESS_XA_START:
- {
- SessionXAStartMessage message = (SessionXAStartMessage)packet;
- session.handleXAStart(message);
- break;
- }
- case SESS_XA_SUSPEND:
- {
- session.handleXASuspend(packet);
- break;
- }
- case SESS_XA_PREPARE:
- {
- SessionXAPrepareMessage message = (SessionXAPrepareMessage)packet;
- session.handleXAPrepare(message);
- break;
- }
- case SESS_XA_INDOUBT_XIDS:
- {
- session.handleGetInDoubtXids(packet);
- break;
- }
- case SESS_XA_GET_TIMEOUT:
- {
- session.handleGetXATimeout(packet);
- break;
- }
- case SESS_XA_SET_TIMEOUT:
- {
- SessionXASetTimeoutMessage message = (SessionXASetTimeoutMessage)packet;
- session.handleSetXATimeout(message);
- break;
- }
- case SESS_START:
- {
- session.handleStart(packet);
- break;
- }
- case SESS_STOP:
- {
- session.handleStop(packet);
- break;
- }
- case SESS_CLOSE:
- {
- session.handleClose(packet);
- break;
- }
- case SESS_CONSUMER_CLOSE:
- {
- SessionConsumerCloseMessage message = (SessionConsumerCloseMessage)packet;
- session.handleCloseConsumer(message);
- break;
- }
- case SESS_FLOWTOKEN:
- {
- SessionConsumerFlowCreditMessage message = (SessionConsumerFlowCreditMessage)packet;
- session.handleReceiveConsumerCredits(message);
- break;
- }
- case SESS_SEND:
- {
- SessionSendMessage message = (SessionSendMessage)packet;
- session.handleSend(message);
- break;
- }
- case SESS_SEND_LARGE:
- {
- SessionSendLargeMessage message = (SessionSendLargeMessage)packet;
- session.handleSendLargeMessage(message);
- break;
- }
- case SESS_SEND_CONTINUATION:
- {
- SessionSendContinuationMessage message = (SessionSendContinuationMessage)packet;
- session.handleSendContinuations(message);
- break;
- }
- case SESS_FORCE_CONSUMER_DELIVERY:
- {
- SessionForceConsumerDelivery message = (SessionForceConsumerDelivery)packet;
- session.handleForceConsumerDelivery(message);
- break;
- }
- case PacketImpl.SESS_PRODUCER_REQUEST_CREDITS:
- {
- SessionRequestProducerCreditsMessage message = (SessionRequestProducerCreditsMessage)packet;
- session.handleRequestProducerCredits(message);
- break;
- }
}
+ catch (HornetQXAException e)
+ {
+ response = new SessionXAResponseMessage(true, e.errorCode, e.getMessage());
+ }
+ catch (HornetQException e)
+ {
+ response = new HornetQExceptionMessage((HornetQException)e);
+ }
+ catch (Throwable t)
+ {
+ log.error("Caught unexpected exception", t);
+ }
+
+ sendResponse(packet, response, flush, closeChannel);
}
- catch (Throwable t)
- {
- ServerSessionPacketHandler.log.error("Caught unexpected exception", t);
- }
finally
{
storageManager.completeOperations();
storageManager.clearContext();
}
}
+
+ private void sendResponse(final Packet confirmPacket,
+ final Packet response,
+ final boolean flush,
+ final boolean closeChannel)
+ {
+ storageManager.afterCompleteOperations(new IOAsyncTask()
+ {
+ public void onError(final int errorCode, final String errorMessage)
+ {
+ log.warn("Error processing IOCallback code = " + errorCode + " message = " + errorMessage);
+
+ HornetQExceptionMessage exceptionMessage = new HornetQExceptionMessage(new HornetQException(errorCode,
+ errorMessage));
+
+ doConfirmAndResponse(confirmPacket, exceptionMessage, flush, closeChannel);
+ }
+
+ public void done()
+ {
+ doConfirmAndResponse(confirmPacket, response, flush, closeChannel);
+ }
+ });
+ }
+
+ private void doConfirmAndResponse(final Packet confirmPacket,
+ final Packet response,
+ final boolean flush,
+ final boolean closeChannel)
+ {
+ if (confirmPacket != null)
+ {
+ channel.confirm(confirmPacket);
+
+ if (flush)
+ {
+ channel.flushConfirmations();
+ }
+ }
+
+ if (response != null)
+ {
+ channel.send(response);
+ }
+
+ if (closeChannel)
+ {
+ channel.close();
+ }
+ }
+
+ private void handleCloseSession()
+ {
+ storageManager.afterCompleteOperations(new IOAsyncTask()
+ {
+ public void onError(int errorCode, String errorMessage)
+ {
+ }
+
+ public void done()
+ {
+ try
+ {
+ session.close();
+ }
+ catch (Exception e)
+ {
+ log.error("Failed to close session", e);
+ }
+ }
+ });
+ }
+
+ public int sendLargeMessage(long consumerID, byte[] headerBuffer, long bodySize, int deliveryCount)
+ {
+ Packet packet = new SessionReceiveLargeMessage(consumerID, headerBuffer, bodySize, deliveryCount);
+
+ channel.send(packet);
+
+ return packet.getPacketSize();
+ }
+
+ public int sendLargeMessageContinuation(long consumerID, byte[] body, boolean continues, boolean requiresResponse)
+ {
+ Packet packet = new SessionReceiveContinuationMessage(consumerID, body, continues, requiresResponse);
+
+ channel.send(packet);
+
+ return packet.getPacketSize();
+ }
+
+ public int sendMessage(ServerMessage message, long consumerID, int deliveryCount)
+ {
+ Packet packet = new SessionReceiveMessage(consumerID, message, deliveryCount);
+
+ channel.send(packet);
+
+ return packet.getPacketSize();
+ }
+
+ public void sendProducerCreditsMessage(int credits, SimpleString address, int offset)
+ {
+ Packet packet = new SessionProducerCreditsMessage(credits, address, offset);
+
+ channel.send(packet);
+ }
}
Modified: trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java 2010-01-20 13:24:55 UTC (rev 8810)
+++ trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java 2010-01-20 14:38:53 UTC (rev 8811)
@@ -152,8 +152,7 @@
ClientSessionFactory sf = createInVMFactory();
if (sendingBlocking)
- {
- sf.setBlockOnNonDurableSend(true);
+ { sf.setBlockOnNonDurableSend(true);
sf.setBlockOnDurableSend(true);
sf.setBlockOnAcknowledge(true);
}
@@ -523,7 +522,7 @@
}
session.close();
-
+
long globalSize = server.getPostOffice().getPagingManager().getTotalMemory();
Assert.assertEquals(0l, globalSize);
Assert.assertEquals(0, ((Queue)server.getPostOffice().getBinding(ADDRESS).getBindable()).getDeliveringCount());
Modified: trunk/tests/src/org/hornetq/tests/integration/xa/BasicXaTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/xa/BasicXaTest.java 2010-01-20 13:24:55 UTC (rev 8810)
+++ trunk/tests/src/org/hornetq/tests/integration/xa/BasicXaTest.java 2010-01-20 14:38:53 UTC (rev 8811)
@@ -160,6 +160,8 @@
clientSession = sessionFactory.createSession(true, false, false);
+ log.info("committing");
+
clientSession.commit(xid, false);
clientSession.start();
ClientConsumer clientConsumer = clientSession.createConsumer(atestq);
More information about the hornetq-commits
mailing list