qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From lqu...@apache.org
Subject svn commit: r1789687 [1/2] - in /qpid/java/trunk/broker-plugins: amqp-1-0-bdb-store/src/main/java/org/apache/qpid/server/protocol/v1_0/store/bdb/ amqp-1-0-jdbc-store/src/main/java/org/apache/qpid/server/protocol/v1_0/store/jdbc/ amqp-1-0-protocol/src/m...
Date Fri, 31 Mar 2017 13:38:45 GMT
Author: lquack
Date: Fri Mar 31 13:38:45 2017
New Revision: 1789687

URL: http://svn.apache.org/viewvc?rev=1789687&view=rev
Log:
QPID-7658: [Java Broker] LinkRegistry: Address review comment: Make Link and LinkEndpoint generic

Modified:
    qpid/java/trunk/broker-plugins/amqp-1-0-bdb-store/src/main/java/org/apache/qpid/server/protocol/v1_0/store/bdb/BDBLinkStore.java
    qpid/java/trunk/broker-plugins/amqp-1-0-bdb-store/src/main/java/org/apache/qpid/server/protocol/v1_0/store/bdb/LinkValue.java
    qpid/java/trunk/broker-plugins/amqp-1-0-jdbc-store/src/main/java/org/apache/qpid/server/protocol/v1_0/store/jdbc/JDBCLinkStore.java
    qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AbstractLinkEndpoint.java
    qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AbstractReceivingLinkEndpoint.java
    qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java
    qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Delivery.java
    qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ErrantLinkEndpoint.java
    qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkDefinition.java
    qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkDefinitionImpl.java
    qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkEndpoint.java
    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/LinkKey.java
    qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkRegistry.java
    qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkRegistryImpl.java
    qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Link_1_0.java
    qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/NullLinkStoreFactory.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
    qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/store/AbstractLinkStore.java
    qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/store/LinkStore.java
    qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/store/LinkStoreUpdater.java
    qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/store/LinkStoreUpdaterImpl.java
    qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/TestLinkStoreFactory.java
    qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/store/LinkStoreTestCase.java

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-bdb-store/src/main/java/org/apache/qpid/server/protocol/v1_0/store/bdb/BDBLinkStore.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-1-0-bdb-store/src/main/java/org/apache/qpid/server/protocol/v1_0/store/bdb/BDBLinkStore.java?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-bdb-store/src/main/java/org/apache/qpid/server/protocol/v1_0/store/bdb/BDBLinkStore.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-bdb-store/src/main/java/org/apache/qpid/server/protocol/v1_0/store/bdb/BDBLinkStore.java Fri Mar 31 13:38:45 2017
@@ -45,6 +45,8 @@ import org.apache.qpid.server.protocol.v
 import org.apache.qpid.server.protocol.v1_0.LinkKey;
 import org.apache.qpid.server.protocol.v1_0.store.AbstractLinkStore;
 import org.apache.qpid.server.protocol.v1_0.store.LinkStoreUpdater;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Source;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Target;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.TerminusDurability;
 import org.apache.qpid.server.store.StoreException;
 import org.apache.qpid.server.store.berkeleydb.BDBEnvironmentContainer;
@@ -64,7 +66,7 @@ public class BDBLinkStore extends Abstra
     }
 
     @Override
