diff options
Diffstat (limited to 'qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server')
12 files changed, 1068 insertions, 192 deletions
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java index 89d681111b..5affe3019c 100644 --- a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java +++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java @@ -104,7 +104,8 @@ public class ConsumerTarget_0_10 extends AbstractConsumerTarget implements FlowC _name = name; } - public boolean isSuspended() + @Override + public boolean doIsSuspended() { return getState()!=State.ACTIVE || _deleted.get() || _session.isClosing() || _session.getConnectionModel().isStopped(); // TODO check for Session suspension } @@ -158,6 +159,10 @@ public class ConsumerTarget_0_10 extends AbstractConsumerTarget implements FlowC return _name; } + public void transportStateChanged() + { + _creditManager.restoreCredit(0, 0); + } public static class AddMessageDispositionListenerAction implements Runnable { @@ -191,7 +196,7 @@ public class ConsumerTarget_0_10 extends AbstractConsumerTarget implements FlowC private final AddMessageDispositionListenerAction _postIdSettingAction; - public long send(final ConsumerImpl consumer, final MessageInstance entry, boolean batch) + public void doSend(final ConsumerImpl consumer, final MessageInstance entry, boolean batch) { ServerMessage serverMsg = entry.getMessage(); @@ -342,7 +347,6 @@ public class ConsumerTarget_0_10 extends AbstractConsumerTarget implements FlowC { recordUnacknowledged(entry); } - return size; } void recordUnacknowledged(MessageInstance entry) @@ -555,10 +559,10 @@ public class ConsumerTarget_0_10 extends AbstractConsumerTarget implements FlowC switch(flowMode) { case CREDIT: - _creditManager = new CreditCreditManager(0l,0l); + _creditManager = new CreditCreditManager(0l, 0l, _session.getConnection().getProtocolEngine()); break; case WINDOW: - _creditManager = new WindowCreditManager(0l,0l); + _creditManager = new WindowCreditManager(0l, 0l, _session.getConnection().getProtocolEngine()); break; default: // this should never happen, as 0-10 is finalised and so the enum should never change @@ -628,7 +632,6 @@ public class ConsumerTarget_0_10 extends AbstractConsumerTarget implements FlowC public void flushBatched() { - _session.getConnection().flush(); } @@ -657,4 +660,10 @@ public class ConsumerTarget_0_10 extends AbstractConsumerTarget implements FlowC { return _unacknowledgedCount.longValue(); } + + @Override + protected void processClosed() + { + + } } diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/CreditCreditManager.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/CreditCreditManager.java index 8dddac9809..dd43ae7e11 100644 --- a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/CreditCreditManager.java +++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/CreditCreditManager.java @@ -21,48 +21,27 @@ package org.apache.qpid.server.protocol.v0_10; +import org.apache.qpid.server.protocol.ServerProtocolEngine; import org.apache.qpid.server.flow.AbstractFlowCreditManager; public class CreditCreditManager extends AbstractFlowCreditManager implements FlowCreditManager_0_10 { + private final ServerProtocolEngine _serverProtocolEngine; private volatile long _bytesCredit; private volatile long _messageCredit; - public CreditCreditManager(long bytesCredit, long messageCredit) + public CreditCreditManager(long bytesCredit, long messageCredit, final ServerProtocolEngine serverProtocolEngine) { + _serverProtocolEngine = serverProtocolEngine; _bytesCredit = bytesCredit; _messageCredit = messageCredit; setSuspended(!hasCredit()); } - - public synchronized void setCreditLimits(final long bytesCredit, final long messageCredit) - { - _bytesCredit = bytesCredit; - _messageCredit = messageCredit; - - setSuspended(!hasCredit()); - - } - - - public long getMessageCredit() - { - return _messageCredit == -1L - ? Long.MAX_VALUE - : _messageCredit; - } - - public long getBytesCredit() - { - return _bytesCredit == -1L - ? Long.MAX_VALUE - : _bytesCredit; - } - public synchronized void restoreCredit(final long messageCredit, final long bytesCredit) { + setSuspended(!hasCredit()); } @@ -107,12 +86,17 @@ public class CreditCreditManager extends AbstractFlowCreditManager implements Fl public synchronized boolean hasCredit() { // Note !=, if credit is < 0 that indicates infinite credit - return (_bytesCredit != 0L && _messageCredit != 0L); + return (_bytesCredit != 0L && _messageCredit != 0L && !_serverProtocolEngine.isTransportBlockedForWriting()); } public synchronized boolean useCreditForMessage(long msgSize) { - if(_messageCredit >= 0L) + if (_serverProtocolEngine.isTransportBlockedForWriting()) + { + setSuspended(true); + return false; + } + else if(_messageCredit >= 0L) { if(_messageCredit > 0) { diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngineCreator_0_10.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngineCreator_0_10.java index 30aecdb2d2..4231045afd 100644 --- a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngineCreator_0_10.java +++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngineCreator_0_10.java @@ -23,7 +23,7 @@ package org.apache.qpid.server.protocol.v0_10; import java.net.InetSocketAddress; import java.net.SocketAddress; -import org.apache.qpid.protocol.ServerProtocolEngine; +import org.apache.qpid.server.protocol.ServerProtocolEngine; import org.apache.qpid.server.model.Broker; import org.apache.qpid.server.model.Protocol; import org.apache.qpid.server.model.Transport; @@ -86,7 +86,10 @@ public class ProtocolEngineCreator_0_10 implements ProtocolEngineCreator conn.setRemoteAddress(network.getRemoteAddress()); conn.setLocalAddress(network.getLocalAddress()); - return new ProtocolEngine_0_10( conn, network); + ProtocolEngine_0_10 protocolEngine = new ProtocolEngine_0_10(conn, network); + conn.setProtocolEngine(protocolEngine); + + return protocolEngine; } diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java index 854cd388b9..e391bd6771 100755 --- a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java +++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java @@ -24,18 +24,23 @@ import java.net.SocketAddress; import java.nio.ByteBuffer; import java.security.AccessController; import java.security.PrivilegedAction; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import javax.security.auth.Subject; import org.apache.log4j.Logger; -import org.apache.qpid.protocol.ServerProtocolEngine; +import org.apache.qpid.server.protocol.ServerProtocolEngine; +import org.apache.qpid.server.consumer.ConsumerImpl; import org.apache.qpid.server.logging.messages.ConnectionMessages; +import org.apache.qpid.server.model.Consumer; import org.apache.qpid.server.model.Port; +import org.apache.qpid.server.protocol.AMQSessionModel; +import org.apache.qpid.server.util.Action; +import org.apache.qpid.transport.ByteBufferSender; import org.apache.qpid.transport.Constant; -import org.apache.qpid.transport.Sender; import org.apache.qpid.transport.network.Assembler; -import org.apache.qpid.transport.network.Disassembler; import org.apache.qpid.transport.network.InputHandler; import org.apache.qpid.transport.network.NetworkConnection; @@ -52,13 +57,20 @@ public class ProtocolEngine_0_10 extends InputHandler implements ServerProtocol private ServerConnection _connection; private long _createTime = System.currentTimeMillis(); - private long _lastReadTime; - private long _lastWriteTime; + private long _lastReadTime = _createTime; + private long _lastWriteTime = _createTime; + private volatile boolean _transportBlockedForWriting; + + private final AtomicReference<Thread> _messageAssignmentSuspended = new AtomicReference<>(); + + private final AtomicBoolean _stateChanged = new AtomicBoolean(); + private final AtomicReference<Action<ServerProtocolEngine>> _workListener = new AtomicReference<>(); + public ProtocolEngine_0_10(ServerConnection conn, NetworkConnection network) { - super(new Assembler(conn)); + super(new ServerAssembler(conn)); _connection = conn; if(network != null) @@ -67,7 +79,33 @@ public class ProtocolEngine_0_10 extends InputHandler implements ServerProtocol } } - public void setNetworkConnection(final NetworkConnection network, final Sender<ByteBuffer> sender) + @Override + public boolean isMessageAssignmentSuspended() + { + Thread lock = _messageAssignmentSuspended.get(); + return lock != null && _messageAssignmentSuspended.get() != Thread.currentThread(); + } + + @Override + public void setMessageAssignmentSuspended(final boolean messageAssignmentSuspended) + { + _messageAssignmentSuspended.set(messageAssignmentSuspended ? Thread.currentThread() : null); + + if(!messageAssignmentSuspended) + { + for(AMQSessionModel<?,?> session : _connection.getSessionModels()) + { + for(Consumer<?> consumer : session.getConsumers()) + { + ((ConsumerImpl)consumer).getTarget().notifyCurrentState(); + } + } + } + } + + + + public void setNetworkConnection(final NetworkConnection network, final ByteBufferSender sender) { if(!getSubject().equals(Subject.getSubject(AccessController.getContext()))) { @@ -87,7 +125,7 @@ public class ProtocolEngine_0_10 extends InputHandler implements ServerProtocol _network = network; _connection.setNetworkConnection(network); - Disassembler disassembler = new Disassembler(wrapSender(sender), Constant.MIN_MAX_FRAME_SIZE); + ServerDisassembler disassembler = new ServerDisassembler(wrapSender(sender), Constant.MIN_MAX_FRAME_SIZE); _connection.setSender(disassembler); _connection.addFrameSizeObserver(disassembler); // FIXME Two log messages to maintain compatibility with earlier protocol versions @@ -96,23 +134,15 @@ public class ProtocolEngine_0_10 extends InputHandler implements ServerProtocol } } - private Sender<ByteBuffer> wrapSender(final Sender<ByteBuffer> sender) + private ByteBufferSender wrapSender(final ByteBufferSender sender) { - return new Sender<ByteBuffer>() + return new ByteBufferSender() { @Override - public void setIdleTimeout(int i) - { - sender.setIdleTimeout(i); - - } - - @Override public void send(ByteBuffer msg) { _lastWriteTime = System.currentTimeMillis(); sender.send(msg); - } @Override @@ -190,6 +220,11 @@ public class ProtocolEngine_0_10 extends InputHandler implements ServerProtocol return _writtenBytes; } + @Override + public void encryptedTransport() + { + } + public void writerIdle() { _connection.doHeartBeat(); @@ -215,11 +250,6 @@ public class ProtocolEngine_0_10 extends InputHandler implements ServerProtocol return getRemoteAddress().toString(); } - public String getAuthId() - { - return _connection.getAuthorizedPrincipal() == null ? null : _connection.getAuthorizedPrincipal().getName(); - } - public boolean isDurable() { return false; @@ -246,4 +276,54 @@ public class ProtocolEngine_0_10 extends InputHandler implements ServerProtocol { return _connection.getAuthorizedSubject(); } + + @Override + public boolean isTransportBlockedForWriting() + { + return _transportBlockedForWriting; + } + + @Override + public void setTransportBlockedForWriting(final boolean blocked) + { + _transportBlockedForWriting = blocked; + _connection.transportStateChanged(); + } + + @Override + public void processPending() + { + _connection.processPending(); + + } + + @Override + public boolean hasWork() + { + return _stateChanged.get(); + } + + @Override + public void notifyWork() + { + _stateChanged.set(true); + + final Action<ServerProtocolEngine> listener = _workListener.get(); + if(listener != null) + { + listener.performAction(this); + } + } + + @Override + public void clearWork() + { + _stateChanged.set(false); + } + + @Override + public void setWorkListener(final Action<ServerProtocolEngine> listener) + { + _workListener.set(listener); + } } diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerAssembler.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerAssembler.java new file mode 100644 index 0000000000..456c9d36d9 --- /dev/null +++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerAssembler.java @@ -0,0 +1,57 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF 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.apache.qpid.server.protocol.v0_10; + + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.qpid.transport.network.Assembler; +import org.apache.qpid.transport.network.NetworkEvent; + +public class ServerAssembler extends Assembler +{ + private static final Logger LOGGER = LoggerFactory.getLogger(ServerAssembler.class); + + + private final ServerConnection _connection; + + public ServerAssembler(final ServerConnection connection) + { + super(connection); + _connection = connection; + } + + @Override + public void received(final NetworkEvent event) + { + if (!_connection.isIgnoreFutureInput()) + { + super.received(event); + } + else + { + LOGGER.debug("Ignored network event " + event + " as connection is ignoring further input "); + } + } + + +} diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java index 8567be37f0..2280377fca 100644 --- a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java +++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java @@ -23,6 +23,7 @@ package org.apache.qpid.server.protocol.v0_10; import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.CONNECTION_FORMAT; import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.SOCKET_FORMAT; import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.USER_FORMAT; +import static org.apache.qpid.transport.Connection.State.CLOSING; import java.net.SocketAddress; import java.security.Principal; @@ -30,6 +31,8 @@ import java.security.PrivilegedAction; import java.text.MessageFormat; import java.util.ArrayList; import java.util.List; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; @@ -37,6 +40,7 @@ import java.util.concurrent.atomic.AtomicLong; import javax.security.auth.Subject; import org.apache.qpid.protocol.AMQConstant; +import org.apache.qpid.server.protocol.ServerProtocolEngine; import org.apache.qpid.server.connection.ConnectionPrincipal; import org.apache.qpid.server.logging.EventLogger; import org.apache.qpid.server.logging.LogSubject; @@ -66,7 +70,6 @@ public class ServerConnection extends Connection implements AMQConnectionModel<S { private final Broker<?> _broker; - private Runnable _onOpenTask; private AtomicBoolean _logClosed = new AtomicBoolean(false); private final Subject _authorizedSubject = new Subject(); @@ -75,20 +78,26 @@ public class ServerConnection extends Connection implements AMQConnectionModel<S private final long _connectionId; private final Object _reference = new Object(); private VirtualHostImpl<?,?,?> _virtualHost; - private AmqpPort<?> _port; - private AtomicLong _lastIoTime = new AtomicLong(); + private final AmqpPort<?> _port; + private final AtomicLong _lastIoTime = new AtomicLong(); private boolean _blocking; - private Transport _transport; + private final Transport _transport; - private final CopyOnWriteArrayList<Action<? super ServerConnection>> _taskList = + private final CopyOnWriteArrayList<Action<? super ServerConnection>> _connectionCloseTaskList = new CopyOnWriteArrayList<Action<? super ServerConnection>>(); + private final Queue<Action<? super ServerConnection>> _asyncTaskList = + new ConcurrentLinkedQueue<>(); + private final CopyOnWriteArrayList<SessionModelListener> _sessionListeners = new CopyOnWriteArrayList<SessionModelListener>(); private volatile boolean _stopped; private int _messageCompressionThreshold; - private int _maxMessageSize; + private final int _maxMessageSize; + + private ServerProtocolEngine _serverProtocolEngine; + private boolean _ignoreFutureInput; public ServerConnection(final long connectionId, Broker<?> broker, @@ -140,10 +149,6 @@ public class ServerConnection extends Connection implements AMQConnectionModel<S if (state == State.OPEN) { - if (_onOpenTask != null) - { - _onOpenTask.run(); - } getEventLogger().message(ConnectionMessages.OPEN(getClientId(), "0-10", getClientVersion(), @@ -189,6 +194,16 @@ public class ServerConnection extends Connection implements AMQConnectionModel<S super.setConnectionDelegate(delegate); } + public ServerProtocolEngine getProtocolEngine() + { + return _serverProtocolEngine; + } + + public void setProtocolEngine(final ServerProtocolEngine serverProtocolEngine) + { + _serverProtocolEngine = serverProtocolEngine; + } + public VirtualHostImpl<?,?,?> getVirtualHost() { return _virtualHost; @@ -237,28 +252,32 @@ public class ServerConnection extends Connection implements AMQConnectionModel<S return _stopped; } - public void onOpen(final Runnable task) - { - _onOpenTask = task; - } - - public void closeSession(ServerSession session, AMQConstant cause, String message) + public void closeSessionAsync(final ServerSession session, final AMQConstant cause, final String message) { - ExecutionException ex = new ExecutionException(); - ExecutionErrorCode code = ExecutionErrorCode.INTERNAL_ERROR; - try - { - code = ExecutionErrorCode.get(cause.getCode()); - } - catch (IllegalArgumentException iae) + addAsyncTask(new Action<ServerConnection>() { - // Ignore, already set to INTERNAL_ERROR - } - ex.setErrorCode(code); - ex.setDescription(message); - session.invoke(ex); - session.close(cause, message); + @Override + public void performAction(final ServerConnection conn) + { + ExecutionException ex = new ExecutionException(); + ExecutionErrorCode code = ExecutionErrorCode.INTERNAL_ERROR; + try + { + code = ExecutionErrorCode.get(cause.getCode()); + } + catch (IllegalArgumentException iae) + { + // Ignore, already set to INTERNAL_ERROR + } + ex.setErrorCode(code); + ex.setDescription(message); + session.invoke(ex); + + session.close(cause, message); + } + }); + } public LogSubject getLogSubject() @@ -355,25 +374,35 @@ public class ServerConnection extends Connection implements AMQConnectionModel<S } } - public void close(AMQConstant cause, String message) + public void closeAsync(final AMQConstant cause, final String message) { - closeSubscriptions(); - performDeleteTasks(); - ConnectionCloseCode replyCode = ConnectionCloseCode.NORMAL; - try - { - replyCode = ConnectionCloseCode.get(cause.getCode()); - } - catch (IllegalArgumentException iae) + + addAsyncTask(new Action<ServerConnection>() { - // Ignore - } - close(replyCode, message); + @Override + public void performAction(final ServerConnection object) + { + closeSubscriptions(); + performDeleteTasks(); + + setState(CLOSING); + ConnectionCloseCode replyCode = ConnectionCloseCode.NORMAL; + try + { + replyCode = ConnectionCloseCode.get(cause.getCode()); + } + catch (IllegalArgumentException iae) + { + // Ignore + } + sendConnectionClose(replyCode, message); + } + }); } protected void performDeleteTasks() { - for(Action<? super ServerConnection> task : _taskList) + for(Action<? super ServerConnection> task : _connectionCloseTaskList) { task.performAction(this); } @@ -646,13 +675,19 @@ public class ServerConnection extends Connection implements AMQConnectionModel<S @Override public void addDeleteTask(final Action<? super ServerConnection> task) { - _taskList.add(task); + _connectionCloseTaskList.add(task); + } + + private void addAsyncTask(final Action<ServerConnection> action) + { + _asyncTaskList.add(action); + notifyWork(); } @Override public void removeDeleteTask(final Action<? super ServerConnection> task) { - _taskList.remove(task); + _connectionCloseTaskList.remove(task); } public int getMessageCompressionThreshold() @@ -664,4 +699,51 @@ public class ServerConnection extends Connection implements AMQConnectionModel<S { return _maxMessageSize; } + + public void transportStateChanged() + { + for (AMQSessionModel ssn : getSessionModels()) + { + ssn.transportStateChanged(); + } + } + + @Override + public void notifyWork() + { + _serverProtocolEngine.notifyWork(); + } + + + @Override + public boolean isMessageAssignmentSuspended() + { + return _serverProtocolEngine.isMessageAssignmentSuspended(); + } + + public void processPending() + { + while(_asyncTaskList.peek() != null) + { + Action<? super ServerConnection> asyncAction = _asyncTaskList.poll(); + asyncAction.performAction(this); + } + + for (AMQSessionModel session : getSessionModels()) + { + session.processPending(); + } + + } + + public void closeAndIgnoreFutureInput() + { + _ignoreFutureInput = true; + getSender().close(); + } + + public boolean isIgnoreFutureInput() + { + return _ignoreFutureInput; + } } diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java index 6e2a6cac7d..7f646b43b4 100644 --- a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java +++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java @@ -250,7 +250,7 @@ public class ServerConnectionDelegate extends ServerDelegate ") above the server's offered limit (" + getChannelMax() +")"); //Due to the error we must forcefully close the connection without negotiation - sconn.getSender().close(); + sconn.closeAndIgnoreFutureInput(); return; } @@ -261,7 +261,8 @@ public class ServerConnectionDelegate extends ServerDelegate ") above the server's offered limit (" + getFrameMax() +")"); //Due to the error we must forcefully close the connection without negotiation - sconn.getSender().close(); + sconn.closeAndIgnoreFutureInput(); + return; } else if(okMaxFrameSize > 0 && okMaxFrameSize < Constant.MIN_MAX_FRAME_SIZE) @@ -271,7 +272,7 @@ public class ServerConnectionDelegate extends ServerDelegate ") below the minimum permitted size (" + Constant.MIN_MAX_FRAME_SIZE +")"); //Due to the error we must forcefully close the connection without negotiation - sconn.getSender().close(); + sconn.closeAndIgnoreFutureInput(); return; } else if(okMaxFrameSize == 0) diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerDisassembler.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerDisassembler.java new file mode 100644 index 0000000000..a42238a40d --- /dev/null +++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerDisassembler.java @@ -0,0 +1,248 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF 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.apache.qpid.server.protocol.v0_10; + +import static java.lang.Math.min; +import static org.apache.qpid.transport.network.Frame.FIRST_FRAME; +import static org.apache.qpid.transport.network.Frame.FIRST_SEG; +import static org.apache.qpid.transport.network.Frame.HEADER_SIZE; +import static org.apache.qpid.transport.network.Frame.LAST_FRAME; +import static org.apache.qpid.transport.network.Frame.LAST_SEG; + +import java.nio.ByteBuffer; + +import org.apache.qpid.transport.ByteBufferSender; +import org.apache.qpid.transport.FrameSizeObserver; +import org.apache.qpid.transport.Header; +import org.apache.qpid.transport.Method; +import org.apache.qpid.transport.ProtocolDelegate; +import org.apache.qpid.transport.ProtocolError; +import org.apache.qpid.transport.ProtocolEvent; +import org.apache.qpid.transport.ProtocolEventSender; +import org.apache.qpid.transport.ProtocolHeader; +import org.apache.qpid.transport.SegmentType; +import org.apache.qpid.transport.Struct; +import org.apache.qpid.transport.codec.Encoder; +import org.apache.qpid.transport.network.Frame; + +/** + * Disassembler + */ +public final class ServerDisassembler implements ProtocolEventSender, ProtocolDelegate<Void>, FrameSizeObserver +{ + private final ByteBufferSender _sender; + private int _maxPayload; + private final Object _sendLock = new Object(); + private final Encoder _encoder = new ServerEncoder(); + + public ServerDisassembler(ByteBufferSender sender, int maxFrame) + { + _sender = sender; + if (maxFrame <= HEADER_SIZE || maxFrame >= 64 * 1024) + { + throw new IllegalArgumentException("maxFrame must be > HEADER_SIZE and < 64K: " + maxFrame); + } + _maxPayload = maxFrame - HEADER_SIZE; + } + + public void send(ProtocolEvent event) + { + synchronized (_sendLock) + { + event.delegate(null, this); + } + } + + public void flush() + { + synchronized (_sendLock) + { + _sender.flush(); + } + } + + public void close() + { + synchronized (_sendLock) + { + _sender.close(); + } + } + + private void frame(byte flags, byte type, byte track, int channel, int size, ByteBuffer buf) + { + ByteBuffer data = ByteBuffer.wrap(new byte[HEADER_SIZE]); + + data.put(0, flags); + data.put(1, type); + data.putShort(2, (short) (size + HEADER_SIZE)); + data.put(5, track); + data.putShort(6, (short) channel); + + + ByteBuffer dup = buf.duplicate(); + dup.limit(dup.position() + size); + buf.position(buf.position() + size); + _sender.send(data); + _sender.send(dup); + + + } + + private void fragment(byte flags, SegmentType type, ProtocolEvent event, ByteBuffer buf) + { + byte typeb = (byte) type.getValue(); + byte track = event.getEncodedTrack() == Frame.L4 ? (byte) 1 : (byte) 0; + + int remaining = buf.remaining(); + boolean first = true; + while (true) + { + int size = min(_maxPayload, remaining); + remaining -= size; + + byte newflags = flags; + if (first) + { + newflags |= FIRST_FRAME; + first = false; + } + if (remaining == 0) + { + newflags |= LAST_FRAME; + } + + frame(newflags, typeb, track, event.getChannel(), size, buf); + + if (remaining == 0) + { + break; + } + } + } + + public void init(Void v, ProtocolHeader header) + { + _sender.send(header.toByteBuffer()); + _sender.flush(); +} + + public void control(Void v, Method method) + { + method(method, SegmentType.CONTROL); + } + + public void command(Void v, Method method) + { + method(method, SegmentType.COMMAND); + } + + private void method(Method method, SegmentType type) + { + Encoder enc = _encoder; + enc.init(); + enc.writeUint16(method.getEncodedType()); + if (type == SegmentType.COMMAND) + { + if (method.isSync()) + { + enc.writeUint16(0x0101); + } + else + { + enc.writeUint16(0x0100); + } + } + method.write(enc); + int methodLimit = enc.position(); + + byte flags = FIRST_SEG; + + boolean payload = method.hasPayload(); + if (!payload) + { + flags |= LAST_SEG; + } + + int headerLimit = -1; + if (payload) + { + final Header hdr = method.getHeader(); + if (hdr != null) + { + if (hdr.getDeliveryProperties() != null) + { + enc.writeStruct32(hdr.getDeliveryProperties()); + } + if (hdr.getMessageProperties() != null) + { + enc.writeStruct32(hdr.getMessageProperties()); + } + if (hdr.getNonStandardProperties() != null) + { + for (Struct st : hdr.getNonStandardProperties()) + { + enc.writeStruct32(st); + } + } + } + + headerLimit = enc.position(); + } + synchronized (_sendLock) + { + ByteBuffer buf = enc.underlyingBuffer(); + buf.position(0); + buf.limit(methodLimit); + + fragment(flags, type, method, buf.duplicate()); + if (payload) + { + ByteBuffer body = method.getBody(); + buf.limit(headerLimit); + buf.position(methodLimit); + + fragment(body == null ? LAST_SEG : 0x0, SegmentType.HEADER, method, buf.duplicate()); + if (body != null) + { + fragment(LAST_SEG, SegmentType.BODY, method, body.duplicate()); + } + + } + } + } + + public void error(Void v, ProtocolError error) + { + throw new IllegalArgumentException(String.valueOf(error)); + } + + @Override + public void setMaxFrameSize(final int maxFrame) + { + if (maxFrame <= HEADER_SIZE || maxFrame >= 64*1024) + { + throw new IllegalArgumentException("maxFrame must be > HEADER_SIZE and < 64K: " + maxFrame); + } + this._maxPayload = maxFrame - HEADER_SIZE; + + } +} diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerEncoder.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerEncoder.java new file mode 100644 index 0000000000..6437015208 --- /dev/null +++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerEncoder.java @@ -0,0 +1,371 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF 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.apache.qpid.server.protocol.v0_10; + +import java.nio.BufferOverflowException; +import java.nio.ByteBuffer; +import java.util.UUID; + +import org.apache.qpid.transport.codec.AbstractEncoder; + + +public final class ServerEncoder extends AbstractEncoder +{ + public static final int DEFAULT_CAPACITY = 8192; + private final int _threshold; + private ByteBuffer _out; + private int _segment; + private int _initialCapacity; + + public ServerEncoder() + { + this(DEFAULT_CAPACITY); + } + + public ServerEncoder(int capacity) + { + _initialCapacity = capacity; + _threshold = capacity/16; + _out = ByteBuffer.allocate(capacity); + _segment = 0; + } + + public void init() + { + _out.position(_out.limit()); + _out.limit(_out.capacity()); + _out = _out.slice(); + if(_out.remaining() < _threshold) + { + _out = ByteBuffer.allocate(_initialCapacity); + } + _segment = 0; + } + + public ByteBuffer buffer() + { + int pos = _out.position(); + _out.position(_segment); + ByteBuffer slice = _out.slice(); + slice.limit(pos - _segment); + _out.position(pos); + return slice; + } + + public int position() + { + return _out.position(); + } + + public ByteBuffer underlyingBuffer() + { + return _out; + } + + private void grow(int size) + { + ByteBuffer old = _out; + int capacity = old.capacity(); + _out = ByteBuffer.allocate(Math.max(Math.max(capacity + size, 2*capacity), _initialCapacity)); + old.flip(); + _out.put(old); + } + + protected void doPut(byte b) + { + try + { + _out.put(b); + } + catch (BufferOverflowException e) + { + grow(1); + _out.put(b); + } + } + + protected void doPut(ByteBuffer src) + { + try + { + _out.put(src); + } + catch (BufferOverflowException e) + { + grow(src.remaining()); + _out.put(src); + } + } + + protected void put(byte[] bytes) + { + try + { + _out.put(bytes); + } + catch (BufferOverflowException e) + { + grow(bytes.length); + _out.put(bytes); + } + } + + public void writeUint8(short b) + { + assert b < 0x100; + + try + { + _out.put((byte) b); + } + catch (BufferOverflowException e) + { + grow(1); + _out.put((byte) b); + } + } + + public void writeUint16(int s) + { + assert s < 0x10000; + + try + { + _out.putShort((short) s); + } + catch (BufferOverflowException e) + { + grow(2); + _out.putShort((short) s); + } + } + + public void writeUint32(long i) + { + assert i < 0x100000000L; + + try + { + _out.putInt((int) i); + } + catch (BufferOverflowException e) + { + grow(4); + _out.putInt((int) i); + } + } + + public void writeUint64(long l) + { + try + { + _out.putLong(l); + } + catch (BufferOverflowException e) + { + grow(8); + _out.putLong(l); + } + } + + public int beginSize8() + { + int pos = _out.position(); + try + { + _out.put((byte) 0); + } + catch (BufferOverflowException e) + { + grow(1); + _out.put((byte) 0); + } + return pos; + } + + public void endSize8(int pos) + { + int cur = _out.position(); + _out.put(pos, (byte) (cur - pos - 1)); + } + + public int beginSize16() + { + int pos = _out.position(); + try + { + _out.putShort((short) 0); + } + catch (BufferOverflowException e) + { + grow(2); + _out.putShort((short) 0); + } + return pos; + } + + public void endSize16(int pos) + { + int cur = _out.position(); + _out.putShort(pos, (short) (cur - pos - 2)); + } + + public int beginSize32() + { + int pos = _out.position(); + try + { + _out.putInt(0); + } + catch (BufferOverflowException e) + { + grow(4); + _out.putInt(0); + } + return pos; + + } + + public void endSize32(int pos) + { + int cur = _out.position(); + _out.putInt(pos, (cur - pos - 4)); + + } + + public void writeDouble(double aDouble) + { + try + { + _out.putDouble(aDouble); + } + catch(BufferOverflowException exception) + { + grow(8); + _out.putDouble(aDouble); + } + } + + public void writeInt16(short aShort) + { + try + { + _out.putShort(aShort); + } + catch(BufferOverflowException exception) + { + grow(2); + _out.putShort(aShort); + } + } + + public void writeInt32(int anInt) + { + try + { + _out.putInt(anInt); + } + catch(BufferOverflowException exception) + { + grow(4); + _out.putInt(anInt); + } + } + + public void writeInt64(long aLong) + { + try + { + _out.putLong(aLong); + } + catch(BufferOverflowException exception) + { + grow(8); + _out.putLong(aLong); + } + } + + public void writeInt8(byte aByte) + { + try + { + _out.put(aByte); + } + catch(BufferOverflowException exception) + { + grow(1); + _out.put(aByte); + } + } + + public void writeBin128(byte[] byteArray) + { + byteArray = (byteArray != null) ? byteArray : new byte [16]; + + assert byteArray.length == 16; + + try + { + _out.put(byteArray); + } + catch(BufferOverflowException exception) + { + grow(16); + _out.put(byteArray); + } + } + + public void writeBin128(UUID id) + { + byte[] data = new byte[16]; + + long msb = id.getMostSignificantBits(); + long lsb = id.getLeastSignificantBits(); + + assert data.length == 16; + for (int i=7; i>=0; i--) + { + data[i] = (byte)(msb & 0xff); + msb = msb >> 8; + } + + for (int i=15; i>=8; i--) + { + data[i] = (byte)(lsb & 0xff); + lsb = (lsb >> 8); + } + writeBin128(data); + } + + public void writeFloat(float aFloat) + { + try + { + _out.putFloat(aFloat); + } + catch(BufferOverflowException exception) + { + grow(4); + _out.putFloat(aFloat); + } + } + +} diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java index 223de4f84e..67204427fb 100644 --- a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java +++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java @@ -56,6 +56,7 @@ import org.apache.qpid.server.TransactionTimeoutHelper; import org.apache.qpid.server.TransactionTimeoutHelper.CloseAction; import org.apache.qpid.server.connection.SessionPrincipal; import org.apache.qpid.server.consumer.ConsumerImpl; +import org.apache.qpid.server.consumer.ConsumerTarget; import org.apache.qpid.server.logging.LogMessage; import org.apache.qpid.server.logging.LogSubject; import org.apache.qpid.server.logging.messages.ChannelMessages; @@ -74,7 +75,7 @@ import org.apache.qpid.server.queue.AMQQueue; import org.apache.qpid.server.security.AuthorizationHolder; import org.apache.qpid.server.store.MessageStore; import org.apache.qpid.server.store.StoreException; -import org.apache.qpid.server.store.StoreFuture; +import org.apache.qpid.server.util.FutureResult; import org.apache.qpid.server.store.StoredMessage; import org.apache.qpid.server.store.TransactionLogResource; import org.apache.qpid.server.txn.AlreadyKnownDtxException; @@ -136,6 +137,7 @@ public class ServerSession extends Session private org.apache.qpid.server.model.Session<?> _modelObject; private long _blockTime; private long _blockingTimeout; + private boolean _wireBlockingState; public static interface MessageDispositionChangeListener { @@ -188,7 +190,7 @@ public class ServerSession extends Session @Override public void doTimeoutAction(String reason) { - getConnectionModel().closeSession(ServerSession.this, AMQConstant.RESOURCE_ERROR, reason); + getConnectionModel().closeSessionAsync(ServerSession.this, AMQConstant.RESOURCE_ERROR, reason); } }, getVirtualHost()); @@ -207,10 +209,6 @@ public class ServerSession extends Session if (state == State.OPEN) { getVirtualHost().getEventLogger().message(ChannelMessages.CREATE()); - if(_blocking.get()) - { - invokeBlock(); - } } } else @@ -244,6 +242,17 @@ public class ServerSession extends Session invoke(new MessageStop("")); } + private void invokeUnblock() + { + MessageFlow mf = new MessageFlow(); + mf.setUnit(MessageCreditUnit.MESSAGE); + mf.setDestination(""); + _outstandingCredit.set(Integer.MAX_VALUE); + mf.setValue(Integer.MAX_VALUE); + invoke(mf); + } + + @Override protected boolean isFull(int id) { @@ -823,12 +832,11 @@ public class ServerSession extends Session if(_blocking.compareAndSet(false,true)) { + getVirtualHost().getEventLogger().message(_logSubject, ChannelMessages.FLOW_ENFORCED(name)); if(getState() == State.OPEN) { - invokeBlock(); + getConnection().notifyWork(); } - _blockTime = System.currentTimeMillis(); - getVirtualHost().getEventLogger().message(_logSubject, ChannelMessages.FLOW_ENFORCED(name)); } @@ -852,28 +860,30 @@ public class ServerSession extends Session { if(_blocking.compareAndSet(true,false) && !isClosing()) { - _blockTime = 0l; getVirtualHost().getEventLogger().message(_logSubject, ChannelMessages.FLOW_REMOVED()); - MessageFlow mf = new MessageFlow(); - mf.setUnit(MessageCreditUnit.MESSAGE); - mf.setDestination(""); - _outstandingCredit.set(Integer.MAX_VALUE); - mf.setValue(Integer.MAX_VALUE); - invoke(mf); - - + getConnection().notifyWork(); } } } + boolean blockingTimeoutExceeded() { long blockTime = _blockTime; - boolean b = _blocking.get() && blockTime != 0 && (System.currentTimeMillis() - blockTime) > _blockingTimeout; + boolean b = _wireBlockingState && blockTime != 0 && (System.currentTimeMillis() - blockTime) > _blockingTimeout; return b; } @Override + public void transportStateChanged() + { + for(ConsumerTarget_0_10 consumerTarget : getSubscriptions()) + { + consumerTarget.transportStateChanged(); + } + } + + @Override public Object getConnectionReference() { return getConnection().getReference(); @@ -1002,17 +1012,17 @@ public class ServerSession extends Session return _unfinishedCommandsQueue.isEmpty() ? null : _unfinishedCommandsQueue.getLast(); } - public void recordFuture(final StoreFuture future, final ServerTransaction.Action action) + public void recordFuture(final FutureResult future, final ServerTransaction.Action action) { _unfinishedCommandsQueue.add(new AsyncCommand(future, action)); } private static class AsyncCommand { - private final StoreFuture _future; + private final FutureResult _future; private ServerTransaction.Action _action; - public AsyncCommand(final StoreFuture future, final ServerTransaction.Action action) + public AsyncCommand(final FutureResult future, final ServerTransaction.Action action) { _future = future; _action = action; @@ -1125,6 +1135,32 @@ public class ServerSession extends Session } } + @Override + public void processPending() + { + boolean desiredBlockingState = _blocking.get(); + if (desiredBlockingState != _wireBlockingState) + { + _wireBlockingState = desiredBlockingState; + + if (desiredBlockingState) + { + invokeBlock(); + } + else + { + invokeUnblock(); + } + _blockTime = desiredBlockingState ? System.currentTimeMillis() : 0; + } + + + for(ConsumerTarget target : getSubscriptions()) + { + target.processPending(); + } + } + public final long getMaxUncommittedInMemorySize() { diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java index 8632d04048..dd634c36ff 100644 --- a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java +++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java @@ -36,8 +36,10 @@ import org.apache.log4j.Logger; import org.apache.qpid.common.AMQPFilterTypes; import org.apache.qpid.exchange.ExchangeDefaults; import org.apache.qpid.protocol.AMQConstant; +import org.apache.qpid.server.protocol.ServerProtocolEngine; import org.apache.qpid.server.consumer.ConsumerImpl; import org.apache.qpid.server.exchange.ExchangeImpl; +import org.apache.qpid.server.virtualhost.VirtualHostUnavailableException; import org.apache.qpid.server.filter.AMQInvalidArgumentException; import org.apache.qpid.server.filter.ArrivalTimeFilter; import org.apache.qpid.server.filter.FilterManager; @@ -58,7 +60,7 @@ import org.apache.qpid.server.queue.AMQQueue; import org.apache.qpid.server.queue.QueueArgumentsConverter; import org.apache.qpid.server.store.MessageStore; import org.apache.qpid.server.store.StoreException; -import org.apache.qpid.server.store.StoreFuture; +import org.apache.qpid.server.util.FutureResult; import org.apache.qpid.server.store.StoredMessage; import org.apache.qpid.server.txn.AlreadyKnownDtxException; import org.apache.qpid.server.txn.DtxNotSelectedException; @@ -133,7 +135,7 @@ public class ServerSessionDelegate extends SessionDelegate serverSession.accept(method.getTransfers()); if(!serverSession.isTransactional()) { - serverSession.recordFuture(StoreFuture.IMMEDIATE_FUTURE, + serverSession.recordFuture(FutureResult.IMMEDIATE_FUTURE, new CommandProcessedAction(serverSession, method)); } } @@ -246,8 +248,8 @@ public class ServerSessionDelegate extends SessionDelegate } else { - - FlowCreditManager_0_10 creditManager = new WindowCreditManager(0L,0L); + ServerProtocolEngine serverProtocolEngine = getServerConnection(session).getProtocolEngine(); + FlowCreditManager_0_10 creditManager = new WindowCreditManager(0L,0L, serverProtocolEngine); FilterManager filterManager = null; try @@ -421,58 +423,69 @@ public class ServerSessionDelegate extends SessionDelegate new MessageTransferMessage(storeMessage, serverSession.getReference()); MessageReference<MessageTransferMessage> reference = message.newReference(); - final InstanceProperties instanceProperties = new InstanceProperties() + try { - @Override - public Object getProperty(final Property prop) + final InstanceProperties instanceProperties = new InstanceProperties() { - switch (prop) + @Override + public Object getProperty(final Property prop) { - case EXPIRATION: - return message.getExpiration(); - case IMMEDIATE: - return message.isImmediate(); - case MANDATORY: - return (delvProps == null || !delvProps.getDiscardUnroutable()) - && xfr.getAcceptMode() == MessageAcceptMode.EXPLICIT; - case PERSISTENT: - return message.isPersistent(); - case REDELIVERED: - return delvProps.getRedelivered(); + switch (prop) + { + case EXPIRATION: + return message.getExpiration(); + case IMMEDIATE: + return message.isImmediate(); + case MANDATORY: + return (delvProps == null || !delvProps.getDiscardUnroutable()) + && xfr.getAcceptMode() == MessageAcceptMode.EXPLICIT; + case PERSISTENT: + return message.isPersistent(); + case REDELIVERED: + return delvProps.getRedelivered(); + } + return null; } - return null; - } - }; + }; - int enqueues = serverSession.enqueue(message, instanceProperties, destination); + int enqueues = serverSession.enqueue(message, instanceProperties, destination); - if (enqueues == 0) - { - if ((delvProps == null || !delvProps.getDiscardUnroutable()) - && xfr.getAcceptMode() == MessageAcceptMode.EXPLICIT) + if (enqueues == 0) { - RangeSet rejects = RangeSetFactory.createRangeSet(); - rejects.add(xfr.getId()); - MessageReject reject = new MessageReject(rejects, MessageRejectCode.UNROUTABLE, "Unroutable"); - ssn.invoke(reject); + if ((delvProps == null || !delvProps.getDiscardUnroutable()) + && xfr.getAcceptMode() == MessageAcceptMode.EXPLICIT) + { + RangeSet rejects = RangeSetFactory.createRangeSet(); + rejects.add(xfr.getId()); + MessageReject reject = new MessageReject(rejects, MessageRejectCode.UNROUTABLE, "Unroutable"); + ssn.invoke(reject); + } + else + { + virtualHost.getEventLogger().message(ExchangeMessages.DISCARDMSG(destination.getName(), + messageMetaData.getRoutingKey())); + } + } + + if (serverSession.isTransactional()) + { + serverSession.processed(xfr); } else { - virtualHost.getEventLogger().message(ExchangeMessages.DISCARDMSG(destination.getName(), - messageMetaData.getRoutingKey())); + serverSession.recordFuture(FutureResult.IMMEDIATE_FUTURE, + new CommandProcessedAction(serverSession, xfr)); } } - - if (serverSession.isTransactional()) + catch (VirtualHostUnavailableException e) { - serverSession.processed(xfr); + getServerConnection(serverSession).closeAsync(AMQConstant.CONNECTION_FORCED, e.getMessage()); } - else + finally { - serverSession.recordFuture(StoreFuture.IMMEDIATE_FUTURE, - new CommandProcessedAction(serverSession, xfr)); + reference.release(); } - reference.release(); + } } @@ -589,7 +602,7 @@ public class ServerSessionDelegate extends SessionDelegate { try { - ((ServerSession)session).endDtx(method.getXid(), method.getFail(), method.getSuspend()); + ((ServerSession) session).endDtx(method.getXid(), method.getFail(), method.getSuspend()); } catch (TimeoutDtxException e) { diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/WindowCreditManager.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/WindowCreditManager.java index 8e48741b91..a7b08e3f83 100644 --- a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/WindowCreditManager.java +++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/WindowCreditManager.java @@ -21,11 +21,14 @@ package org.apache.qpid.server.protocol.v0_10; import org.apache.log4j.Logger; + +import org.apache.qpid.server.protocol.ServerProtocolEngine; import org.apache.qpid.server.flow.AbstractFlowCreditManager; public class WindowCreditManager extends AbstractFlowCreditManager implements FlowCreditManager_0_10 { private static final Logger LOGGER = Logger.getLogger(WindowCreditManager.class); + private final ServerProtocolEngine _serverProtocolEngine; private volatile long _bytesCreditLimit; private volatile long _messageCreditLimit; @@ -33,39 +36,22 @@ public class WindowCreditManager extends AbstractFlowCreditManager implements Fl private volatile long _bytesUsed; private volatile long _messageUsed; - public WindowCreditManager() - { - this(0L, 0L); - } - - public WindowCreditManager(long bytesCreditLimit, long messageCreditLimit) + public WindowCreditManager(long bytesCreditLimit, + long messageCreditLimit, + ServerProtocolEngine serverProtocolEngine) { + _serverProtocolEngine = serverProtocolEngine; _bytesCreditLimit = bytesCreditLimit; _messageCreditLimit = messageCreditLimit; setSuspended(!hasCredit()); } - public long getBytesCreditLimit() - { - return _bytesCreditLimit; - } - public long getMessageCreditLimit() { return _messageCreditLimit; } - public synchronized void setCreditLimits(final long bytesCreditLimit, final long messageCreditLimit) - { - _bytesCreditLimit = bytesCreditLimit; - _messageCreditLimit = messageCreditLimit; - - setSuspended(!hasCredit()); - - } - - public long getMessageCredit() { return _messageCreditLimit == -1L @@ -121,12 +107,18 @@ public class WindowCreditManager extends AbstractFlowCreditManager implements Fl public synchronized boolean hasCredit() { return (_bytesCreditLimit < 0L || _bytesCreditLimit > _bytesUsed) - && (_messageCreditLimit < 0L || _messageCreditLimit > _messageUsed); + && (_messageCreditLimit < 0L || _messageCreditLimit > _messageUsed) + && !_serverProtocolEngine.isTransportBlockedForWriting(); } public synchronized boolean useCreditForMessage(final long msgSize) { - if(_messageCreditLimit >= 0L) + if (_serverProtocolEngine.isTransportBlockedForWriting()) + { + setSuspended(true); + return false; + } + else if(_messageCreditLimit >= 0L) { if(_messageUsed < _messageCreditLimit) { |