WFLY-8417 Create DistributedWorkManager implementation based on WF cl…

…ustering API

clustering/server/src/main/java/org/wildfly/clustering/server/group/AddressableNode.java

@@ -21,6 +21,7 @@
  */
 package org.wildfly.clustering.server.group;
 
+import java.io.Serializable;
 import java.net.InetSocketAddress;
 
 import org.jgroups.Address;
@@ -32,7 +33,8 @@
  * and transport socket binding.
  * @author Paul Ferraro
  */
-public class AddressableNode implements Node, Addressable, Comparable<AddressableNode> {
+public class AddressableNode implements Node, Addressable, Comparable<AddressableNode>, Serializable {
+    private static final long serialVersionUID = -7707210981640344598L;
 
     private final Address address;
     private final String name;

connector/pom.xml

@@ -33,13 +33,21 @@
     <artifactId>wildfly-connector</artifactId>
     <name>WildFly: Connector Subsystem</name>
     <dependencies>
+        <dependency>
+            <groupId>org.jgroups</groupId>
+            <artifactId>jgroups</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.wildfly.core</groupId>
             <artifactId>wildfly-server</artifactId>
         </dependency>
         <dependency>
             <groupId>org.wildfly</groupId>
-            <artifactId>wildfly-clustering-jgroups-spi</artifactId>
+            <artifactId>wildfly-clustering-marshalling-jboss</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.wildfly</groupId>
+            <artifactId>wildfly-clustering-spi</artifactId>
         </dependency>
         <dependency>
             <groupId>org.wildfly</groupId>
@@ -122,6 +130,11 @@
             <scope>provided</scope>
             <optional>true</optional>
         </dependency>
+        <dependency>
+            <groupId>org.kohsuke.metainf-services</groupId>
+            <artifactId>metainf-services</artifactId>
+            <scope>provided</scope>
+        </dependency>
         <dependency>
             <groupId>javax.validation</groupId>
             <artifactId>validation-api</artifactId>

connector/src/main/java/org/jboss/as/connector/services/workmanager/DistributedWorkManagerService.java

@@ -27,22 +27,21 @@
 import java.util.concurrent.Executor;
 
 import org.jboss.as.connector.security.ElytronSecurityIntegration;
-import org.jboss.as.connector.services.workmanager.transport.ForkChannelTransport;
+import org.jboss.as.connector.services.workmanager.transport.CommandDispatcherTransport;
 import org.jboss.as.connector.util.ConnectorServices;
 import org.jboss.as.txn.integration.JBossContextXATerminator;
 import org.jboss.jca.core.security.picketbox.PicketBoxSecurityIntegration;
 import org.jboss.jca.core.spi.workmanager.Address;
 import org.jboss.jca.core.tx.jbossts.XATerminatorImpl;
 import org.jboss.jca.core.workmanager.WorkManagerCoordinator;
-import org.jboss.jca.core.workmanager.transport.remote.jgroups.JGroupsTransport;
 import org.jboss.msc.inject.Injector;
 import org.jboss.msc.service.Service;
 import org.jboss.msc.service.StartContext;
 import org.jboss.msc.service.StartException;
 import org.jboss.msc.service.StopContext;
 import org.jboss.msc.value.InjectedValue;
 import org.jboss.threads.BlockingExecutor;
-import org.wildfly.clustering.jgroups.spi.ChannelFactory;
+import org.wildfly.clustering.dispatcher.CommandDispatcherFactory;
 
 /**
  * A WorkManager Service.
@@ -60,7 +59,7 @@ public final class DistributedWorkManagerService implements Service<NamedDistrib
 
     private final InjectedValue<JBossContextXATerminator> xaTerminator = new InjectedValue<JBossContextXATerminator>();
 
-    private final InjectedValue<ChannelFactory> jGroupsChannelFactory = new InjectedValue<ChannelFactory>();
+    private final InjectedValue<CommandDispatcherFactory> dispatcherFactory = new InjectedValue<>();
 
     /**
      * create an instance
@@ -82,16 +81,9 @@ public NamedDistributedWorkManager getValue() throws IllegalStateException {
     public void start(StartContext context) throws StartException {
         ROOT_LOGGER.debugf("Starting JCA DistributedWorkManager: ", value.getName());
 
-        ChannelFactory factory = this.jGroupsChannelFactory.getValue();
-        JGroupsTransport transport = new ForkChannelTransport(factory);
-        try {
-            transport.setChannel(factory.createChannel(this.value.getName()));
-            transport.setClusterName(this.value.getName());
-            this.value.setTransport(transport);
-        } catch (Exception e) {
-            ROOT_LOGGER.trace("failed to start JGroups channel", e);
-            throw ROOT_LOGGER.failedToStartJGroupsChannel(this.value.getName(), this.value.getName());
-        }
+        CommandDispatcherTransport transport = new CommandDispatcherTransport(this.dispatcherFactory.getValue(), this.value.getName());
+
+        this.value.setTransport(transport);
 
         BlockingExecutor longRunning = (BlockingExecutor) executorLong.getOptionalValue();
         if (longRunning != null) {
@@ -111,15 +103,14 @@ public void start(StartContext context) throws StartException {
         } else {
             this.value.setSecurityIntegration(new PicketBoxSecurityIntegration());
         }
-        transport.setId(String.valueOf(transport.getChannel().hashCode()));
 
         try {
             transport.startup();
         } catch (Throwable throwable) {
             ROOT_LOGGER.trace("failed to start DWM transport:", throwable);
             throw ROOT_LOGGER.failedToStartDWMTransport(this.value.getName());
         }
-        transport.register(new Address(value.getId(), value.getName(), value.getTransport().getId()));
+        transport.register(new Address(value.getId(), value.getName(), transport.getId()));
 
         WorkManagerCoordinator.getInstance().registerWorkManager(value);
 
@@ -159,8 +150,7 @@ public Injector<JBossContextXATerminator> getXaTerminatorInjector() {
         return xaTerminator;
     }
 
-    public Injector<ChannelFactory> getJGroupsChannelFactoryInjector() {
-        return jGroupsChannelFactory;
+    public Injector<CommandDispatcherFactory> getCommandDispatcherFactoryInjector() {
+        return this.dispatcherFactory;
     }
-
 }

connector/src/main/java/org/jboss/as/connector/services/workmanager/transport/AddWorkManagerCommand.java

@@ -0,0 +1,52 @@
+/*
+ * JBoss, Home of Professional Open Source.
+ * Copyright 2017, Red Hat, Inc., and individual contributors
+ * as indicated by the @author tags. See the copyright.txt file in the
+ * distribution for a full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+
+package org.jboss.as.connector.services.workmanager.transport;
+
+import org.jboss.jca.core.spi.workmanager.Address;
+import org.wildfly.clustering.dispatcher.Command;
+import org.wildfly.clustering.group.Node;
+
+/**
+ * Equivalent to {@link org.jboss.jca.core.workmanager.transport.remote.jgroups.JGroupsTransport#addWorkManager(java.util.Map, org.jgroups.Address)}.
+ * @author Paul Ferraro
+ */
+public class AddWorkManagerCommand implements Command<Void, CommandDispatcherTransport> {
+    private static final long serialVersionUID = -6747024371979702527L;
+
+    private final Address address;
+    private final Node member;
+
+    public AddWorkManagerCommand(Address address, Node member) {
+        this.address = address;
+        this.member = member;
+    }
+
+    @Override
+    public Void execute(CommandDispatcherTransport transport) {
+        transport.localWorkManagerAdd(this.address, this.member);
+
+        transport.localUpdateShortRunningFree(this.address, transport.getShortRunningFree(this.address));
+        transport.localUpdateLongRunningFree(this.address, transport.getLongRunningFree(this.address));
+        return null;
+    }
+}

connector/src/main/java/org/jboss/as/connector/services/workmanager/transport/ClearDistributedStatisticsCommand.java

@@ -0,0 +1,46 @@
+/*
+ * JBoss, Home of Professional Open Source.
+ * Copyright 2017, Red Hat, Inc., and individual contributors
+ * as indicated by the @author tags. See the copyright.txt file in the
+ * distribution for a full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+
+package org.jboss.as.connector.services.workmanager.transport;
+
+import org.jboss.jca.core.spi.workmanager.Address;
+import org.wildfly.clustering.dispatcher.Command;
+
+/**
+ * Equivalent to {@link org.jboss.jca.core.workmanager.transport.remote.jgroups.JGroupsTransport#clearDistributedStatistics(java.util.Map)}.
+ * @author Paul Ferraro
+ */
+public class ClearDistributedStatisticsCommand implements Command<Void, CommandDispatcherTransport> {
+    private static final long serialVersionUID = -1590205163985739077L;
+
+    private final Address address;
+
+    public ClearDistributedStatisticsCommand(Address address) {
+        this.address = address;
+    }
+
+    @Override
+    public Void execute(CommandDispatcherTransport transport) {
+        transport.localClearDistributedStatistics(this.address);
+        return null;
+    }
+}