-    protected Collection<LinkDefinition> doOpenAndLoad(final LinkStoreUpdater updater) throws StoreException
+    protected Collection<LinkDefinition<Source, Target>> doOpenAndLoad(final LinkStoreUpdater updater) throws StoreException
     {
         try
         {
@@ -77,7 +79,7 @@ public class BDBLinkStore extends Abstra
     }
 
     @Override
-    protected void doSaveLink(final LinkDefinition link)
+    protected void doSaveLink(final LinkDefinition<Source, Target> link)
     {
         try
         {
@@ -91,7 +93,7 @@ public class BDBLinkStore extends Abstra
     }
 
     @Override
-    protected void doDeleteLink(final LinkDefinition linkDefinition)
+    protected void doDeleteLink(final LinkDefinition<Source, Target> linkDefinition)
     {
         LinkKey linkKey = new LinkKey(linkDefinition);
         try
@@ -145,10 +147,10 @@ public class BDBLinkStore extends Abstra
     }
 
 
-    private Collection<LinkDefinition> getLinkDefinitions(final LinkStoreUpdater updater)
+    private Collection<LinkDefinition<Source, Target>> getLinkDefinitions(final LinkStoreUpdater updater)
     {
         Database linksDatabase = getEnvironmentFacade().openDatabase(LINKS_DB_NAME, DEFAULT_DATABASE_CONFIG);
-        Collection<LinkDefinition> links = new HashSet<>();
+        Collection<LinkDefinition<Source, Target>> links = new HashSet<>();
 
         ModelVersion currentVersion =
                 new ModelVersion(BrokerModel.MODEL_MAJOR_VERSION, BrokerModel.MODEL_MINOR_VERSION);
@@ -168,7 +170,7 @@ public class BDBLinkStore extends Abstra
             {
                 LinkKey linkKey = keyEntryBinding.entryToObject(key);
                 LinkValue linkValue = linkValueEntryBinding.entryToObject(value);
-                LinkDefinition link = new LinkDefinitionImpl(linkKey.getRemoteContainerId(), linkKey.getLinkName(), linkKey.getRole(), linkValue.getSource(), linkValue.getTarget());
+                LinkDefinition<Source, Target> link = new LinkDefinitionImpl<>(linkKey.getRemoteContainerId(), linkKey.getLinkName(), linkKey.getRole(), linkValue.getSource(), linkValue.getTarget());
                 links.add(link);
             }
         }
@@ -180,7 +182,7 @@ public class BDBLinkStore extends Abstra
             try
             {
                 linksDatabase = getEnvironmentFacade().clearDatabase(txn, LINKS_DB_NAME, DEFAULT_DATABASE_CONFIG);
-                for (LinkDefinition link : links)
+                for (LinkDefinition<Source, Target> link : links)
                 {
                     save(linksDatabase, txn, link);
                 }
@@ -208,7 +210,7 @@ public class BDBLinkStore extends Abstra
         linksVersionDb.put(txn, key, value);
     }
 
-    private void save(Database database, Transaction txn, final LinkDefinition link)
+    private void save(Database database, Transaction txn, final LinkDefinition<Source, Target> link)
     {
         DatabaseEntry key = new DatabaseEntry();
         DatabaseEntry value = new DatabaseEntry();

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-bdb-store/src/main/java/org/apache/qpid/server/protocol/v1_0/store/bdb/LinkValue.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-1-0-bdb-store/src/main/java/org/apache/qpid/server/protocol/v1_0/store/bdb/LinkValue.java?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-bdb-store/src/main/java/org/apache/qpid/server/protocol/v1_0/store/bdb/LinkValue.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-bdb-store/src/main/java/org/apache/qpid/server/protocol/v1_0/store/bdb/LinkValue.java Fri Mar 31 13:38:45 2017
@@ -20,34 +20,34 @@
 package org.apache.qpid.server.protocol.v1_0.store.bdb;
 
 import org.apache.qpid.server.protocol.v1_0.LinkDefinition;
-import org.apache.qpid.server.protocol.v1_0.type.BaseSource;
-import org.apache.qpid.server.protocol.v1_0.type.BaseTarget;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Source;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Target;
 
 public class LinkValue
 {
     static final byte CURRENT_VERSION = 0;
-    private final BaseSource _source;
-    private final BaseTarget _target;
+    private final Source _source;
+    private final Target _target;
     private final byte _version;
 
-    public LinkValue(final BaseSource source, final BaseTarget target, final byte version)
+    public LinkValue(final Source source, final Target target, final byte version)
     {
         _source = source;
         _target = target;
         _version = version;
     }
 
-    public LinkValue(final LinkDefinition link)
+    public LinkValue(final LinkDefinition<Source, Target> link)
     {
         this(link.getSource(), link.getTarget(), CURRENT_VERSION);
     }
 
-    public BaseSource getSource()
+    public Source getSource()
     {
         return _source;
     }
 
-    public BaseTarget getTarget()
+    public Target getTarget()
     {
         return _target;
     }

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-jdbc-store/src/main/java/org/apache/qpid/server/protocol/v1_0/store/jdbc/JDBCLinkStore.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-1-0-jdbc-store/src/main/java/org/apache/qpid/server/protocol/v1_0/store/jdbc/JDBCLinkStore.java?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-jdbc-store/src/main/java/org/apache/qpid/server/protocol/v1_0/store/jdbc/JDBCLinkStore.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-jdbc-store/src/main/java/org/apache/qpid/server/protocol/v1_0/store/jdbc/JDBCLinkStore.java Fri Mar 31 13:38:45 2017
@@ -52,6 +52,8 @@ import org.apache.qpid.server.protocol.v
 import org.apache.qpid.server.protocol.v1_0.store.AbstractLinkStore;
 import org.apache.qpid.server.protocol.v1_0.store.LinkStoreUpdater;
 import org.apache.qpid.server.protocol.v1_0.store.LinkStoreUtils;
+import org.apache.qpid.server.protocol.v1_0.type.BaseSource;
+import org.apache.qpid.server.protocol.v1_0.type.BaseTarget;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Source;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Target;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.TerminusDurability;
@@ -81,9 +83,9 @@ public class JDBCLinkStore extends Abstr
     }
 
     @Override
-    protected Collection<LinkDefinition> doOpenAndLoad(final LinkStoreUpdater updater) throws StoreException
+    protected Collection<LinkDefinition<Source, Target>> doOpenAndLoad(final LinkStoreUpdater updater) throws StoreException
     {
-        Collection<LinkDefinition> linkDefinitions;
+        Collection<LinkDefinition<Source, Target>> linkDefinitions;
         try
         {
             checkTransactionIsolationLevel();
@@ -117,7 +119,7 @@ public class JDBCLinkStore extends Abstr
     }
 
     @Override
-    protected void doSaveLink(final LinkDefinition link) throws StoreException
+    protected void doSaveLink(final LinkDefinition<Source, Target> link) throws StoreException
     {
         String linkKey = generateLinkKey(link);
         Connection connection = getConnection();
@@ -164,7 +166,7 @@ public class JDBCLinkStore extends Abstr
     }
 
     @Override
-    protected void doDeleteLink(final LinkDefinition link) throws StoreException
+    protected void doDeleteLink(final LinkDefinition<Source, Target> link) throws StoreException
     {
 
         try (Connection connection = getConnection();
@@ -276,10 +278,11 @@ public class JDBCLinkStore extends Abstr
         return _tableNamePrefix + VERSION_TABLE_NAME_SUFFIX;
     }
 
-    private Collection<LinkDefinition> performUpdate(final LinkStoreUpdater updater,
-                                                     Collection<LinkDefinition> linkDefinitions,
-                                                     final ModelVersion storedVersion,
-                                                     final ModelVersion currentVersion) throws SQLException
+    private Collection<LinkDefinition<Source, Target>> performUpdate(final LinkStoreUpdater updater,
+                                                                     Collection<LinkDefinition<Source, Target>> linkDefinitions,
+                                                                     final ModelVersion storedVersion,
+                                                                     final ModelVersion currentVersion)
+            throws SQLException
     {
         linkDefinitions = updater.update(storedVersion.toString(), linkDefinitions);
         Connection connection = getConnection();
@@ -292,7 +295,7 @@ public class JDBCLinkStore extends Abstr
                 statement.execute("DELETE FROM " + getLinksTableName());
             }
 
-            for (LinkDefinition linkDefinition : linkDefinitions)
+            for (LinkDefinition<? extends BaseSource, ? extends BaseTarget> linkDefinition : linkDefinitions)
             {
                 insert(connection, generateLinkKey(linkDefinition), linkDefinition);
             }
@@ -318,9 +321,9 @@ public class JDBCLinkStore extends Abstr
         return linkDefinitions;
     }
 
-    private Collection<LinkDefinition> getLinks() throws SQLException
+    private Collection<LinkDefinition<Source, Target>> getLinks() throws SQLException
     {
-        Collection<LinkDefinition> links = new ArrayList<>();
+        Collection<LinkDefinition<Source, Target>> links = new ArrayList<>();
         try (Connection connection = getConnection();
              Statement statement = connection.createStatement();
              ResultSet resultSet = statement.executeQuery(String.format(
@@ -335,7 +338,7 @@ public class JDBCLinkStore extends Abstr
                 Source source = (Source) getBlobAsAmqpObject(resultSet, 4);
                 Target target = (Target) getBlobAsAmqpObject(resultSet, 5);
 
-                links.add(new LinkDefinitionImpl(remoteContainerId, linkName, role, source, target));
+                links.add(new LinkDefinitionImpl<>(remoteContainerId, linkName, role, source, target));
             }
         }
         return links;
@@ -431,7 +434,9 @@ public class JDBCLinkStore extends Abstr
     }
 
 
-    private void insert(final Connection connection, final String linkKey, final LinkDefinition linkDefinition)
+    private void insert(final Connection connection,
+                        final String linkKey,
+                        final LinkDefinition<? extends BaseSource, ? extends BaseTarget> linkDefinition)
             throws SQLException
     {
         try (PreparedStatement statement = connection.prepareStatement(String.format(
@@ -451,7 +456,9 @@ public class JDBCLinkStore extends Abstr
         }
     }
 
-    private void update(final Connection connection, final String linkKey, final LinkDefinition linkDefinition)
+    private void update(final Connection connection,
+                        final String linkKey,
+                        final LinkDefinition<? extends BaseSource, ? extends BaseTarget> linkDefinition)
             throws SQLException
     {
         try (PreparedStatement statement = connection.prepareStatement(String.format(
@@ -500,7 +507,7 @@ public class JDBCLinkStore extends Abstr
         saveBytesAsBlob(statement, index, value.getBytes(UTF_8));
     }
 
-    private String generateLinkKey(final LinkDefinition linkDefinition)
+    private String generateLinkKey(final LinkDefinition<?, ?> linkDefinition)
     {
         MessageDigest md;
         try

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AbstractLinkEndpoint.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/AbstractLinkEndpoint.java?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AbstractLinkEndpoint.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AbstractLinkEndpoint.java Fri Mar 31 13:38:45 2017
@@ -47,10 +47,10 @@ 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.SenderSettleMode;
 
-public abstract class AbstractLinkEndpoint implements LinkEndpoint
+public abstract class AbstractLinkEndpoint<S extends BaseSource, T extends BaseTarget> implements LinkEndpoint<S, T>
 {
     private static final Logger LOGGER = LoggerFactory.getLogger(AbstractLinkEndpoint.class);
-    private final Link_1_0 _link;
+    private final Link_1_0<S, T> _link;
     private final Session_1_0 _session;
     private Object _flowTransactionId;
     private SenderSettleMode _sendingSettlementMode;
@@ -82,7 +82,7 @@ public abstract class AbstractLinkEndpoi
     }
 
 
-    AbstractLinkEndpoint(final Session_1_0 session, final Link_1_0 link)
+    AbstractLinkEndpoint(final Session_1_0 session, final Link_1_0<S, T> link)
     {
         _session = session;
         _link = link;
@@ -158,13 +158,13 @@ public abstract class AbstractLinkEndpoi
     }
 
     @Override
-    public BaseSource getSource()
+    public S getSource()
     {
         return _link.getSource();
     }
 
     @Override
-    public BaseTarget getTarget()
+    public T getTarget()
     {
         return _link.getTarget();
     }
@@ -468,7 +468,7 @@ public abstract class AbstractLinkEndpoi
         }
     }
 
-    public Link_1_0 getLink()
+    protected Link_1_0<S, T> getLink()
     {
         return _link;
     }

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AbstractReceivingLinkEndpoint.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?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AbstractReceivingLinkEndpoint.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:45 2017
@@ -27,11 +27,13 @@ import java.util.Map;
 
 import org.apache.qpid.server.protocol.v1_0.messaging.SectionDecoder;
 import org.apache.qpid.server.protocol.v1_0.messaging.SectionDecoderImpl;
+import org.apache.qpid.server.protocol.v1_0.type.BaseTarget;
 import org.apache.qpid.server.protocol.v1_0.type.Binary;
 import org.apache.qpid.server.protocol.v1_0.type.DeliveryState;
 import org.apache.qpid.server.protocol.v1_0.type.Outcome;
 import org.apache.qpid.server.protocol.v1_0.type.Symbol;
 import org.apache.qpid.server.protocol.v1_0.type.UnsignedInteger;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Source;
 import org.apache.qpid.server.protocol.v1_0.type.transaction.TransactionalState;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Attach;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Error;
@@ -39,7 +41,7 @@ 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 AbstractReceivingLinkEndpoint extends AbstractLinkEndpoint
+public abstract class AbstractReceivingLinkEndpoint<T extends BaseTarget> extends AbstractLinkEndpoint<Source, T>
 {
     private final SectionDecoder _sectionDecoder;
     private UnsignedInteger _lastDeliveryId;
@@ -88,7 +90,7 @@ public abstract class AbstractReceivingL
     }
 
 
-    public AbstractReceivingLinkEndpoint(final Session_1_0 session, final Link_1_0 link)
+    public AbstractReceivingLinkEndpoint(final Session_1_0 session, final Link_1_0<Source, T> link)
     {
         super(session, link);
         _sectionDecoder = new SectionDecoderImpl(session.getConnection()

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_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/ConsumerTarget_1_0.java?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java Fri Mar 31 13:38:45 2017
@@ -56,6 +56,7 @@ import org.apache.qpid.server.protocol.v
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Modified;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Rejected;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Released;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Target;
 import org.apache.qpid.server.protocol.v1_0.type.transaction.TransactionalState;
 import org.apache.qpid.server.protocol.v1_0.type.transport.SenderSettleMode;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Transfer;
@@ -605,9 +606,7 @@ class ConsumerTarget_1_0 extends Abstrac
     @Override
     public String getTargetAddress()
     {
-        BaseTarget target = _linkEndpoint.getTarget();
-
-        return target instanceof org.apache.qpid.server.protocol.v1_0.type.messaging.Target ? ((org.apache.qpid.server.protocol.v1_0.type.messaging.Target) target).getAddress() : _linkEndpoint.getLinkName();
+        return _linkEndpoint.getTarget().getAddress();
     }
 
     @Override

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Delivery.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/Delivery.java?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Delivery.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Delivery.java Fri Mar 31 13:38:45 2017
@@ -19,6 +19,8 @@
 
 package org.apache.qpid.server.protocol.v1_0;
 
+import org.apache.qpid.server.protocol.v1_0.type.BaseSource;
+import org.apache.qpid.server.protocol.v1_0.type.BaseTarget;
 import org.apache.qpid.server.protocol.v1_0.type.Binary;
 import org.apache.qpid.server.protocol.v1_0.type.UnsignedInteger;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Transfer;
@@ -27,12 +29,12 @@ public class Delivery
 {
     private final UnsignedInteger _deliveryId;
     private final Binary _deliveryTag;
-    private final LinkEndpoint _linkEndpoint;
+    private final LinkEndpoint<? extends BaseSource, ? extends BaseTarget> _linkEndpoint;
     private boolean _complete;
     private boolean _settled;
     private int _numberOfTransfers = 0;
 
-    public Delivery(Transfer transfer, final LinkEndpoint endpoint)
+    public Delivery(Transfer transfer, final LinkEndpoint<? extends BaseSource, ? extends BaseTarget> endpoint)
     {
         _settled = Boolean.TRUE.equals(transfer.getSettled());
         _deliveryId = transfer.getDeliveryId();
@@ -79,7 +81,7 @@ public class Delivery
         return _deliveryId;
     }
 
-    public LinkEndpoint getLinkEndpoint()
+    public LinkEndpoint<? extends BaseSource, ? extends BaseTarget> getLinkEndpoint()
     {
         return _linkEndpoint;
     }

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ErrantLinkEndpoint.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/ErrantLinkEndpoint.java?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ErrantLinkEndpoint.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ErrantLinkEndpoint.java Fri Mar 31 13:38:45 2017
@@ -30,14 +30,14 @@ import org.apache.qpid.server.protocol.v
 import org.apache.qpid.server.protocol.v1_0.type.transport.Flow;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Role;
 
-public class ErrantLinkEndpoint implements LinkEndpoint
+public class ErrantLinkEndpoint<S extends BaseSource, T extends BaseTarget> implements LinkEndpoint<S, T>
 {
-    private final Link_1_0 _link;
+    private final Link_1_0<S, T> _link;
     private final Session_1_0 _session;
     private Error _error;
     private UnsignedInteger _localHandle;
 
-    ErrantLinkEndpoint(Link_1_0 link, Session_1_0 session, Error error)
+    ErrantLinkEndpoint(Link_1_0<S, T> link, Session_1_0 session, Error error)
     {
         _link = link;
         _session = session;
@@ -51,13 +51,13 @@ public class ErrantLinkEndpoint implemen
     }
 
     @Override
-    public BaseSource getSource()
+    public S getSource()
     {
         return null;
     }
 
     @Override
-    public BaseTarget getTarget()
+    public T getTarget()
     {
         return null;
     }

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkDefinition.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/LinkDefinition.java?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkDefinition.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkDefinition.java Fri Mar 31 13:38:45 2017
@@ -25,7 +25,7 @@ import org.apache.qpid.server.protocol.v
 import org.apache.qpid.server.protocol.v1_0.type.BaseTarget;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Role;
 
-public interface LinkDefinition extends LinkModel
+public interface LinkDefinition<S extends BaseSource, T extends BaseTarget> extends LinkModel
 {
     String getRemoteContainerId();
 
@@ -33,7 +33,7 @@ public interface LinkDefinition extends
 
     Role getRole();
 
-    BaseSource getSource();
+    S getSource();
 
-    BaseTarget getTarget();
+    T getTarget();
 }

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkDefinitionImpl.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/LinkDefinitionImpl.java?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkDefinitionImpl.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkDefinitionImpl.java Fri Mar 31 13:38:45 2017
@@ -24,19 +24,19 @@ import org.apache.qpid.server.protocol.v
 import org.apache.qpid.server.protocol.v1_0.type.BaseTarget;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Role;
 
-public class LinkDefinitionImpl implements LinkDefinition
+public class LinkDefinitionImpl<S extends BaseSource, T extends BaseTarget> implements LinkDefinition<S, T>
 {
     private final String _remoteContainerId;
     private final String _name;
     private final Role _role;
-    private final BaseSource _source;
-    private final BaseTarget _target;
+    private final S _source;
+    private final T _target;
 
     public LinkDefinitionImpl(final String remoteContainerId,
                               final String name,
                               final Role role,
-                              final BaseSource source,
-                              final BaseTarget target)
+                              final S source,
+                              final T target)
     {
         _remoteContainerId = remoteContainerId;
         _name = name;
@@ -64,13 +64,13 @@ public class LinkDefinitionImpl implemen
     }
 
     @Override
-    public BaseSource getSource()
+    public S getSource()
     {
         return _source;
     }
 
     @Override
-    public BaseTarget getTarget()
+    public T getTarget()
     {
         return _target;
     }

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkEndpoint.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/LinkEndpoint.java?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkEndpoint.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkEndpoint.java Fri Mar 31 13:38:45 2017
@@ -30,13 +30,13 @@ import org.apache.qpid.server.protocol.v
 import org.apache.qpid.server.protocol.v1_0.type.transport.Flow;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Role;
 
-public interface LinkEndpoint
+public interface LinkEndpoint<S extends BaseSource, T extends BaseTarget>
 {
     Role getRole();
 
-    BaseSource getSource();
+    S getSource();
 
-    BaseTarget getTarget();
+    T getTarget();
 
     Session_1_0 getSession();
 

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=1789687&r1=1789686&r2=1789687&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:45 2017
@@ -30,6 +30,7 @@ import org.slf4j.LoggerFactory;
 import org.apache.qpid.server.protocol.v1_0.type.AmqpErrorException;
 import org.apache.qpid.server.protocol.v1_0.type.BaseSource;
 import org.apache.qpid.server.protocol.v1_0.type.BaseTarget;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Source;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Target;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.TerminusDurability;
 import org.apache.qpid.server.protocol.v1_0.type.transaction.Coordinator;
@@ -38,9 +39,8 @@ import org.apache.qpid.server.protocol.v
 import org.apache.qpid.server.protocol.v1_0.type.transport.Error;
 import org.apache.qpid.server.protocol.v1_0.type.transport.LinkError;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Role;
-import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
 
-public class LinkImpl implements Link_1_0
+public class LinkImpl<S extends BaseSource, T extends BaseTarget> implements Link_1_0<S, T>
 {
     private static final Logger LOGGER = LoggerFactory.getLogger(LinkImpl.class);
 
@@ -49,9 +49,9 @@ public class LinkImpl implements Link_1_
     private final Role _role;
     private final LinkRegistry _linkRegistry;
 
-    private volatile LinkEndpoint _linkEndpoint;
-    private volatile BaseSource _source;
-    private volatile BaseTarget _target;
+    private volatile LinkEndpoint<S, T> _linkEndpoint;
+    private volatile S _source;
+    private volatile T _target;
 
     public LinkImpl(final String remoteContainerId, final String linkName, final Role role, final LinkRegistry linkRegistry)
     {
@@ -61,14 +61,14 @@ public class LinkImpl implements Link_1_
         _linkRegistry = linkRegistry;
     }
 
-    public LinkImpl(final LinkDefinition linkDefinition, final LinkRegistry linkRegistry)
+    public LinkImpl(final LinkDefinition<S, T> linkDefinition, final LinkRegistry linkRegistry)
     {
         this(linkDefinition.getRemoteContainerId(), linkDefinition.getName(), linkDefinition.getRole(), linkRegistry);
         setTermini(linkDefinition.getSource(), linkDefinition.getTarget());
     }
 
     @Override
-    public final ListenableFuture<? extends LinkEndpoint> attach(final Session_1_0 session, final Attach attach)
+    public final ListenableFuture<? extends LinkEndpoint<S, T>> attach(final Session_1_0 session, final Attach attach)
     {
         try
         {
@@ -91,7 +91,7 @@ public class LinkImpl implements Link_1_
 
                 _linkEndpoint.receiveAttach(attach);
                 _linkRegistry.linkChanged(this);
-                return Futures.immediateFuture((LinkEndpoint) _linkEndpoint);
+                return Futures.immediateFuture(_linkEndpoint);
             }
         }
         catch (Exception e)
@@ -101,62 +101,59 @@ public class LinkImpl implements Link_1_
         }
     }
 
-    private synchronized ListenableFuture<LinkEndpoint> stealLink(final Session_1_0 session, final Attach attach)
+    private synchronized ListenableFuture<LinkEndpoint<S, T>> stealLink(final Session_1_0 session, final Attach attach)
     {
-        final SettableFuture<LinkEndpoint> returnFuture = SettableFuture.create();
-        _linkEndpoint.getSession().doOnIOThreadAsync(new Runnable()
-        {
-            @Override
-            public void run()
-            {
-                _linkEndpoint.close(new Error(LinkError.STOLEN,
-                                              String.format("Link is being stolen by connection '%s'",
-                                                            session.getConnection())));
-                try
-                {
-                    returnFuture.set(attach(session, attach).get());
-                }
-                catch (InterruptedException e)
+        final SettableFuture<LinkEndpoint<S, T>> returnFuture = SettableFuture.create();
+        _linkEndpoint.getSession().doOnIOThreadAsync(
+                () ->
                 {
-                    returnFuture.setException(e);
-                    Thread.currentThread().interrupt();
-                }
-                catch (ExecutionException e)
-                {
-                    returnFuture.setException(e.getCause());
-                }
-            }
-        });
+                    _linkEndpoint.close(new Error(LinkError.STOLEN,
+                                                  String.format("Link is being stolen by connection '%s'",
+                                                                session.getConnection())));
+                    try
+                    {
+                        returnFuture.set(attach(session, attach).get());
+                    }
+                    catch (InterruptedException e)
+                    {
+                        returnFuture.setException(e);
+                        Thread.currentThread().interrupt();
+                    }
+                    catch (ExecutionException e)
+                    {
+                        returnFuture.setException(e.getCause());
+                    }
+                });
         return returnFuture;
     }
 
-    private LinkEndpoint createLinkEndpoint(final Session_1_0 session, final Attach attach)
+    private LinkEndpoint<S, T> createLinkEndpoint(final Session_1_0 session, final Attach attach)
     {
-        LinkEndpoint linkEndpoint = null;
+        final LinkEndpoint<S, T> linkEndpoint;
         if (_role == Role.SENDER)
         {
-            linkEndpoint = new SendingLinkEndpoint(session, this);
+            linkEndpoint = (LinkEndpoint<S, T>) new SendingLinkEndpoint(session, (LinkImpl<Source, Target>) this);
         }
         else if (attach.getTarget() instanceof Coordinator)
         {
-            linkEndpoint = new TxnCoordinatorReceivingLinkEndpoint(session, this);
+            linkEndpoint = (LinkEndpoint<S, T>) new TxnCoordinatorReceivingLinkEndpoint(session, (LinkImpl<Source, Coordinator>) this);
         }
         else
         {
-            linkEndpoint = new StandardReceivingLinkEndpoint(session, this);
+            linkEndpoint = (LinkEndpoint<S, T>) new StandardReceivingLinkEndpoint(session, (LinkImpl<Source, Target>) this);
         }
         return linkEndpoint;
     }
 
-    private ListenableFuture<? extends LinkEndpoint> rejectLink(final Session_1_0 session, Throwable t)
+    private ListenableFuture<? extends LinkEndpoint<S, T>> rejectLink(final Session_1_0 session, Throwable t)
     {
         if (t instanceof AmqpErrorException)
         {
-            _linkEndpoint = new ErrantLinkEndpoint(this, session, ((AmqpErrorException) t).getError());
+            _linkEndpoint = new ErrantLinkEndpoint<>(this, session, ((AmqpErrorException) t).getError());
         }
         else
         {
-            _linkEndpoint = new ErrantLinkEndpoint(this, session, new Error(AmqpError.INTERNAL_ERROR, t.getMessage()));
+            _linkEndpoint = new ErrantLinkEndpoint<>(this, session, new Error(AmqpError.INTERNAL_ERROR, t.getMessage()));
         }
         return Futures.immediateFuture(_linkEndpoint);
     }
@@ -188,31 +185,31 @@ public class LinkImpl implements Link_1_
     }
 
     @Override
-    public BaseSource getSource()
+    public S getSource()
     {
         return _source;
     }
 
     @Override
-    public void setSource(BaseSource source)
+    public void setSource(S source)
     {
         setTermini(source, _target);
     }
 
     @Override
-    public BaseTarget getTarget()
+    public T getTarget()
     {
         return _target;
     }
 
     @Override
-    public void setTarget(BaseTarget target)
+    public void setTarget(T target)
     {
         setTermini(_source, target);
     }
 
     @Override
-    public void setTermini(BaseSource source, BaseTarget target)
+    public void setTermini(S source, T target)
     {
         _source = source;
         _target = target;

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkKey.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/LinkKey.java?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkKey.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkKey.java Fri Mar 31 13:38:45 2017
@@ -33,7 +33,7 @@ public class LinkKey
         _role = role;
     }
 
-    public LinkKey(final LinkDefinition link)
+    public LinkKey(final LinkDefinition<?, ?> link)
     {
         this(link.getRemoteContainerId(), link.getName(), link.getRole());
     }

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkRegistry.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/LinkRegistry.java?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkRegistry.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkRegistry.java Fri Mar 31 13:38:45 2017
@@ -20,14 +20,16 @@
 
 package org.apache.qpid.server.protocol.v1_0;
 
+import org.apache.qpid.server.protocol.v1_0.type.BaseSource;
+import org.apache.qpid.server.protocol.v1_0.type.BaseTarget;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.TerminusDurability;
 import org.apache.qpid.server.virtualhost.LinkRegistryModel;
 
 public interface LinkRegistry extends LinkRegistryModel
 {
-    void linkClosed(final Link_1_0 link);
+    void linkClosed(final Link_1_0<? extends BaseSource, ? extends BaseTarget> link);
 
-    void linkChanged(final Link_1_0 link);
+    void linkChanged(final Link_1_0<? extends BaseSource, ? extends BaseTarget> link);
 
     TerminusDurability getHighestSupportedTerminusDurability();
 }

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkRegistryImpl.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/LinkRegistryImpl.java?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkRegistryImpl.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkRegistryImpl.java Fri Mar 31 13:38:45 2017
@@ -32,6 +32,8 @@ import org.apache.qpid.server.plugin.Qpi
 import org.apache.qpid.server.protocol.v1_0.store.LinkStore;
 import org.apache.qpid.server.protocol.v1_0.store.LinkStoreFactory;
 import org.apache.qpid.server.protocol.v1_0.store.LinkStoreUpdaterImpl;
+import org.apache.qpid.server.protocol.v1_0.type.BaseSource;
+import org.apache.qpid.server.protocol.v1_0.type.BaseTarget;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Source;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Target;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.TerminusDurability;
@@ -41,8 +43,8 @@ import org.apache.qpid.server.util.Serve
 public class LinkRegistryImpl implements LinkRegistry
 {
     private static final Logger LOGGER = LoggerFactory.getLogger(LinkRegistryImpl.class);
-    private final ConcurrentMap<LinkKey, Link_1_0> _sendingLinkRegistry = new ConcurrentHashMap<>();
-    private final ConcurrentMap<LinkKey, Link_1_0> _receivingLinkRegistry = new ConcurrentHashMap<>();
+    private final ConcurrentMap<LinkKey, Link_1_0<? extends BaseSource, ? extends BaseTarget>> _sendingLinkRegistry = new ConcurrentHashMap<>();
+    private final ConcurrentMap<LinkKey, Link_1_0<? extends BaseSource, ? extends BaseTarget>> _receivingLinkRegistry = new ConcurrentHashMap<>();
 
     private final NamedAddressSpace _addressSpace;
 
@@ -71,35 +73,36 @@ public class LinkRegistryImpl implements
     }
 
     @Override
-    public Link_1_0 getSendingLink(final String remoteContainerId, final String linkName)
+    public Link_1_0<? extends BaseSource, ? extends BaseTarget> getSendingLink(final String remoteContainerId, final String linkName)
     {
         return getLinkFromRegistry(remoteContainerId, linkName, _sendingLinkRegistry, Role.SENDER);
     }
 
     @Override
-    public Link_1_0 getReceivingLink(final String remoteContainerId, final String linkName)
+    public Link_1_0<? extends BaseSource, ? extends BaseTarget> getReceivingLink(final String remoteContainerId, final String linkName)
     {
         return getLinkFromRegistry(remoteContainerId, linkName, _receivingLinkRegistry, Role.RECEIVER);
     }
 
     @Override
-    public void linkClosed(final Link_1_0 link)
+    public void linkClosed(final Link_1_0<? extends BaseSource, ? extends BaseTarget>  link)
     {
-        ConcurrentMap<LinkKey, Link_1_0> linkRegistry = getLinkRegistry(link.getRole());
+        ConcurrentMap<LinkKey, Link_1_0<? extends BaseSource, ? extends BaseTarget>> linkRegistry =
+                getLinkRegistry(link.getRole());
         linkRegistry.remove(new LinkKey(link));
-        _linkStore.deleteLink(link);
+        if (isDurableLink(link))
+        {
+            _linkStore.deleteLink((Link_1_0<Source, Target>) link);
+        }
     }
 
     @Override
-    public void linkChanged(final Link_1_0 link)
+    public void linkChanged(final Link_1_0<? extends BaseSource, ? extends BaseTarget> link)
     {
         getLinkRegistry(link.getRole()).putIfAbsent(new LinkKey(link), link);
-        if ((link.getRole() == Role.SENDER && link.getSource() instanceof Source
-             && ((Source) link.getSource()).getDurable() != TerminusDurability.NONE)
-            || (link.getRole() == Role.RECEIVER && link.getTarget() instanceof Target
-                && ((Target) link.getTarget()).getDurable() != TerminusDurability.NONE))
+        if (isDurableLink(link))
         {
-            _linkStore.saveLink(link);
+            _linkStore.saveLink((Link_1_0<Source, Target>) link);
         }
     }
 
@@ -113,11 +116,11 @@ public class LinkRegistryImpl implements
     @Override
     public void open()
     {
-        Collection<LinkDefinition> links = _linkStore.openAndLoad(new LinkStoreUpdaterImpl());
-        for(LinkDefinition link: links)
+        Collection<LinkDefinition<Source, Target>> links = _linkStore.openAndLoad(new LinkStoreUpdaterImpl());
+        for(LinkDefinition<? extends BaseSource, ? extends BaseTarget> link: links)
         {
-            ConcurrentMap<LinkKey, Link_1_0> linkRegistry = getLinkRegistry(link.getRole());
-            linkRegistry.put(new LinkKey(link), new LinkImpl(link, this));
+            ConcurrentMap<LinkKey, Link_1_0<? extends BaseSource, ? extends BaseTarget>> linkRegistry = getLinkRegistry(link.getRole());
+            linkRegistry.put(new LinkKey(link), new LinkImpl<>(link, this));
         }
     }
 
@@ -133,14 +136,22 @@ public class LinkRegistryImpl implements
         _linkStore.delete();
     }
 
-    private Link_1_0 getLinkFromRegistry(final String remoteContainerId,
-                                         final String linkName,
-                                         final ConcurrentMap<LinkKey, Link_1_0> linkRegistry,
-                                         final Role role)
+    private boolean isDurableLink(final Link_1_0<? extends BaseSource, ? extends BaseTarget> link)
+    {
+        return (link.getRole() == Role.SENDER && link.getSource() instanceof Source
+                && ((Source) link.getSource()).getDurable() != TerminusDurability.NONE)
+               || (link.getRole() == Role.RECEIVER && link.getTarget() instanceof Target
+                   && ((Target) link.getTarget()).getDurable() != TerminusDurability.NONE);
+    }
+
+    private Link_1_0<? extends BaseSource, ? extends BaseTarget> getLinkFromRegistry(final String remoteContainerId,
+                                                                                     final String linkName,
+                                                                                     final ConcurrentMap<LinkKey, Link_1_0<? extends BaseSource, ? extends BaseTarget>> linkRegistry,
+                                                                                     final Role role)
     {
         LinkKey linkKey = new LinkKey(remoteContainerId, linkName, role);
-        Link_1_0 newLink = new LinkImpl(remoteContainerId, linkName, role, this);
-        Link_1_0 link = linkRegistry.putIfAbsent(linkKey, newLink);
+        Link_1_0<? extends BaseSource, ? extends BaseTarget> newLink = new LinkImpl(remoteContainerId, linkName, role, this);
+        Link_1_0<? extends BaseSource, ? extends BaseTarget> link = linkRegistry.putIfAbsent(linkKey, newLink);
         if (link == null)
         {
             link = newLink;
@@ -148,9 +159,9 @@ public class LinkRegistryImpl implements
         return link;
     }
 
-    private ConcurrentMap<LinkKey, Link_1_0> getLinkRegistry(final Role role)
+    private ConcurrentMap<LinkKey, Link_1_0<? extends BaseSource, ? extends BaseTarget>> getLinkRegistry(final Role role)
     {
-        ConcurrentMap<LinkKey, Link_1_0> linkRegistry;
+        ConcurrentMap<LinkKey, Link_1_0<? extends BaseSource, ? extends BaseTarget>> linkRegistry;
         if (Role.SENDER == role)
         {
             linkRegistry = _sendingLinkRegistry;

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Link_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/Link_1_0.java?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Link_1_0.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Link_1_0.java Fri Mar 31 13:38:45 2017
@@ -27,19 +27,19 @@ import org.apache.qpid.server.protocol.v
 import org.apache.qpid.server.protocol.v1_0.type.messaging.TerminusDurability;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Attach;
 
-public interface Link_1_0 extends LinkDefinition
+public interface Link_1_0<S extends BaseSource, T extends BaseTarget> extends LinkDefinition<S, T>
 {
-    ListenableFuture<? extends LinkEndpoint> attach(Session_1_0 session, final Attach attach);
+    ListenableFuture<? extends LinkEndpoint<S, T>> attach(Session_1_0 session, final Attach attach);
 
     void linkClosed();
 
     void discardEndpoint();
 
-    void setSource(BaseSource source);
+    void setSource(S source);
 
-    void setTarget(BaseTarget target);
+    void setTarget(T target);
 
-    void setTermini(BaseSource source, BaseTarget target);
+    void setTermini(S source, T target);
 
     TerminusDurability getHighestSupportedTerminusDurability();
 }

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/NullLinkStoreFactory.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/NullLinkStoreFactory.java?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/NullLinkStoreFactory.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/NullLinkStoreFactory.java Fri Mar 31 13:38:45 2017
@@ -28,9 +28,12 @@ import org.apache.qpid.server.plugin.Plu
 import org.apache.qpid.server.protocol.v1_0.store.LinkStore;
 import org.apache.qpid.server.protocol.v1_0.store.LinkStoreFactory;
 import org.apache.qpid.server.protocol.v1_0.store.LinkStoreUpdater;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Source;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Target;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.TerminusDurability;
 import org.apache.qpid.server.store.StoreException;
 
+@SuppressWarnings("unused")
 @PluggableService
 public class NullLinkStoreFactory implements LinkStoreFactory
 {
@@ -46,8 +49,7 @@ public class NullLinkStoreFactory implem
         return new LinkStore()
         {
             @Override
-            public Collection<LinkDefinition> openAndLoad(final LinkStoreUpdater updater)
-                    throws StoreException, StoreException
+            public Collection<LinkDefinition<Source, Target>> openAndLoad(final LinkStoreUpdater updater) throws StoreException
             {
                 return Collections.emptyList();
             }
@@ -58,12 +60,12 @@ public class NullLinkStoreFactory implem
             }
 
             @Override
-            public void saveLink(final LinkDefinition link)
+            public void saveLink(final LinkDefinition<Source, Target> link)
             {
             }
 
             @Override
-            public void deleteLink(final LinkDefinition link)
+            public void deleteLink(final LinkDefinition<Source, Target> link)
             {
             }
 

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=1789687&r1=1789686&r2=1789687&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:45 2017
@@ -48,6 +48,7 @@ import org.apache.qpid.server.model.Name
 import org.apache.qpid.server.model.NotFoundException;
 import org.apache.qpid.server.model.Queue;
 import org.apache.qpid.server.protocol.v1_0.type.AmqpErrorException;
+import org.apache.qpid.server.protocol.v1_0.type.BaseTarget;
 import org.apache.qpid.server.protocol.v1_0.type.Binary;
 import org.apache.qpid.server.protocol.v1_0.type.DeliveryState;
 import org.apache.qpid.server.protocol.v1_0.type.Outcome;
@@ -75,7 +76,7 @@ import org.apache.qpid.server.txn.Server
 import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
 import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
 
-public class SendingLinkEndpoint extends AbstractLinkEndpoint
+public class SendingLinkEndpoint extends AbstractLinkEndpoint<Source, Target>
 {
     private static final Logger LOGGER = LoggerFactory.getLogger(SendingLinkEndpoint.class);
 
@@ -96,7 +97,7 @@ public class SendingLinkEndpoint extends
     private ConsumerTarget_1_0 _consumerTarget;
     private MessageInstanceConsumer<ConsumerTarget_1_0> _consumer;
 
-    public SendingLinkEndpoint(final Session_1_0 session, final LinkImpl link)
+    public SendingLinkEndpoint(final Session_1_0 session, final LinkImpl<Source, Target> link)
     {
         super(session, link);
         setDeliveryCount(UnsignedInteger.valueOf(0));
@@ -113,7 +114,7 @@ public class SendingLinkEndpoint extends
     {
         // TODO FIXME: this method might modify the source. this is not good encapsulation. furthermore if it does so then it should inform the link/linkregistry about it!
         _destination = destination;
-        final Source source = (Source) getSource();
+        final Source source = getSource();
 
         EnumSet<ConsumerOption> options = EnumSet.noneOf(ConsumerOption.class);
 
@@ -199,22 +200,12 @@ public class SendingLinkEndpoint extends
 
     void createConsumerTarget() throws AmqpErrorException
     {
-        final Source source = (Source) getSource();
+        final Source source = getSource();
         _consumerTarget = new ConsumerTarget_1_0(this,
                                          _destination instanceof ExchangeDestination ? true : source.getDistributionMode() != StdDistMode.COPY);
         try
         {
-            final String name;
-            if(getTarget() instanceof Target)
-            {
-                Target target = (Target) getTarget();
-                name = target.getAddress() == null ? getLinkName() : target.getAddress();
-            }
-            else
-            {
-                name = getLinkName();
-            }
-
+            final String name = getTarget().getAddress() == null ? getLinkName() : getTarget().getAddress();
             _consumer = _destination.getMessageSource()
                                     .addConsumer(_consumerTarget,
                                                  _consumerFilters,
@@ -299,7 +290,7 @@ public class SendingLinkEndpoint extends
         }
 
         Source newSource = (Source) attach.getSource();
-        Source oldSource = (Source) getSource();
+        Source oldSource = getSource();
 
         final SendingDestination destination = getSession().getSendingDestination(getLinkName(), oldSource);
         prepareConsumerOptionsAndFilters(destination);
@@ -330,7 +321,7 @@ public class SendingLinkEndpoint extends
         }
 
         Source newSource = (Source) attach.getSource();
-        Source oldSource = (Source) getSource();
+        Source oldSource = getSource();
 
         final SendingDestination destination = getSession().getSendingDestination(getLinkName(), oldSource);
         prepareConsumerOptionsAndFilters(destination);
@@ -368,7 +359,7 @@ public class SendingLinkEndpoint extends
             throw new AmqpErrorException(new Error(AmqpError.NOT_FOUND, ""));
         }
 
-        final SendingDestination destination = getSession().getSendingDestination(getLinkName(), (Source) getSource());
+        final SendingDestination destination = getSession().getSendingDestination(getLinkName(), getSource());
         prepareConsumerOptionsAndFilters(destination);
 
         attachReceived(attach);
@@ -387,7 +378,7 @@ public class SendingLinkEndpoint extends
 
     public TerminusDurability getTerminusDurability()
     {
-        return ((Source) getSource()).getDurable();
+        return getSource().getDurable();
     }
 
     public boolean transfer(final Transfer xfr, final boolean decrementCredit)
@@ -504,7 +495,7 @@ public class SendingLinkEndpoint extends
     {
         getConsumerTarget().close();
 
-        TerminusExpiryPolicy expiryPolicy = ((Source) getSource()).getExpiryPolicy();
+        TerminusExpiryPolicy expiryPolicy = (getSource()).getExpiryPolicy();
         if (Boolean.TRUE.equals(detach.getClosed())
             || TerminusExpiryPolicy.LINK_DETACH.equals(expiryPolicy)
             || (TerminusExpiryPolicy.SESSION_END.equals(expiryPolicy) && getSession().isClosing())
@@ -633,7 +624,7 @@ public class SendingLinkEndpoint extends
         super.attachReceived(attach);
 
         Target target = (Target) attach.getTarget();
-        Source source = (Source) getSource();
+        Source source = getSource();
         if (source == null)
         {
             source = new Source();

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=1789687&r1=1789686&r2=1789687&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:45 2017
@@ -74,6 +74,8 @@ import org.apache.qpid.server.model.Sess
 import org.apache.qpid.server.protocol.v1_0.codec.QpidByteBufferUtils;
 import org.apache.qpid.server.protocol.v1_0.framing.OversizeFrameException;
 import org.apache.qpid.server.protocol.v1_0.type.AmqpErrorException;
+import org.apache.qpid.server.protocol.v1_0.type.BaseSource;
+import org.apache.qpid.server.protocol.v1_0.type.BaseTarget;
 import org.apache.qpid.server.protocol.v1_0.type.Binary;
 import org.apache.qpid.server.protocol.v1_0.type.DeliveryState;
 import org.apache.qpid.server.protocol.v1_0.type.ErrorCondition;
@@ -132,9 +134,9 @@ public class Session_1_0 extends Abstrac
 
     private SessionState _sessionState;
 
-    private final Map<LinkEndpoint, UnsignedInteger> _endpointToOutputHandle = new HashMap<>();
-    private final Map<UnsignedInteger, LinkEndpoint> _inputHandleToEndpoint = new HashMap<>();
-    private final Set<LinkEndpoint> _associatedLinkEndpoints = new HashSet<>();
+    private final Map<LinkEndpoint<? extends BaseSource, ? extends BaseTarget>, UnsignedInteger> _endpointToOutputHandle = new HashMap<>();
+    private final Map<UnsignedInteger, LinkEndpoint<? extends BaseSource, ? extends BaseTarget>> _inputHandleToEndpoint = new HashMap<>();
+    private final Set<LinkEndpoint<? extends BaseSource, ? extends BaseTarget>> _associatedLinkEndpoints = new HashSet<>();
 
     private final short _receivingChannel;
     private final short _sendingChannel;
@@ -211,7 +213,7 @@ public class Session_1_0 extends Abstrac
             }
             else
             {
-                final Link_1_0 link;
+                final Link_1_0<? extends BaseSource, ? extends BaseTarget> link;
                 if (attach.getRole() == Role.RECEIVER)
                 {
                     link = getAddressSpace().getSendingLink(getConnection().getRemoteContainerId(), attach.getName());
@@ -221,7 +223,7 @@ public class Session_1_0 extends Abstrac
                     link = getAddressSpace().getReceivingLink(getConnection().getRemoteContainerId(), attach.getName());
                 }
 
-                final ListenableFuture<? extends LinkEndpoint> future = link.attach(this, attach);
+                final ListenableFuture<? extends LinkEndpoint<?,?>> future = link.attach(this, attach);
 
                 addFutureCallback(future, new EndpointCreationCallback(attach), MoreExecutors.directExecutor());
             }
@@ -408,7 +410,7 @@ public class Session_1_0 extends Abstrac
     public void receiveFlow(final Flow flow)
     {
         UnsignedInteger handle = flow.getHandle();
-        final LinkEndpoint endpoint = handle == null ? null : _inputHandleToEndpoint.get(handle);
+        final LinkEndpoint<? extends BaseSource, ? extends BaseTarget> endpoint = handle == null ? null : _inputHandleToEndpoint.get(handle);
 
         final UnsignedInteger nextOutgoingId =
                 flow.getNextIncomingId() == null ? _initialOutgoingId : flow.getNextIncomingId();
@@ -421,8 +423,8 @@ public class Session_1_0 extends Abstrac
         }
         else
         {
-            final Collection<LinkEndpoint> allLinkEndpoints = _inputHandleToEndpoint.values();
-            for (LinkEndpoint le : allLinkEndpoints)
+            final Collection<LinkEndpoint<? extends BaseSource, ? extends BaseTarget>> allLinkEndpoints = _inputHandleToEndpoint.values();
+            for (LinkEndpoint<? extends BaseSource, ? extends BaseTarget> le : allLinkEndpoints)
             {
                 le.flowStateChanged();
             }
@@ -566,7 +568,7 @@ public class Session_1_0 extends Abstrac
         _nextIncomingTransferId.incr();
 
         UnsignedInteger inputHandle = transfer.getHandle();
-        LinkEndpoint linkEndpoint = _inputHandleToEndpoint.get(inputHandle);
+        LinkEndpoint<? extends BaseSource, ? extends BaseTarget> linkEndpoint = _inputHandleToEndpoint.get(inputHandle);
 
         if (linkEndpoint == null)
         {
@@ -762,7 +764,7 @@ public class Session_1_0 extends Abstrac
                 ExchangeDestination newExchangeDestination = (ExchangeDestination) newDestination;
                 if (oldExchangeDestination.getQueue() != newExchangeDestination.getQueue())
                 {
-                    Source oldSource = (Source) linkEndpoint.getSource();
+                    Source oldSource = linkEndpoint.getSource();
                     oldSource.setAddress(newAddress);
                     oldSource.setFilter(newSource.getFilter());
                     return true;
@@ -1108,8 +1110,8 @@ public class Session_1_0 extends Abstrac
 
     void remoteEnd(End end)
     {
-        Set<LinkEndpoint> associatedLinkEndpoints = new HashSet<>(_associatedLinkEndpoints);
-        for (LinkEndpoint linkEndpoint : associatedLinkEndpoints)
+        Set<LinkEndpoint<? extends BaseSource, ? extends BaseTarget>> associatedLinkEndpoints = new HashSet<>(_associatedLinkEndpoints);
+        for (LinkEndpoint<? extends BaseSource, ? extends BaseTarget> linkEndpoint : associatedLinkEndpoints)
         {
             linkEndpoint.remoteDetached(new Detach());
             linkEndpoint.destroy();
@@ -1200,7 +1202,7 @@ public class Session_1_0 extends Abstrac
     @Override
     public void transportStateChanged()
     {
-        for (LinkEndpoint linkEndpoint : _endpointToOutputHandle.keySet())
+        for (LinkEndpoint<? extends BaseSource, ? extends BaseTarget> linkEndpoint : _endpointToOutputHandle.keySet())
         {
             if (linkEndpoint instanceof SendingLinkEndpoint)
             {
@@ -1236,7 +1238,7 @@ public class Session_1_0 extends Abstrac
         {
             messageWithSubject(ChannelMessages.FLOW_ENFORCED(queue.getName()));
 
-            for (LinkEndpoint linkEndpoint : _endpointToOutputHandle.keySet())
+            for (LinkEndpoint<? extends BaseSource, ? extends BaseTarget> linkEndpoint : _endpointToOutputHandle.keySet())
             {
                 if (linkEndpoint instanceof AbstractReceivingLinkEndpoint
                     && isQueueDestinationForLink(queue, ((AbstractReceivingLinkEndpoint) linkEndpoint).getReceivingDestination()))
@@ -1276,7 +1278,7 @@ public class Session_1_0 extends Abstrac
             {
                 messageWithSubject(ChannelMessages.FLOW_REMOVED());
             }
-            for (LinkEndpoint linkEndpoint : _endpointToOutputHandle.keySet())
+            for (LinkEndpoint<? extends BaseSource, ? extends BaseTarget> linkEndpoint : _endpointToOutputHandle.keySet())
             {
                 if (linkEndpoint instanceof AbstractReceivingLinkEndpoint
                         && isQueueDestinationForLink(queue, ((AbstractReceivingLinkEndpoint) linkEndpoint).getReceivingDestination()))
@@ -1307,7 +1309,7 @@ public class Session_1_0 extends Abstrac
         {
             messageWithSubject(ChannelMessages.FLOW_ENFORCED("** All Queues **"));
 
-            for (LinkEndpoint linkEndpoint : _endpointToOutputHandle.keySet())
+            for (LinkEndpoint<? extends BaseSource, ? extends BaseTarget> linkEndpoint : _endpointToOutputHandle.keySet())
             {
                 if (linkEndpoint instanceof AbstractReceivingLinkEndpoint)
                 {
@@ -1339,7 +1341,7 @@ public class Session_1_0 extends Abstrac
             {
                 messageWithSubject(ChannelMessages.FLOW_REMOVED());
             }
-            for (LinkEndpoint linkEndpoint : _endpointToOutputHandle.keySet())
+            for (LinkEndpoint<? extends BaseSource, ? extends BaseTarget> linkEndpoint : _endpointToOutputHandle.keySet())
             {
                 if (linkEndpoint instanceof AbstractReceivingLinkEndpoint
                     && !_blockingEntities.contains(((AbstractReceivingLinkEndpoint) linkEndpoint).getReceivingDestination()))
@@ -1512,9 +1514,9 @@ public class Session_1_0 extends Abstrac
         return "Session_1_0[" + _connection + ": " + _sendingChannel + ']';
     }
 
-    public void dissociateEndpoint(LinkEndpoint linkEndpoint)
+    public void dissociateEndpoint(LinkEndpoint<? extends BaseSource, ? extends BaseTarget> linkEndpoint)
     {
-        for (Map.Entry<UnsignedInteger, LinkEndpoint> entry : _inputHandleToEndpoint.entrySet())
+        for (Map.Entry<UnsignedInteger, LinkEndpoint<? extends BaseSource, ? extends BaseTarget>> entry : _inputHandleToEndpoint.entrySet())
         {
             if (entry.getValue() == linkEndpoint)
             {
@@ -1530,7 +1532,7 @@ public class Session_1_0 extends Abstrac
     {
         if(_inputHandleToEndpoint.containsKey(handle))
         {
-            LinkEndpoint endpoint = _inputHandleToEndpoint.remove(handle);
+            LinkEndpoint<? extends BaseSource, ? extends BaseTarget> endpoint = _inputHandleToEndpoint.remove(handle);
             endpoint.remoteDetached(detach);
             _endpointToOutputHandle.remove(endpoint);
         }
@@ -1602,7 +1604,7 @@ public class Session_1_0 extends Abstrac
         return primaryDomain;
     }
 
-    private class EndpointCreationCallback<T extends LinkEndpoint> implements FutureCallback<T>
+    private class EndpointCreationCallback<T extends LinkEndpoint<? extends BaseSource, ? extends BaseTarget>> implements FutureCallback<T>
     {
 
         private final Attach _attach;
@@ -1613,7 +1615,7 @@ public class Session_1_0 extends Abstrac
         }
 
         @Override
-        public void onSuccess(final LinkEndpoint endpoint)
+        public void onSuccess(final T endpoint)
         {
             doOnIOThreadAsync(new Runnable()
             {

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=1789687&r1=1789686&r2=1789687&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:45 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 AbstractReceivingLinkEndpoint
+public class StandardReceivingLinkEndpoint extends AbstractReceivingLinkEndpoint<Target>
 {
     private static final Logger LOGGER = LoggerFactory.getLogger(StandardReceivingLinkEndpoint.class);
 
@@ -79,7 +79,7 @@ public class StandardReceivingLinkEndpoi
     private Map<Binary, Outcome> _unsettledMap = Collections.synchronizedMap(new HashMap<Binary, Outcome>());
 
     public StandardReceivingLinkEndpoint(final Session_1_0 session,
-                                         final Link_1_0 link)
+                                         final Link_1_0<Source, Target> link)
     {
         super(session, link);
     }
@@ -93,7 +93,7 @@ public class StandardReceivingLinkEndpoi
 
     private TerminusDurability getDurability()
     {
-        return ((Target) getTarget()).getDurable();
+        return getTarget().getDurable();
     }
 
     @Override
@@ -241,7 +241,7 @@ public class StandardReceivingLinkEndpoi
                     getReceivingDestination().authorizePublish(session.getSecurityToken(), routingAddress);
 
                     Outcome outcome = getReceivingDestination().send(serverMessage, routingAddress, transaction, null);
-                    Source source = (Source) getSource();
+                    Source source = getSource();
 
                     DeliveryState resultantState;
 
@@ -497,16 +497,16 @@ public class StandardReceivingLinkEndpoi
     @Override
     protected void recoverLink(final Attach attach) throws AmqpErrorException
     {
-        if (getTarget() == null || !(getTarget() instanceof Target))
+        if (getTarget() == null)
         {
             throw new AmqpErrorException(new Error(AmqpError.NOT_FOUND,
                                                    String.format("Link '%s' not found", getLinkName())));
         }
 
         Source source = (Source) attach.getSource();
-        Target target = (Target) getTarget();
+        Target target = getTarget();
 
-        final ReceivingDestination destination = getSession().getReceivingDestination((Target) getTarget());
+        final ReceivingDestination destination = getSession().getReceivingDestination(getTarget());
         target.setCapabilities(destination.getCapabilities());
         setCapabilities(Arrays.asList(destination.getCapabilities()));
         setDestination(destination);

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=1789687&r1=1789686&r2=1789687&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:45 2017
@@ -47,12 +47,12 @@ 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 AbstractReceivingLinkEndpoint
+public class TxnCoordinatorReceivingLinkEndpoint extends AbstractReceivingLinkEndpoint<Coordinator>
 {
     private final LinkedHashMap<Integer, ServerTransaction> _createdTransactions = new LinkedHashMap<>();
     private ArrayList<Transfer> _incompleteMessage;
 
-    public TxnCoordinatorReceivingLinkEndpoint(final Session_1_0 session, final Link_1_0 link)
+    public TxnCoordinatorReceivingLinkEndpoint(final Session_1_0 session, final Link_1_0<Source, Coordinator> link)
     {
         super(session, link);
     }

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/store/AbstractLinkStore.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/store/AbstractLinkStore.java?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/store/AbstractLinkStore.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/store/AbstractLinkStore.java Fri Mar 31 13:38:45 2017
@@ -23,7 +23,8 @@ import java.util.Collection;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 import org.apache.qpid.server.protocol.v1_0.LinkDefinition;
-import org.apache.qpid.server.protocol.v1_0.LinkKey;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Source;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Target;
 import org.apache.qpid.server.store.StoreException;
 
 public abstract class AbstractLinkStore implements LinkStore
@@ -31,14 +32,14 @@ public abstract class AbstractLinkStore
     private final ReentrantReadWriteLock _useOrCloseRWLock = new ReentrantReadWriteLock(true);
     private volatile StoreState _storeState = StoreState.CLOSED;
 
-    protected abstract Collection<LinkDefinition> doOpenAndLoad(final LinkStoreUpdater updater);
+    protected abstract Collection<LinkDefinition<Source, Target>> doOpenAndLoad(final LinkStoreUpdater updater);
     protected abstract void doClose();
     protected abstract void doDelete();
-    protected abstract void doSaveLink(final LinkDefinition link);
-    protected abstract void doDeleteLink(final LinkDefinition link);
+    protected abstract void doSaveLink(final LinkDefinition<Source, Target> link);
+    protected abstract void doDeleteLink(final LinkDefinition<Source, Target> link);
 
     @Override
-    public final Collection<LinkDefinition> openAndLoad(final LinkStoreUpdater updater) throws StoreException
+    public final Collection<LinkDefinition<Source, Target>> openAndLoad(final LinkStoreUpdater updater) throws StoreException
     {
         _useOrCloseRWLock.readLock().lock();
         try
@@ -48,7 +49,7 @@ public abstract class AbstractLinkStore
                 throw new StoreException("Store is already opened");
             }
 
-            Collection<LinkDefinition> linkDefinitions = doOpenAndLoad(updater);
+            Collection<LinkDefinition<Source, Target>> linkDefinitions = doOpenAndLoad(updater);
             _storeState = StoreState.OPENED;
             return linkDefinitions;
         }
@@ -74,7 +75,7 @@ public abstract class AbstractLinkStore
     }
 
     @Override
-    public final void saveLink(final LinkDefinition link) throws StoreException
+    public final void saveLink(final LinkDefinition<Source, Target> link) throws StoreException
     {
         _useOrCloseRWLock.readLock().lock();
         try
@@ -93,7 +94,7 @@ public abstract class AbstractLinkStore
     }
 
     @Override
-    public final void deleteLink(final LinkDefinition link) throws StoreException
+    public final void deleteLink(final LinkDefinition<Source, Target> link) throws StoreException
     {
         _useOrCloseRWLock.readLock().lock();
         try

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/store/LinkStore.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/store/LinkStore.java?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/store/LinkStore.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/store/LinkStore.java Fri Mar 31 13:38:45 2017
@@ -22,19 +22,21 @@ package org.apache.qpid.server.protocol.
 import java.util.Collection;
 
 import org.apache.qpid.server.protocol.v1_0.LinkDefinition;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Source;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Target;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.TerminusDurability;
 import org.apache.qpid.server.store.StoreException;
 
 
 public interface LinkStore
 {
-    Collection<LinkDefinition> openAndLoad(LinkStoreUpdater updater) throws StoreException;
+    Collection<LinkDefinition<Source, Target>> openAndLoad(LinkStoreUpdater updater) throws StoreException;
 
     void close() throws StoreException;
 
-    void saveLink(LinkDefinition link) throws StoreException;
+    void saveLink(LinkDefinition<Source, Target> link) throws StoreException;
 
-    void deleteLink(LinkDefinition link) throws StoreException;
+    void deleteLink(LinkDefinition<Source, Target> link) throws StoreException;
 
     void delete();
 

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/store/LinkStoreUpdater.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/store/LinkStoreUpdater.java?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/store/LinkStoreUpdater.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/store/LinkStoreUpdater.java Fri Mar 31 13:38:45 2017
@@ -23,8 +23,10 @@ package org.apache.qpid.server.protocol.
 import java.util.Collection;
 
 import org.apache.qpid.server.protocol.v1_0.LinkDefinition;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Source;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Target;
 
 public interface LinkStoreUpdater
 {
-    Collection<LinkDefinition> update(String currentVersion, Collection<LinkDefinition> linkDefinitions);
+    Collection<LinkDefinition<Source, Target>> update(String currentVersion, Collection<LinkDefinition<Source, Target>> linkDefinitions);
 }

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/store/LinkStoreUpdaterImpl.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/store/LinkStoreUpdaterImpl.java?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/store/LinkStoreUpdaterImpl.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/store/LinkStoreUpdaterImpl.java Fri Mar 31 13:38:45 2017
@@ -23,12 +23,14 @@ package org.apache.qpid.server.protocol.
 import java.util.Collection;
 
 import org.apache.qpid.server.protocol.v1_0.LinkDefinition;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Source;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Target;
 
 public class LinkStoreUpdaterImpl implements LinkStoreUpdater
 {
     @Override
-    public Collection<LinkDefinition> update(final String currentVersion,
-                                             final Collection<LinkDefinition> linkDefinitions)
+    public Collection<LinkDefinition<Source, Target>> update(final String currentVersion,
+                                                             final Collection<LinkDefinition<Source, Target>> linkDefinitions)
     {
         return linkDefinitions;
     }

Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/TestLinkStoreFactory.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/TestLinkStoreFactory.java?rev=1789687&r1=1789686&r2=1789687&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/TestLinkStoreFactory.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/TestLinkStoreFactory.java Fri Mar 31 13:38:45 2017
@@ -28,6 +28,8 @@ import org.apache.qpid.server.plugin.Plu
 import org.apache.qpid.server.protocol.v1_0.store.LinkStore;
 import org.apache.qpid.server.protocol.v1_0.store.LinkStoreFactory;
 import org.apache.qpid.server.protocol.v1_0.store.LinkStoreUpdater;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Source;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Target;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.TerminusDurability;
 import org.apache.qpid.server.store.StoreException;
 import org.apache.qpid.server.virtualhost.TestMemoryVirtualHost;
@@ -48,7 +50,7 @@ public class TestLinkStoreFactory implem
         return new LinkStore()
         {
             @Override
-            public Collection<LinkDefinition> openAndLoad(final LinkStoreUpdater updater) throws StoreException
+            public Collection<LinkDefinition<Source, Target>> openAndLoad(final LinkStoreUpdater updater) throws StoreException
             {
                 return Collections.emptyList();
             }
@@ -60,13 +62,13 @@ public class TestLinkStoreFactory implem
             }
 
             @Override
-            public void saveLink(final LinkDefinition link) throws StoreException
+            public void saveLink(final LinkDefinition<Source, Target> link) throws StoreException
             {
 
             }
 
             @Override
-            public void deleteLink(final LinkDefinition link) throws StoreException
+            public void deleteLink(final LinkDefinition<Source, Target> link) throws StoreException
             {
 
             }



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org


Mime
View raw message