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@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@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@jboss.org>Andy Taylor</a> * @author <a href="mailto:clebert.suconic@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);