WFLY-11913 JGroups client sockets are not registered with the socket …

…binding manager
wildfly · May 6, 2019 · 8ff8e60 · 8ff8e60
1 parent 1fbc174
commit 8ff8e60
Show file tree

Hide file tree

Showing 29 changed files with 1,001 additions and 146 deletions.
diff --git a/...oups/extension/src/main/java/org/jboss/as/clustering/jgroups/subsystem/JGroupsSchema.java b/...oups/extension/src/main/java/org/jboss/as/clustering/jgroups/subsystem/JGroupsSchema.java
@@ -37,9 +37,10 @@ public enum JGroupsSchema implements Schema<JGroupsSchema> {
     VERSION_3_0(3, 0), // WildFly 9
     VERSION_4_0(4, 0), // WildFly 10
     VERSION_5_0(5, 0), // WildFly 11
-    VERSION_6_0(6, 0), // WildFly 12
+    VERSION_6_0(6, 0), // WildFly 12-16
+    VERSION_7_0(7, 0), // WildFly 17
     ;
-    public static final JGroupsSchema CURRENT = VERSION_6_0;
+    public static final JGroupsSchema CURRENT = VERSION_7_0;
 
     private final int major;
     private final int minor;

diff --git a/...on/src/main/java/org/jboss/as/clustering/jgroups/subsystem/JGroupsSubsystemXMLReader.java b/...on/src/main/java/org/jboss/as/clustering/jgroups/subsystem/JGroupsSubsystemXMLReader.java
@@ -405,6 +405,12 @@ private void parseTransport(XMLExtendedStreamReader reader, PathAddress stackAdd
                         break;
                     }
                 }
+                case CLIENT_SOCKET_BINDING: {
+                    if (this.schema.since(JGroupsSchema.VERSION_7_0)) {
+                        readAttribute(reader, i, operation, SocketTransportResourceDefinition.Attribute.CLIENT_SOCKET_BINDING);
+                        break;
+                    }
+                }
                 default: {
                     this.parseProtocolAttribute(reader, i, operation);
                 }
