You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by lq...@apache.org on 2017/03/31 13:38:27 UTC
svn commit: r1789686 -
/qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/
Author: lquack
Date: Fri Mar 31 13:38:27 2017
New Revision: 1789686
URL: http://svn.apache.org/viewvc?rev=1789686&view=rev
Log:
QPID-7658: [Java Broker] LinkRegistry: Address review comments
Added:
qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AbstractReceivingLinkEndpoint.java
- copied, changed from r1789671, qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ReceivingLinkEndpoint.java
Removed:
qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ReceivingLinkEndpoint.java
Modified:
qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkImpl.java
qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLinkEndpoint.java
qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/StandardReceivingLinkEndpoint.java
qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/TxnCoordinatorReceivingLinkEndpoint.java
Copied: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AbstractReceivingLinkEndpoint.java (from r1789671, qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ReceivingLinkEndpoint.java)
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AbstractReceivingLinkEndpoint.java?p2=qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AbstractReceivingLinkEndpoint.java&p1=qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ReceivingLinkEndpoint.java&r1=1789671&r2=1789686&rev=1789686&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ReceivingLinkEndpoint.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AbstractReceivingLinkEndpoint.java Fri Mar 31 13:38:27 2017
@@ -39,11 +39,15 @@ import org.apache.qpid.server.protocol.v
import org.apache.qpid.server.protocol.v1_0.type.transport.Role;
import org.apache.qpid.server.protocol.v1_0.type.transport.Transfer;
-public abstract class ReceivingLinkEndpoint extends AbstractLinkEndpoint
+public abstract class AbstractReceivingLinkEndpoint extends AbstractLinkEndpoint
{
private final SectionDecoder _sectionDecoder;
private UnsignedInteger _lastDeliveryId;
private ReceivingDestination _receivingDestination;
+ private Map<Binary, Object> _unsettledMap = new LinkedHashMap<>();
+ private Map<Binary, TransientState> _unsettledIds = new LinkedHashMap<>();
+ private boolean _creditWindow;
+
private static class TransientState
{
@@ -83,14 +87,8 @@ public abstract class ReceivingLinkEndpo
}
}
- private Map<Binary, Object> _unsettledMap = new LinkedHashMap<Binary, Object>();
- private Map<Binary, TransientState> _unsettledIds = new LinkedHashMap<Binary, TransientState>();
- private boolean _creditWindow;
- private boolean _remoteDrain;
- private UnsignedInteger _drainLimit;
-
- public ReceivingLinkEndpoint(final Session_1_0 session, final Link_1_0 link)
+ public AbstractReceivingLinkEndpoint(final Session_1_0 session, final Link_1_0 link)
{
super(session, link);
_sectionDecoder = new SectionDecoderImpl(session.getConnection()
@@ -169,19 +167,8 @@ public abstract class ReceivingLinkEndpo
@Override public void receiveFlow(final Flow flow)
{
- _remoteDrain = Boolean.TRUE.equals((Boolean) flow.getDrain());
setAvailable(flow.getAvailable());
setDeliveryCount(flow.getDeliveryCount());
- if(isDrained())
- {
-
- }
- }
-
-
- public boolean isDrained()
- {
- return getDrain() && getDeliveryCount().equals(getDrainLimit());
}
public boolean settled(final Binary deliveryTag)
@@ -305,11 +292,6 @@ public abstract class ReceivingLinkEndpo
{
}
- public UnsignedInteger getDrainLimit()
- {
- return _drainLimit;
- }
-
UnsignedInteger getLastDeliveryId()
{
return _lastDeliveryId;
Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkImpl.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkImpl.java?rev=1789686&r1=1789685&r2=1789686&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkImpl.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkImpl.java Fri Mar 31 13:38:27 2017
@@ -87,12 +87,6 @@ public class LinkImpl implements Link_1_
if (_linkEndpoint == null)
{
_linkEndpoint = createLinkEndpoint(session, attach);
- if (_linkEndpoint == null)
- {
- throw new ConnectionScopedRuntimeException(String.format(
- "LinkEndpoint creation failed for attach: %s",
- attach));
- }
}
_linkEndpoint.receiveAttach(attach);
@@ -143,17 +137,13 @@ public class LinkImpl implements Link_1_
{
linkEndpoint = new SendingLinkEndpoint(session, this);
}
- else if (_role == Role.RECEIVER && attach.getTarget() != null)
+ else if (attach.getTarget() instanceof Coordinator)
{
-
- if (attach.getTarget() instanceof Target)
- {
- linkEndpoint = new StandardReceivingLinkEndpoint(session, this);
- }
- else if (attach.getTarget() instanceof Coordinator)
- {
- linkEndpoint = new TxnCoordinatorReceivingLinkEndpoint(session, this);
- }
+ linkEndpoint = new TxnCoordinatorReceivingLinkEndpoint(session, this);
+ }
+ else
+ {
+ linkEndpoint = new StandardReceivingLinkEndpoint(session, this);
}
return linkEndpoint;
}
Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLinkEndpoint.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLinkEndpoint.java?rev=1789686&r1=1789685&r2=1789686&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLinkEndpoint.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLinkEndpoint.java Fri Mar 31 13:38:27 2017
@@ -90,7 +90,6 @@ public class SendingLinkEndpoint extends
private final List<MessageInstance> _resumeFullTransfers = new ArrayList<>();
private volatile boolean _draining = false;
private final ConcurrentMap<Binary, UnsettledAction> _unsettledActionMap = new ConcurrentHashMap<>();
- private TerminusDurability _durability;
private SendingDestination _destination;
private EnumSet<ConsumerOption> _consumerOptions;
private FilterManager _consumerFilters;
@@ -515,35 +514,36 @@ public class SendingLinkEndpoint extends
Modified state = new Modified();
state.setDeliveryFailed(true);
- for(UnsettledAction action : _unsettledActionMap.values())
+ for (UnsettledAction action : _unsettledActionMap.values())
{
- action.process(state,Boolean.TRUE);
+ action.process(state, Boolean.TRUE);
}
_unsettledActionMap.clear();
+ Error closingError = null;
if (getDestination() instanceof ExchangeDestination
&& getSession().getConnection().getAddressSpace() instanceof QueueManagingVirtualHost)
{
try
{
- ((QueueManagingVirtualHost) getSession().getConnection().getAddressSpace()).removeSubscriptionQueue(((ExchangeDestination) getDestination()).getQueue().getName());
+ ((QueueManagingVirtualHost) getSession().getConnection().getAddressSpace()).removeSubscriptionQueue(
+ ((ExchangeDestination) getDestination()).getQueue().getName());
}
catch (AccessControlException e)
{
LOGGER.error("Error unregistering subscription", e);
- close(new Error(AmqpError.NOT_ALLOWED, "Error unregistering subscription"));
+ closingError = new Error(AmqpError.NOT_ALLOWED, "Error unregistering subscription");
}
catch (IllegalStateException e)
{
- close(new Error(AmqpError.RESOURCE_LOCKED, e.getMessage()));
+ closingError = new Error(AmqpError.RESOURCE_LOCKED, e.getMessage());
}
catch (NotFoundException e)
{
- close(new Error(AmqpError.NOT_FOUND, e.getMessage()));
+ closingError = new Error(AmqpError.NOT_FOUND, e.getMessage());
}
}
-
- close();
+ close(closingError);
}
else if (detach.getError() != null)
{
Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java?rev=1789686&r1=1789685&r2=1789686&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java Fri Mar 31 13:38:27 2017
@@ -205,8 +205,9 @@ public class Session_1_0 extends Abstrac
UnsignedInteger inputHandle = attach.getHandle();
if (_inputHandleToEndpoint.containsKey(inputHandle))
{
- getConnection().close(new Error(SessionError.HANDLE_IN_USE, "inputHandle of Attach already in use: " + attach.toString()));
- throw new ConnectionScopedRuntimeException(String.format("Input Handle '%d' already in use", inputHandle.intValue()));
+ String errorMessage = String.format("Input Handle '%d' already in use", inputHandle.intValue());
+ getConnection().close(new Error(SessionError.HANDLE_IN_USE, errorMessage));
+ throw new ConnectionScopedRuntimeException(errorMessage);
}
else
{
@@ -575,7 +576,7 @@ public class Session_1_0 extends Abstrac
_connection.close(error);
}
- else if(!(linkEndpoint instanceof ReceivingLinkEndpoint))
+ else if(!(linkEndpoint instanceof AbstractReceivingLinkEndpoint))
{
Error error = new Error();
@@ -586,7 +587,7 @@ public class Session_1_0 extends Abstrac
}
else
{
- ReceivingLinkEndpoint endpoint = ((ReceivingLinkEndpoint) linkEndpoint);
+ AbstractReceivingLinkEndpoint endpoint = ((AbstractReceivingLinkEndpoint) linkEndpoint);
UnsignedInteger deliveryId = transfer.getDeliveryId();
if (deliveryId == null)
@@ -1237,8 +1238,8 @@ public class Session_1_0 extends Abstrac
for (LinkEndpoint linkEndpoint : _endpointToOutputHandle.keySet())
{
- if (linkEndpoint instanceof ReceivingLinkEndpoint
- && isQueueDestinationForLink(queue, ((ReceivingLinkEndpoint) linkEndpoint).getReceivingDestination()))
+ if (linkEndpoint instanceof AbstractReceivingLinkEndpoint
+ && isQueueDestinationForLink(queue, ((AbstractReceivingLinkEndpoint) linkEndpoint).getReceivingDestination()))
{
linkEndpoint.setStopped(true);
}
@@ -1277,8 +1278,8 @@ public class Session_1_0 extends Abstrac
}
for (LinkEndpoint linkEndpoint : _endpointToOutputHandle.keySet())
{
- if (linkEndpoint instanceof ReceivingLinkEndpoint
- && isQueueDestinationForLink(queue, ((ReceivingLinkEndpoint) linkEndpoint).getReceivingDestination()))
+ if (linkEndpoint instanceof AbstractReceivingLinkEndpoint
+ && isQueueDestinationForLink(queue, ((AbstractReceivingLinkEndpoint) linkEndpoint).getReceivingDestination()))
{
linkEndpoint.setStopped(false);
}
@@ -1308,7 +1309,7 @@ public class Session_1_0 extends Abstrac
for (LinkEndpoint linkEndpoint : _endpointToOutputHandle.keySet())
{
- if (linkEndpoint instanceof ReceivingLinkEndpoint)
+ if (linkEndpoint instanceof AbstractReceivingLinkEndpoint)
{
linkEndpoint.setStopped(true);
}
@@ -1340,8 +1341,8 @@ public class Session_1_0 extends Abstrac
}
for (LinkEndpoint linkEndpoint : _endpointToOutputHandle.keySet())
{
- if (linkEndpoint instanceof ReceivingLinkEndpoint
- && !_blockingEntities.contains(((ReceivingLinkEndpoint) linkEndpoint).getReceivingDestination()))
+ if (linkEndpoint instanceof AbstractReceivingLinkEndpoint
+ && !_blockingEntities.contains(((AbstractReceivingLinkEndpoint) linkEndpoint).getReceivingDestination()))
{
linkEndpoint.setStopped(false);
}
@@ -1626,20 +1627,12 @@ public class Session_1_0 extends Abstrac
endpoint.sendAttach();
((ErrantLinkEndpoint) endpoint).closeWithError();
}
- else if (attachWasUnsuccessful(endpoint))
- {
- endpoint.sendAttach();
-
- Error error = new Error();
- error.setCondition(AmqpError.NOT_FOUND);
- endpoint.close(error);
- }
else
{
if (endpoint.getRole() == Role.RECEIVER
&& (_blockingEntities.contains(Session_1_0.this)
|| (endpoint instanceof StandardReceivingLinkEndpoint
- && _blockingEntities.contains(((ReceivingLinkEndpoint) endpoint).getReceivingDestination()))))
+ && _blockingEntities.contains(((AbstractReceivingLinkEndpoint) endpoint).getReceivingDestination()))))
{
endpoint.setStopped(true);
}
@@ -1667,22 +1660,6 @@ public class Session_1_0 extends Abstrac
_logger.error(errorMessage, t);
throw new ConnectionScopedRuntimeException(errorMessage, t);
}
-
- private boolean attachWasUnsuccessful(final LinkEndpoint endpoint)
- {
- if (endpoint.getRole().equals(Role.SENDER))
- {
- return endpoint.getSource() == null;
- }
- else if (endpoint.getRole().equals(Role.RECEIVER))
- {
- return endpoint.getTarget() == null;
- }
- else
- {
- throw new IllegalStateException(String.format("Unknown LinkEndpoint role '%s'", endpoint.getRole()));
- }
- }
}
private final class BindingInfo
Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/StandardReceivingLinkEndpoint.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/StandardReceivingLinkEndpoint.java?rev=1789686&r1=1789685&r2=1789686&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/StandardReceivingLinkEndpoint.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/StandardReceivingLinkEndpoint.java Fri Mar 31 13:38:27 2017
@@ -69,7 +69,7 @@ import org.apache.qpid.server.txn.LocalT
import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
-public class StandardReceivingLinkEndpoint extends ReceivingLinkEndpoint
+public class StandardReceivingLinkEndpoint extends AbstractReceivingLinkEndpoint
{
private static final Logger LOGGER = LoggerFactory.getLogger(StandardReceivingLinkEndpoint.class);
Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/TxnCoordinatorReceivingLinkEndpoint.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/TxnCoordinatorReceivingLinkEndpoint.java?rev=1789686&r1=1789685&r2=1789686&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/TxnCoordinatorReceivingLinkEndpoint.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/TxnCoordinatorReceivingLinkEndpoint.java Fri Mar 31 13:38:27 2017
@@ -47,7 +47,7 @@ import org.apache.qpid.server.txn.LocalT
import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
-public class TxnCoordinatorReceivingLinkEndpoint extends ReceivingLinkEndpoint
+public class TxnCoordinatorReceivingLinkEndpoint extends AbstractReceivingLinkEndpoint
{
private final LinkedHashMap<Integer, ServerTransaction> _createdTransactions = new LinkedHashMap<>();
private ArrayList<Transfer> _incompleteMessage;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org