infinispan · pruivo · Sep 20, 2021 · Jan 13, 2021
diff --git a/build-configuration/pom.xml b/build-configuration/pom.xml
@@ -160,7 +160,7 @@
       <version.jboss.shrinkwrap.resolver>3.1.3</version.jboss.shrinkwrap.resolver>
       <version.jboss.xnio>3.7.7.Final</version.jboss.xnio>
       <version.jcip-annotations>1.0</version.jcip-annotations>
-      <version.jgroups>4.2.14.Final</version.jgroups>
+      <version.jgroups>4.2.17.Final</version.jgroups>
       <version.jsr107>1.1.0</version.jsr107>
       <version.junit>4.13.1</version.junit>
       <version.junit5>5.6.2</version.junit5>

diff --git a/core/src/main/java/org/infinispan/configuration/global/TransportConfigurationBuilder.java b/core/src/main/java/org/infinispan/configuration/global/TransportConfigurationBuilder.java
@@ -165,7 +165,7 @@ public TransportConfigurationBuilder clearProperties() {
       return this;
    }
 
-   public TransportConfigurationBuilder addProperty(String key, String value) {
+   public TransportConfigurationBuilder addProperty(String key, Object value) {
       typedProperties.put(key, value);
       return this;
    }

diff --git a/core/src/main/java/org/infinispan/configuration/parsing/AbstractParserContext.java b/core/src/main/java/org/infinispan/configuration/parsing/AbstractParserContext.java
diff --git a/core/src/main/java/org/infinispan/configuration/parsing/ConfigurationBuilderHolder.java b/core/src/main/java/org/infinispan/configuration/parsing/ConfigurationBuilderHolder.java
@@ -4,8 +4,10 @@
 
 import java.lang.ref.WeakReference;
 import java.util.ArrayDeque;
+import java.util.ArrayList;
 import java.util.Deque;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 import org.infinispan.commons.configuration.io.ConfigurationReader;
@@ -22,11 +24,11 @@ public class ConfigurationBuilderHolder implements ConfigurationReaderContext {
    private final GlobalConfigurationBuilder globalConfigurationBuilder;
    private final Map<String, ConfigurationBuilder> namedConfigurationBuilders;
    private ConfigurationBuilder currentConfigurationBuilder;
-   private final Map<Class<? extends ConfigurationParser>, ParserContext> parserContexts;
    private final WeakReference<ClassLoader> classLoader;
    private final Deque<String> scope;
    private final JGroupsConfigurationBuilder jgroupsBuilder;
    private NamespaceMappingParser namespaceMappingParser;
+   private final List<ConfigurationParserListener> listeners = new ArrayList<>();
 
    public ConfigurationBuilderHolder() {
       this(Thread.currentThread().getContextClassLoader());
@@ -39,7 +41,6 @@ public ConfigurationBuilderHolder(ClassLoader classLoader) {
    public ConfigurationBuilderHolder(ClassLoader classLoader, GlobalConfigurationBuilder globalConfigurationBuilder) {
       this.globalConfigurationBuilder = globalConfigurationBuilder;
       this.namedConfigurationBuilders = new HashMap<>();
-      this.parserContexts = new HashMap<>();
       this.jgroupsBuilder = this.globalConfigurationBuilder.transport().jgroups();
       this.classLoader = new WeakReference<>(classLoader);
       scope = new ArrayDeque<>();
@@ -97,23 +98,20 @@ public String getScope() {
       return scope.peek();
    }
 
-   @SuppressWarnings("unchecked")
-   public <T extends ParserContext> T getParserContext(Class<? extends ConfigurationParser> parserClass) {
-      return (T) parserContexts.get(parserClass);
+   public void addParserListener(ConfigurationParserListener listener) {
+      listeners.add(listener);
    }
 
-   public void setParserContext(Class<? extends ConfigurationParser> parserClass, ParserContext context) {
-      parserContexts.put(parserClass, context);
+   public void fireParserListeners() {
+      for (ConfigurationParserListener listener : listeners) {
+         listener.parsingComplete(this);
+      }
    }
 
    public ClassLoader getClassLoader() {
       return classLoader.get();
    }
 
-   Map<Class<? extends ConfigurationParser>, ParserContext> getParserContexts() {
-      return parserContexts;
-   }
-
    public void validate() {
       globalConfigurationBuilder.defaultCacheName().ifPresent(name -> {
          if (!namedConfigurationBuilders.containsKey(name))

diff --git a/core/src/main/java/org/infinispan/configuration/parsing/ConfigurationParser.java b/core/src/main/java/org/infinispan/configuration/parsing/ConfigurationParser.java
@@ -22,7 +22,7 @@ public interface ConfigurationParser {
 
    Namespace[] getNamespaces();
 
-   default void readAttribute(ConfigurationReader reader, String name, String attributeName, String attributeValue, ConfigurationBuilderHolder holder) {
+   default void readAttribute(ConfigurationReader reader, String elementName, int attributeIndex, ConfigurationBuilderHolder holder) {
       throw new UnsupportedOperationException("This parser cannot handle namespaced attributes");
    }
 }
diff --git a/...ration/parsing/ParserContextListener.java → .../parsing/ConfigurationParserListener.java b/...ration/parsing/ParserContextListener.java → .../parsing/ConfigurationParserListener.java
@@ -4,11 +4,11 @@
 
 /**
  * ParserContextListener. An interface which should be implemented by listeners who wish to be
- * notified when a file has been successfully parsed. See {@link ParserContext}
+ * notified when a file has been successfully parsed.
  *
  * @author Tristan Tarrant
  * @since 5.2
  */
-public interface ParserContextListener extends EventListener {
-   void parsingComplete(ParserContext context);
+public interface ConfigurationParserListener extends EventListener {
+   void parsingComplete(ConfigurationBuilderHolder holder);
 }
diff --git a/core/src/main/java/org/infinispan/configuration/parsing/ParseUtils.java b/core/src/main/java/org/infinispan/configuration/parsing/ParseUtils.java
@@ -8,7 +8,6 @@
 import java.util.Iterator;
 import java.util.Set;
 
-
 import org.infinispan.commons.CacheConfigurationException;
 import org.infinispan.commons.configuration.io.ConfigurationReader;
 import org.infinispan.commons.configuration.io.ConfigurationReaderException;
@@ -38,7 +37,11 @@ public static ConfigurationReaderException unexpectedElement(final Configuration
     }
 
     public static <T extends Enum<T>> ConfigurationReaderException unexpectedElement(final ConfigurationReader reader, T element) {
-        return new ConfigurationReaderException("Unexpected element '" + element.toString() + "' encountered", reader.getLocation());
+        return unexpectedElement(reader, element.toString());
+    }
+
+    public static ConfigurationReaderException unexpectedElement(final ConfigurationReader reader, String element) {
+        return new ConfigurationReaderException("Unexpected element '" + element + "' encountered", reader.getLocation());
     }
 
     /**

diff --git a/core/src/main/java/org/infinispan/configuration/parsing/Parser.java b/core/src/main/java/org/infinispan/configuration/parsing/Parser.java
@@ -558,15 +558,34 @@ private void parseJGroupsStack(ConfigurationReader reader, ConfigurationBuilderH
    }
 
    private void parseJGroupsRelay(ConfigurationReader reader, ConfigurationBuilderHolder holder, EmbeddedJGroupsChannelConfigurator stackConfigurator) {
-      String defaultStack = ParseUtils.requireSingleAttribute(reader, Attribute.DEFAULT_STACK);
+      String defaultStack = ParseUtils.requireAttributes(reader, Attribute.DEFAULT_STACK)[0];
+      String defaultCluster = "xsite";
       if (holder.getJGroupsStack(defaultStack) == null) {
          throw CONFIG.missingJGroupsStack(defaultStack);
       }
+      for (int i = 0; i < reader.getAttributeCount(); i++) {
+         if (!ParseUtils.isNoNamespaceAttribute(reader, i))
+            continue;
+         String value = reader.getAttributeValue(i);
+         Attribute attribute = Attribute.forName(reader.getAttributeName(i));
+         switch (attribute) {
+            case DEFAULT_STACK:
+               // Already seen
+               break;
+            case CLUSTER:
+               defaultCluster = value;
+               break;
+            default: {
+               throw ParseUtils.unexpectedAttribute(reader, i);
+            }
+         }
+      }
       while (reader.inTag()) {
          Element element = Element.forName(reader.getLocalName());
          switch (element) {
             case REMOTE_SITE:
                String remoteSite = ParseUtils.requireAttributes(reader, Attribute.NAME)[0];
+               String cluster = defaultCluster;
                String stack = defaultStack;
                for (int i = 0; i < reader.getAttributeCount(); i++) {
                   Attribute attribute = Attribute.forName(reader.getAttributeName(i));
@@ -579,12 +598,15 @@ private void parseJGroupsRelay(ConfigurationReader reader, ConfigurationBuilderH
                            throw CONFIG.missingJGroupsStack(stack);
                         }
                         break;
+                     case CLUSTER:
+                        cluster = reader.getAttributeValue(i);
+                        break;
                      default:
                         throw ParseUtils.unexpectedAttribute(reader, i);
                   }
                }
                ParseUtils.requireNoContent(reader);
-               stackConfigurator.addRemoteSite(remoteSite, holder.getJGroupsStack(stack));
+               stackConfigurator.addRemoteSite(remoteSite, cluster, holder.getJGroupsStack(stack));
                break;
             default:
                throw ParseUtils.unexpectedElement(reader);

diff --git a/core/src/main/java/org/infinispan/configuration/parsing/ParserContext.java b/core/src/main/java/org/infinispan/configuration/parsing/ParserContext.java
diff --git a/core/src/main/java/org/infinispan/configuration/parsing/ParserRegistry.java b/core/src/main/java/org/infinispan/configuration/parsing/ParserRegistry.java
@@ -169,9 +169,7 @@ public ConfigurationBuilderHolder parse(InputStream is, ConfigurationBuilderHold
       ConfigurationReader reader = ConfigurationReader.from(is).withResolver(resourceResolver).withType(mediaType).withProperties(properties).withNamingStrategy(NamingStrategy.KEBAB_CASE).build();
       parse(reader, holder);
       // Fire all parsingComplete events if any
-      for (ParserContext parserContext : holder.getParserContexts().values()) {
-         parserContext.fireParsingComplete();
-      }
+      holder.fireParserListeners();
       return holder;
    }
 
@@ -214,7 +212,7 @@ public void parseAttribute(ConfigurationReader reader, int i, ConfigurationBuild
       NamespaceParserPair parser = findNamespaceParser(namespace, name);
       ConfigurationSchemaVersion oldSchema = reader.getSchema();
       reader.setSchema(Schema.fromNamespaceURI(namespace));
-      parser.parser.readAttribute(reader, name, reader.getAttributeName(i), reader.getAttributeValue(i), holder);
+      parser.parser.readAttribute(reader, name, i, holder);
       reader.setSchema(oldSchema);
    }
 

diff --git a/...in/java/org/infinispan/remoting/transport/jgroups/AbstractJGroupsChannelConfigurator.java b/...in/java/org/infinispan/remoting/transport/jgroups/AbstractJGroupsChannelConfigurator.java
@@ -0,0 +1,31 @@
+package org.infinispan.remoting.transport.jgroups;
+
+
+import org.jgroups.JChannel;
+import org.jgroups.stack.Protocol;
+import org.jgroups.util.SocketFactory;
+
+/**
+ * @author Tristan Tarrant &lt;tristan@infinispan.org&gt;
+ * @since 13.0
+ **/
+public abstract class AbstractJGroupsChannelConfigurator implements JGroupsChannelConfigurator {
+   private SocketFactory socketFactory;
+
+   @Override
+   public void setSocketFactory(SocketFactory socketFactory) {
+      this.socketFactory = socketFactory;
+   }
+
+   public SocketFactory getSocketFactory() {
+      return socketFactory;
+   }
+
+   protected JChannel applySocketFactory(JChannel channel) {
+      if (socketFactory != null) {
+         Protocol protocol = channel.getProtocolStack().getTopProtocol();
+         protocol.setSocketFactory(socketFactory);
+      }
+      return channel;
+   }
+}
diff --git a/...in/java/org/infinispan/remoting/transport/jgroups/EmbeddedJGroupsChannelConfigurator.java b/...in/java/org/infinispan/remoting/transport/jgroups/EmbeddedJGroupsChannelConfigurator.java
@@ -15,6 +15,7 @@
 import org.jgroups.protocols.relay.config.RelayConfig;
 import org.jgroups.stack.Configurator;
 import org.jgroups.stack.Protocol;
+import org.jgroups.util.SocketFactory;
 import org.jgroups.util.StackType;
 
 /**
@@ -23,13 +24,13 @@
  * @author Tristan Tarrant &lt;tristan@infinispan.org&gt;
  * @since 10.0
  **/
-public class EmbeddedJGroupsChannelConfigurator implements JGroupsChannelConfigurator {
+public class EmbeddedJGroupsChannelConfigurator extends AbstractJGroupsChannelConfigurator {
 
    private static final String PROTOCOL_PREFIX = "org.jgroups.protocols.";
 
    private final String name;
    final List<ProtocolConfiguration> stack;
-   final Map<String, JGroupsChannelConfigurator> remoteSites;
+   final Map<String, RemoteSite> remoteSites;
 
    public EmbeddedJGroupsChannelConfigurator(String name, List<ProtocolConfiguration> stack) {
       this.name = name;
@@ -51,16 +52,12 @@ public List<ProtocolConfiguration> getProtocolStack() {
       return stack;
    }
 
-   public Map<String, JGroupsChannelConfigurator> getRemoteSites() {
-      return remoteSites;
-   }
-
    public String getName() {
       return name;
    }
 
    @Override
-   public JChannel createChannel() throws Exception {
+   public JChannel createChannel(String name) throws Exception {
       StackType stackType = org.jgroups.util.Util.getIpStackType();
       List<Protocol> protocols = new ArrayList<>(stack.size());
       for(ProtocolConfiguration c : stack) {
@@ -82,14 +79,21 @@ public JChannel createChannel() throws Exception {
                if (remoteSites.size() == 0) {
                   throw CONFIG.jgroupsRelayWithoutRemoteSites(name);
                }
-               for (Map.Entry<String, JGroupsChannelConfigurator> remoteSite : remoteSites.entrySet()) {
-                  JGroupsChannelConfigurator remoteSiteChannel = remoteSite.getValue();
+               for (Map.Entry<String, RemoteSite> remoteSite : remoteSites.entrySet()) {
+                  JGroupsChannelConfigurator configurator = remoteSite.getValue().configurator;
+                  SocketFactory socketFactory = getSocketFactory();
+                  final String remoteCluster = remoteSite.getValue().cluster;
+                  if (socketFactory instanceof NamedSocketFactory) {
+                     // Create a new NamedSocketFactory using the remote cluster name
+                     socketFactory = new NamedSocketFactory((NamedSocketFactory) socketFactory, remoteCluster);
+                  }
+                  configurator.setSocketFactory(socketFactory);
                   RelayConfig.SiteConfig siteConfig = new RelayConfig.SiteConfig(remoteSite.getKey());
-                  siteConfig.addBridge(new RelayConfig.BridgeConfig(remoteSiteChannel.getName()) {
+                  siteConfig.addBridge(new RelayConfig.BridgeConfig(remoteCluster) {
                      @Override
                      public JChannel createChannel() throws Exception {
                         // TODO The bridge channel is created lazily, and Infinispan doesn't see any errors
-                        return remoteSiteChannel.createChannel();
+                        return configurator.createChannel(getClusterName());
                      }
                   });
                   relay2.addSite(remoteSite.getKey(), siteConfig);
@@ -103,14 +107,14 @@ public JChannel createChannel() throws Exception {
          }
       }
 
-      return new JChannel(protocols);
+      return applySocketFactory(new JChannel(protocols));
    }
 
-   public void addRemoteSite(String remoteSite, JGroupsChannelConfigurator stackConfigurator) {
+   public void addRemoteSite(String remoteSite, String cluster, JGroupsChannelConfigurator stackConfigurator) {
       if (remoteSites.containsKey(remoteSite)) {
          throw CONFIG.duplicateRemoteSite(remoteSite, name);
       } else {
-         remoteSites.put(remoteSite, stackConfigurator);
+         remoteSites.put(remoteSite, new RemoteSite(cluster, stackConfigurator));
       }
    }
 
@@ -213,4 +217,14 @@ public enum StackCombine {
       REMOVE,
       APPEND, // non-public
    }
+
+   public static class RemoteSite {
+      final String cluster;
+      final JGroupsChannelConfigurator configurator;
+
+      public RemoteSite(String cluster, JGroupsChannelConfigurator configurator) {
+         this.cluster = cluster;
+         this.configurator = configurator;
+      }
+   }
 }
diff --git a/...c/main/java/org/infinispan/remoting/transport/jgroups/FileJGroupsChannelConfigurator.java b/...c/main/java/org/infinispan/remoting/transport/jgroups/FileJGroupsChannelConfigurator.java
@@ -16,7 +16,7 @@
  * @author Tristan Tarrant &lt;tristan@infinispan.org&gt;
  * @since 10.0
  **/
-public class FileJGroupsChannelConfigurator implements JGroupsChannelConfigurator {
+public class FileJGroupsChannelConfigurator extends AbstractJGroupsChannelConfigurator {
    private final String name;
    private final String path;
    private final Properties properties;
@@ -45,8 +45,8 @@ public String getName() {
    }
 
    @Override
-   public JChannel createChannel() throws Exception {
-      return new JChannel(this);
+   public JChannel createChannel(String name) throws Exception {
+      return applySocketFactory(new JChannel(this));
    }
 
    public String getPath() {