@@ -470,10 +476,25 @@ private void parseSocketProtocol(XMLExtendedStreamReader reader, PathAddress sta
         operations.put(address, operation);
 
         for (int i = 0; i < reader.getAttributeCount(); i++) {
-            this.parseSocketProtocolAttribute(reader, i, operation);
+            XMLAttribute attribute = XMLAttribute.forName(reader.getAttributeLocalName(i));
+            switch (attribute) {
+                case SOCKET_BINDING: {
+                    readAttribute(reader, i, operation, SocketProtocolResourceDefinition.Attribute.SOCKET_BINDING);
+                    break;
+                }
+                case CLIENT_SOCKET_BINDING: {
+                    if (this.schema.since(JGroupsSchema.VERSION_7_0)) {
+                        readAttribute(reader, i, operation, SocketProtocolResourceDefinition.Attribute.CLIENT_SOCKET_BINDING);
+                        break;
+                    }
+                }
+                default: {
+                    parseProtocolAttribute(reader, i, operation);
+                }
+            }
         }
 
-        require(reader, operation, SocketBindingProtocolResourceDefinition.Attribute.SOCKET_BINDING);
+        require(reader, operation, SocketProtocolResourceDefinition.Attribute.SOCKET_BINDING);
 
         while (reader.hasNext() && (reader.nextTag() != XMLStreamConstants.END_ELEMENT)) {
             this.parseProtocolElement(reader, address, operations);
@@ -488,7 +509,16 @@ private void parseSocketDiscoveryProtocol(XMLExtendedStreamReader reader, PathAd
         operations.put(address, operation);
 
         for (int i = 0; i < reader.getAttributeCount(); i++) {
-            this.parseSocketDiscoveryProtocolAttribute(reader, i, operation);
+            XMLAttribute attribute = XMLAttribute.forName(reader.getAttributeLocalName(i));
+            switch (attribute) {
+                case OUTBOUND_SOCKET_BINDINGS: {
+                    readAttribute(reader, i, operation, SocketDiscoveryProtocolResourceDefinition.Attribute.OUTBOUND_SOCKET_BINDINGS);
+                    break;
+                }
+                default: {
+                    parseProtocolAttribute(reader, i, operation);
+                }
+            }
         }
 
         require(reader, operation, SocketDiscoveryProtocolResourceDefinition.Attribute.OUTBOUND_SOCKET_BINDINGS);
@@ -506,7 +536,16 @@ private void parseJDBCProtocol(XMLExtendedStreamReader reader, PathAddress stack
         operations.put(address, operation);
 
         for (int i = 0; i < reader.getAttributeCount(); i++) {
-            this.parseJDBCProtocolAttribute(reader, i, operation);
+            XMLAttribute attribute = XMLAttribute.forName(reader.getAttributeLocalName(i));
+            switch (attribute) {
+                case DATA_SOURCE: {
+                    readAttribute(reader, i, operation, JDBCProtocolResourceDefinition.Attribute.DATA_SOURCE);
+                    break;
+                }
+                default: {
+                    parseProtocolAttribute(reader, i, operation);
+                }
+            }
         }
 
         require(reader, operation, JDBCProtocolResourceDefinition.Attribute.DATA_SOURCE);
@@ -524,7 +563,20 @@ private void parseEncryptProtocol(XMLExtendedStreamReader reader, PathAddress st
         operations.put(address, operation);
 
         for (int i = 0; i < reader.getAttributeCount(); i++) {
-            this.parseEncryptProtocolAttribute(reader, i, operation);
+            XMLAttribute attribute = XMLAttribute.forName(reader.getAttributeLocalName(i));
+            switch (attribute) {
+                case KEY_ALIAS: {
+                    readAttribute(reader, i, operation, EncryptProtocolResourceDefinition.Attribute.KEY_ALIAS);
+                    break;
+                }
+                case KEY_STORE: {
+                    readAttribute(reader, i, operation, EncryptProtocolResourceDefinition.Attribute.KEY_STORE);
+                    break;
+                }
+                default: {
+                    parseProtocolAttribute(reader, i, operation);
+                }
+            }
         }
 
         require(reader, operation, EncryptProtocolResourceDefinition.Attribute.KEY_ALIAS, EncryptProtocolResourceDefinition.Attribute.KEY_STORE);
@@ -570,62 +622,6 @@ private void parseProtocol(XMLExtendedStreamReader reader, PathAddress stackAddr
         }
     }
 
-    private void parseSocketProtocolAttribute(XMLExtendedStreamReader reader, int index, ModelNode operation) throws XMLStreamException {
-        XMLAttribute attribute = XMLAttribute.forName(reader.getAttributeLocalName(index));
-        switch (attribute) {
-            case SOCKET_BINDING: {
-                readAttribute(reader, index, operation, SocketBindingProtocolResourceDefinition.Attribute.SOCKET_BINDING);
-                break;
-            }
-            default: {
-                parseProtocolAttribute(reader, index, operation);
-            }
-        }
-    }
-
-    private void parseSocketDiscoveryProtocolAttribute(XMLExtendedStreamReader reader, int index, ModelNode operation) throws XMLStreamException {
-        XMLAttribute attribute = XMLAttribute.forName(reader.getAttributeLocalName(index));
-        switch (attribute) {
-            case OUTBOUND_SOCKET_BINDINGS: {
-                readAttribute(reader, index, operation, SocketDiscoveryProtocolResourceDefinition.Attribute.OUTBOUND_SOCKET_BINDINGS);
-                break;
-            }
-            default: {
-                parseProtocolAttribute(reader, index, operation);
-            }
-        }
-    }
-
-    private void parseJDBCProtocolAttribute(XMLExtendedStreamReader reader, int index, ModelNode operation) throws XMLStreamException {
-        XMLAttribute attribute = XMLAttribute.forName(reader.getAttributeLocalName(index));
-        switch (attribute) {
-            case DATA_SOURCE: {
-                readAttribute(reader, index, operation, JDBCProtocolResourceDefinition.Attribute.DATA_SOURCE);
-                break;
-            }
-            default: {
-                parseProtocolAttribute(reader, index, operation);
-            }
-        }
-    }
-
-    private void parseEncryptProtocolAttribute(XMLExtendedStreamReader reader, int index, ModelNode operation) throws XMLStreamException {
-        XMLAttribute attribute = XMLAttribute.forName(reader.getAttributeLocalName(index));
-        switch (attribute) {
-            case KEY_ALIAS: {
-                readAttribute(reader, index, operation, EncryptProtocolResourceDefinition.Attribute.KEY_ALIAS);
-                break;
-            }
-            case KEY_STORE: {
-                readAttribute(reader, index, operation, EncryptProtocolResourceDefinition.Attribute.KEY_STORE);
-                break;
-            }
-            default: {
-                parseProtocolAttribute(reader, index, operation);
-            }
-        }
-    }
-
     private void parseProtocolAttribute(XMLExtendedStreamReader reader, int index, ModelNode operation) throws XMLStreamException {
         XMLAttribute attribute = XMLAttribute.forName(reader.getAttributeLocalName(index));
         switch (attribute) {

diff --git a/...on/src/main/java/org/jboss/as/clustering/jgroups/subsystem/JGroupsSubsystemXMLWriter.java b/...on/src/main/java/org/jboss/as/clustering/jgroups/subsystem/JGroupsSubsystemXMLWriter.java
@@ -134,9 +134,9 @@ private static void writeProtocolAttributes(XMLExtendedStreamWriter writer, Prop
 
         String protocol = property.getName();
         if (containsName(ProtocolRegistration.MulticastProtocol.class, protocol)) {
-            writeAttributes(writer, property.getValue(), SocketBindingProtocolResourceDefinition.Attribute.class);
+            writeAttributes(writer, property.getValue(), MulticastProtocolResourceDefinition.Attribute.class);
         } else if (containsName(ProtocolRegistration.SocketProtocol.class, protocol)) {
-            writeAttributes(writer, property.getValue(), OptionalSocketBindingProtocolResourceDefinition.Attribute.class);
+            writeAttributes(writer, property.getValue(), SocketProtocolResourceDefinition.Attribute.class);
         } else if (containsName(ProtocolRegistration.JdbcProtocol.class, protocol)) {
             writeAttributes(writer, property.getValue(), JDBCProtocolResourceDefinition.Attribute.class);
         } else if (containsName(ProtocolRegistration.EncryptProtocol.class, protocol)) {

diff --git a/...ketBindingProtocolResourceDefinition.java → .../MulticastProtocolResourceDefinition.java b/...ketBindingProtocolResourceDefinition.java → .../MulticastProtocolResourceDefinition.java
@@ -41,7 +41,7 @@
  * Resource definition override for protocols that require a socket-binding.
  * @author Paul Ferraro
  */
-public class SocketBindingProtocolResourceDefinition extends ProtocolResourceDefinition {
+public class MulticastProtocolResourceDefinition extends ProtocolResourceDefinition {
 
     enum Attribute implements org.jboss.as.clustering.controller.Attribute, UnaryOperator<SimpleAttributeDefinitionBuilder> {
         SOCKET_BINDING(ModelDescriptionConstants.SOCKET_BINDING, ModelType.STRING) {
@@ -86,7 +86,7 @@ public ResourceDescriptor apply(ResourceDescriptor descriptor) {
         }
     }
 
-    SocketBindingProtocolResourceDefinition(String name, UnaryOperator<ResourceDescriptor> configurator, ResourceServiceConfiguratorFactory serviceConfiguratorFactory, ResourceServiceConfiguratorFactory parentServiceConfiguratorFactory) {
+    MulticastProtocolResourceDefinition(String name, UnaryOperator<ResourceDescriptor> configurator, ResourceServiceConfiguratorFactory parentServiceConfiguratorFactory) {
-        super(pathElement(name), new ResourceDescriptorConfigurator(configurator), serviceConfiguratorFactory, parentServiceConfiguratorFactory);
+        super(pathElement(name), new ResourceDescriptorConfigurator(configurator), MulticastSocketProtocolConfigurationServiceConfigurator::new, parentServiceConfiguratorFactory);
     }
 }
diff --git a/...clustering/jgroups/subsystem/MulticastSocketProtocolConfigurationServiceConfigurator.java b/...clustering/jgroups/subsystem/MulticastSocketProtocolConfigurationServiceConfigurator.java
@@ -22,7 +22,7 @@
 
 package org.jboss.as.clustering.jgroups.subsystem;
 
-import static org.jboss.as.clustering.jgroups.subsystem.SocketBindingProtocolResourceDefinition.Attribute.SOCKET_BINDING;
+import static org.jboss.as.clustering.jgroups.subsystem.MulticastProtocolResourceDefinition.Attribute.SOCKET_BINDING;
 
 import java.util.Collections;
 import java.util.Map;

diff --git a/...tension/src/main/java/org/jboss/as/clustering/jgroups/subsystem/ProtocolRegistration.java b/...tension/src/main/java/org/jboss/as/clustering/jgroups/subsystem/ProtocolRegistration.java
@@ -113,16 +113,12 @@ static void buildTransformation(ModelVersion version, ResourceTransformationDesc
 
         for (SocketProtocol protocol : EnumSet.allOf(SocketProtocol.class)) {
             PathElement path = ProtocolResourceDefinition.pathElement(protocol.name());
-            if (!JGroupsModel.VERSION_7_0_0.requiresTransformation(version)) {
+            SocketProtocolResourceDefinition.addTransformations(version, parent.addChildResource(path));
-                OptionalSocketBindingProtocolResourceDefinition.addTransformations(version, parent.addChildResource(path));
-            }
         }
 
         for (MulticastProtocol protocol : EnumSet.allOf(MulticastProtocol.class)) {
             PathElement path = ProtocolResourceDefinition.pathElement(protocol.name());
-            if (!JGroupsModel.VERSION_5_0_0.requiresTransformation(version)) {
+            MulticastProtocolResourceDefinition.addTransformations(version, parent.addChildResource(path));
-                SocketBindingProtocolResourceDefinition.addTransformations(version, parent.addChildResource(path));
-            }
         }
 
         for (JdbcProtocol protocol : EnumSet.allOf(JdbcProtocol.class)) {
@@ -194,10 +190,10 @@ public void register(ManagementResourceRegistration registration) {
 
         // Override definitions for protocol types
         for (SocketProtocol protocol : EnumSet.allOf(SocketProtocol.class)) {
-            new OptionalSocketBindingProtocolResourceDefinition(protocol.name(), this.configurator, SocketProtocolConfigurationServiceConfigurator::new, this.parentServiceConfiguratorFactory).register(registration);
+            new SocketProtocolResourceDefinition(protocol.name(), this.configurator, this.parentServiceConfiguratorFactory).register(registration);
         }
         for (MulticastProtocol protocol : EnumSet.allOf(MulticastProtocol.class)) {
-            new SocketBindingProtocolResourceDefinition(protocol.name(), this.configurator, MulticastSocketProtocolConfigurationServiceConfigurator::new, this.parentServiceConfiguratorFactory).register(registration);
+            new MulticastProtocolResourceDefinition(protocol.name(), this.configurator, this.parentServiceConfiguratorFactory).register(registration);
         }
 
         for (JdbcProtocol protocol : EnumSet.allOf(JdbcProtocol.class)) {

diff --git a/...jboss/as/clustering/jgroups/subsystem/SocketProtocolConfigurationServiceConfigurator.java b/...jboss/as/clustering/jgroups/subsystem/SocketProtocolConfigurationServiceConfigurator.java
@@ -22,14 +22,17 @@
 
 package org.jboss.as.clustering.jgroups.subsystem;
 
-import static org.jboss.as.clustering.jgroups.subsystem.OptionalSocketBindingProtocolResourceDefinition.Attribute.SOCKET_BINDING;
+import static org.jboss.as.clustering.jgroups.subsystem.SocketProtocolResourceDefinition.Attribute.CLIENT_SOCKET_BINDING;
+import static org.jboss.as.clustering.jgroups.subsystem.SocketProtocolResourceDefinition.Attribute.SOCKET_BINDING;
 
 import java.net.InetAddress;
+import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
-import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import org.jboss.as.clustering.controller.Attribute;
 import org.jboss.as.clustering.controller.CommonUnaryRequirement;
 import org.jboss.as.controller.OperationContext;
 import org.jboss.as.controller.OperationFailedException;
@@ -39,6 +42,7 @@
 import org.jboss.dmr.ModelNode;
 import org.jboss.msc.service.ServiceBuilder;
 import org.jgroups.protocols.FD_SOCK;
+import org.wildfly.clustering.service.CompositeDependency;
 import org.wildfly.clustering.service.ServiceConfigurator;
 import org.wildfly.clustering.service.ServiceSupplierDependency;
 import org.wildfly.clustering.service.SimpleSupplierDependency;
@@ -51,26 +55,35 @@
 public class SocketProtocolConfigurationServiceConfigurator extends ProtocolConfigurationServiceConfigurator<FD_SOCK> {
 
     private volatile SupplierDependency<SocketBinding> binding;
+    private volatile SupplierDependency<SocketBinding> clientBinding;
 
     public SocketProtocolConfigurationServiceConfigurator(PathAddress address) {
         super(address);
     }
 
     @Override
     public <T> ServiceBuilder<T> register(ServiceBuilder<T> builder) {
-        return super.register(this.binding.register(builder));
+        return super.register(new CompositeDependency(this.binding, this.clientBinding).register(builder));
     }
 
     @Override
     public ServiceConfigurator configure(OperationContext context, ModelNode model) throws OperationFailedException {
-        String bindingName = SOCKET_BINDING.resolveModelAttribute(context, model).asString(null);
+        this.binding = createDependency(context, model, SOCKET_BINDING);
-        this.binding = (bindingName != null) ? new ServiceSupplierDependency<>(CommonUnaryRequirement.SOCKET_BINDING.getServiceName(context, bindingName)) : new SimpleSupplierDependency<>(null);
+        this.clientBinding = createDependency(context, model, CLIENT_SOCKET_BINDING);
         return super.configure(context, model);
     }
 
+    private static SupplierDependency<SocketBinding> createDependency(OperationContext context, ModelNode model, Attribute attribute) throws OperationFailedException {
+        String bindingName = attribute.resolveModelAttribute(context, model).asStringOrNull();
+        return (bindingName != null) ? new ServiceSupplierDependency<>(CommonUnaryRequirement.SOCKET_BINDING.getServiceName(context, bindingName)) : new SimpleSupplierDependency<>(null);
+    }
+
     @Override
     public Map<String, SocketBinding> getSocketBindings() {
-        return Collections.singletonMap("jgroups.fd_sock.srv_sock", this.binding.get());
+        Map<String, SocketBinding> bindings = new HashMap<>();
+        bindings.put("jgroups.fd_sock.srv_sock", this.binding.get());
+        bindings.put("jgroups.fd.ping_sock", this.clientBinding.get());
+        return bindings;
     }
 
     @Override
@@ -93,5 +106,10 @@ public void accept(FD_SOCK protocol) {
                 }
             }
         }
+        SocketBinding clientBinding = this.clientBinding.get();
+        if (clientBinding != null) {
+            InetSocketAddress socketAddress = clientBinding.getSocketAddress();
+            protocol.setValue("client_bind_port", socketAddress.getPort());
+        }
     }
 }