jinterface: Support custom distribution protocols without epmd

Make connections to the Erlang Port Mapper Daemon (epmd) optional
in Jinterface when providing a custom transport factory, as is
possible for Erlang nodes with the -no_epmd -proto_dist options.

The existing Jinterface classes and interfaces have been designed
originally for transport protocols using a socket port number and
didn't support other protocols working without epmd.

To provide backward compatibility and avoid code duplication,
create a new OtpGenericTransportFactory abstract class which does
not expect a socket port number but use the peer and local nodes
instead as generic identifiers. This expands the Jinterface
compatibility to other protocols, for instance Unix Domain Sockets.

Adapt AbstractConnection, AbstractNode, OtpNode and OtpSelf to
accept factory subclasses of the new OtpGenericTransportFactory and
make them compatible with a wider set of alternative distribution
protocols. With such transport factory subclasses, epmd is not used
at all whereas the behavior remains unchanged for regular factories.

Add documentation and test case for OtpGenericTransportFactory

6 files changed, 277 insertions, 41 deletions

diff --git a/lib/jinterface/java_src/com/ericsson/otp/erlang/AbstractConnection.java b/lib/jinterface/java_src/com/ericsson/otp/erlang/AbstractConnection.java
index 5fd38ce145..14a09ce65f 100644
--- a/lib/jinterface/java_src/com/ericsson/otp/erlang/AbstractConnection.java
+++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/AbstractConnection.java
@@ -1,7 +1,7 @@
 /*
  * %CopyrightBegin%
  *
- * Copyright Ericsson AB 2000-2021. All Rights Reserved.
+ * Copyright Ericsson AB 2000-2022. All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -166,18 +166,12 @@ public abstract class AbstractConnection extends Thread {
             throws IOException, OtpAuthException {
         peer = other;
         localNode = self;
-        socket = null;
         int port;
 
         traceLevel = defaultLevel;
         setDaemon(true);
 
-        // now get a connection between the two...
-        port = OtpEpmd.lookupPort(peer);
-        if (port == 0)
-            throw new IOException("No remote node found - cannot connect");
-
-        // now find highest common dist value
+        // Find highest common dist value
         if (peer.proto != self.proto || self.distHigh < peer.distLow
                 || self.distLow > peer.distHigh) {
             throw new IOException("No common protocol found - cannot connect");
@@ -187,7 +181,21 @@ public abstract class AbstractConnection extends Thread {
         peer.distChoose = peer.distHigh > self.distHigh ? self.distHigh
                 : peer.distHigh;
 
-        doConnect(port);
+        // Now get a connection between the two nodes
+        if (self.transportFactory instanceof OtpGenericTransportFactory) {
+            // For alternative distribution protocols using a transport factory
+            // extending the OtpGenericTransportFactory class, the notion of
+            // socket port is not used so the remote node is not registered
+            // with Epmd.
+            doGenericConnect();
+
+        } else {
+            // Get the listening port of the remote node registered with Epmd
+            port = OtpEpmd.lookupPort(peer);
+            if (port == 0)
+                throw new IOException("No remote node found - cannot connect");
+            doPortConnect(port);
+        }
 
         name = peer.node();
         connected = true;
@@ -1052,15 +1060,29 @@ public abstract class AbstractConnection extends Thread {
         }
     }
 
-    protected void doConnect(final int port) throws IOException,
+    protected void doPortConnect(final int port) throws IOException,
             OtpAuthException {
         try {
             socket = peer.createTransport(peer.host(), port);
-
             if (traceLevel >= handshakeThreshold) {
                 System.out.println("-> MD5 CONNECT TO " + peer.host() + ":"
                         + port);
             }
+            doConnect();
+
+        } catch (final OtpAuthException ae) {
+            close();
+            throw ae;
+        } catch (final Exception e) {
+            close();
+            final IOException ioe = new IOException(
+                    "Cannot connect to peer node");
+            ioe.initCause(e);
+            throw ioe;
+        }
+    }
+
+    protected void doConnect() throws IOException, OtpAuthException {
             final int send_name_tag = sendName(peer.distChoose, localNode.flags,
                                                localNode.creation());
             recvStatus();
@@ -1073,6 +1095,17 @@ public abstract class AbstractConnection extends Thread {
             recvChallengeAck(our_challenge);
             cookieOk = true;
             sendCookie = false;
+    }
+
+    protected void doGenericConnect() throws IOException,
+            OtpAuthException {
+        try {
+            socket = peer.createTransport(peer);
+            if (traceLevel >= handshakeThreshold) {
+                System.out.println("-> MD5 CONNECT TO " + peer.node());
+            }
+            doConnect();
+
         } catch (final OtpAuthException ae) {
             close();
             throw ae;
@@ -1296,7 +1329,7 @@ public abstract class AbstractConnection extends Thread {
             throw new IOException("Handshake failed - not enough data");
         }
 
-        final int i = hisname.indexOf('@', 0);
+        final int i = hisname.indexOf('@');
         apeer.node = hisname;
         apeer.alive = hisname.substring(0, i);
         apeer.host = hisname.substring(i + 1, hisname.length());
diff --git a/lib/jinterface/java_src/com/ericsson/otp/erlang/AbstractNode.java b/lib/jinterface/java_src/com/ericsson/otp/erlang/AbstractNode.java
index a3fc37174e..7e8355dcc6 100644
--- a/lib/jinterface/java_src/com/ericsson/otp/erlang/AbstractNode.java
+++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/AbstractNode.java
@@ -1,7 +1,7 @@
 /*
  * %CopyrightBegin%
  *
- * Copyright Ericsson AB 2000-2021. All Rights Reserved.
+ * Copyright Ericsson AB 2000-2022. All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -204,7 +204,7 @@ public class AbstractNode implements OtpTransportFactory {
         this.cookie = cookie;
         this.transportFactory = transportFactory;
 
-        final int i = name.indexOf('@', 0);
+        final int i = name.indexOf('@');
         if (i < 0) {
             alive = name;
             host = localHost;
@@ -285,12 +285,12 @@ public class AbstractNode implements OtpTransportFactory {
         return creation;
     }
 
-	void setCreation(int cr) throws OtpErlangDecodeException {
-		if (cr == 0) {
-			throw new OtpErlangDecodeException("Node creation 0 not allowed");
-		}
-		this.creation = cr;
-	}
+    void setCreation(int cr) throws OtpErlangDecodeException {
+        if (cr == 0) {
+            throw new OtpErlangDecodeException("Node creation 0 not allowed");
+        }
+        this.creation = cr;
+    }
 
     /**
      * Set the authorization cookie used by this node.
@@ -332,4 +332,44 @@ public class AbstractNode implements OtpTransportFactory {
             throws IOException {
         return transportFactory.createServerTransport(port);
     }
+
+    /**
+     * Create a client-side transport for alternative distribution protocols
+     * using a transport factory extending the OtpGenericTransportFactory
+     * abstract class. Connect it to the specified server.
+     *
+     * @param peer
+     *            the peer identifying the server to connect to
+     *
+     */
+    public OtpTransport createTransport(final OtpPeer peer)
+            throws IOException {
+        if (transportFactory instanceof OtpGenericTransportFactory) {
+            return ((OtpGenericTransportFactory) transportFactory)
+                .createTransport(peer);
+        }
+        throw new IOException("Method createTransport(OtpPeer) " +
+                              "applicable only for Nodes with a transport " +
+                              "factory instance of OtpGenericTransportFactory");
+    }
+
+    /**
+     * Create a server-side transport for alternative distribution protocols
+     * using a transport factory extending the OtpGenericTransportFactory
+     * abstract class.
+     *
+     * @param node
+     *            the local node identifying the transport to create server-side
+     *
+     */
+    public OtpServerTransport createServerTransport(final OtpLocalNode node)
+            throws IOException {
+        if (transportFactory instanceof OtpGenericTransportFactory) {
+            return ((OtpGenericTransportFactory) transportFactory)
+                .createServerTransport(node);
+        }
+        throw new IOException("Method createServerTransport(OtpLocalNode) " +
+                              "applicable only for Nodes with a transport " +
+                              "factory instance of OtpGenericTransportFactory");
+    }
 }
diff --git a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpGenericTransportFactory.java b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpGenericTransportFactory.java
new file mode 100644
index 0000000000..c5858c8953
--- /dev/null
+++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpGenericTransportFactory.java
@@ -0,0 +1,87 @@
+/*
+ * Copyright 2022 Jérôme de Bretagne
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * %ExternalCopyright%
+ */
+
+package com.ericsson.otp.erlang;
+
+import java.io.IOException;
+import java.net.InetAddress;
+
+/**
+ * Transport factory abstract class used to create client-side and server-side
+ * transport instances defined using generic peers and local nodes (instead of
+ * a host + port combination as expected in the base OtpTransportFactory).
+ *
+ * It allows the creation of a transport using Unix Domain Sockets for example.
+ *
+ * OtpGenericTransportFactory is created as a subclass of OtpTransportFactory
+ * to keep backwards compatibility and ease the integration within existing
+ * Jinterface code, but in practice it doesn't support the 3 original methods.
+ */
+public abstract class OtpGenericTransportFactory implements OtpTransportFactory {
+
+    /**
+     * Create an instance of a client-side {@link OtpTransport}
+     *
+     * @param peer
+     *            the peer identifying the server to connect to
+     *
+     * @return a new transport object
+     *
+     * @throws IOException
+     */
+    public abstract OtpTransport createTransport(final OtpPeer peer)
+            throws IOException;
+
+    /**
+     * Create an instance of a server-side {@link OtpServerTransport}
+     *
+     * @param node
+     *            the local node identifying the transport to create server-side
+     *
+     * @return a new transport object
+     *
+     * @throws IOException
+     */
+    public abstract OtpServerTransport createServerTransport(final OtpLocalNode node)
+            throws IOException;
+
+
+    /**
+     * Implement the 3 original methods by throwing an exception as the usage
+     * of a port is not supported by this subclass of OtpTransportFactory.
+     */
+    @Override public OtpTransport createTransport(String addr, int port)
+            throws IOException {
+        throw new IOException("Method createTransport(String, int) " +
+                              "not applicable for OtpGenericTransportFactory");
+    }
+
+    @Override public OtpTransport createTransport(final InetAddress addr, final int port)
+            throws IOException {
+        throw new IOException("Method createTransport(InetAddress, int) " +
+                              "not applicable for OtpGenericTransportFactory");
+    }
+
+    @Override public OtpServerTransport createServerTransport(final int port)
+            throws IOException {
+        throw new IOException("Method createServerTransport(int) " +
+                              "not applicable for OtpGenericTransportFactory");
+    }
+
+}
+
diff --git a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpNode.java b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpNode.java
index e52c7714ec..b0e43b9c11 100644
--- a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpNode.java
+++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpNode.java
@@ -52,9 +52,12 @@ import java.util.Iterator;
  *
  * <p>
  * Note that the use of this class requires that Epmd (Erlang Port Mapper
- * Daemon) is running on each cooperating host. This class does not start Epmd
- * automatically as Erlang does, you must start it manually or through some
- * other means. See the Erlang documentation for more information about this.
+ * Daemon) is running on each cooperating host, except when using a transport
+ * factory extending the OtpGenericTransportFactory abstract class.
+ *
+ * This class does not start Epmd automatically as Erlang does, you must start
+ * it manually or through some other means. See the Erlang documentation for
+ * more information about this.
  * </p>
  */
 public class OtpNode extends OtpLocalNode {
@@ -128,7 +131,14 @@ public class OtpNode extends OtpLocalNode {
             final OtpTransportFactory transportFactory) throws IOException {
         super(node, transportFactory);
 
-        init(0);
+        if (transportFactory instanceof OtpGenericTransportFactory) {
+            // For alternative distribution protocols using a transport factory
+            // extending the OtpGenericTransportFactory abstract class, the
+            // local node is used as the identifier for incoming connections.
+            init();
+        } else {
+            init(0);
+        }
     }
 
     /**
@@ -168,7 +178,16 @@ public class OtpNode extends OtpLocalNode {
      */
     public OtpNode(final String node, final String cookie,
             final OtpTransportFactory transportFactory) throws IOException {
-        this(node, cookie, 0, transportFactory);
+        super(node, cookie, transportFactory);
+
+        if (transportFactory instanceof OtpGenericTransportFactory) {
+            // For alternative distribution protocols using a transport factory
+            // extending the OtpGenericTransportFactory abstract class, the
+            // local node is used as the identifier for incoming connections.
+            init();
+        } else {
+            init(0);
+        }
     }
 
     /**
@@ -224,6 +243,10 @@ public class OtpNode extends OtpLocalNode {
         init(port);
     }
 
+    /*
+     * Initialize a node instance with the port number to use for incoming
+     * connections. Specifying 0 lets the system choose an available port.
+     */
     private synchronized void init(final int aport) throws IOException {
         if (!initDone) {
             connections = new Hashtable<String, OtpCookedConnection>(17,
@@ -234,6 +257,20 @@ public class OtpNode extends OtpLocalNode {
         }
     }
 
+    /*
+     * Initialize a node instance using an alternative distribution protocol
+     * with the local node used as the identifier for incoming connections.
+     */
+    private synchronized void init() throws IOException {
+        if (!initDone) {
+            connections = new Hashtable<String, OtpCookedConnection>(17,
+                    (float) 0.95);
+            mboxes = new Mailboxes();
+            acceptor = new Acceptor(this);
+            initDone = true;
+        }
+    }
+
     /**
      * Close the node. Unpublish the node from Epmd (preventing new connections)
      * and close all existing connections.
@@ -446,8 +483,8 @@ public class OtpNode extends OtpLocalNode {
     public boolean ping(final String anode, final long timeout) {
         if (anode.equals(node)) {
             return true;
-        } else if (anode.indexOf('@', 0) < 0
-                && anode.equals(node.substring(0, node.indexOf('@', 0)))) {
+        } else if (anode.indexOf('@') < 0
+                && anode.equals(node.substring(0, node.indexOf('@')))) {
             return true;
         }
 
@@ -527,7 +564,7 @@ public class OtpNode extends OtpLocalNode {
 
             if (t == OtpMsg.regSendTag) {
                 final String name = m.getRecipientName();
-                /* special case for netKernel requests */
+                // special case for netKernel requests
                 if (name.equals("net_kernel")) {
                     return netKernel(m);
                 }
@@ -569,7 +606,10 @@ public class OtpNode extends OtpLocalNode {
 
             if (conn == null) {
                 // in case node had no '@' add localhost info and try again
-                peer = new OtpPeer(anode);
+                peer = new OtpPeer(anode,
+                                   // Pass the transport factory to use
+                                   // when creating connections
+                                   transportFactory);
                 conn = connections.get(peer.node());
 
                 if (conn == null) {
@@ -578,7 +618,7 @@ public class OtpNode extends OtpLocalNode {
                         conn.setFlags(connFlags);
                         addConnection(conn);
                     } catch (final Exception e) {
-                        /* false = outgoing */
+                        // false = outgoing
                         connAttempt(peer.node(), false, e);
                     }
                 }
@@ -773,10 +813,31 @@ public class OtpNode extends OtpLocalNode {
 
             setDaemon(true);
             setName("acceptor");
+            // Publish the node through Epmd
             publishPort();
             start();
         }
 
+        /*
+         * Constructor for alternative distribution protocols using a transport
+         * factory extending the OtpGenericTransportFactory abstract class.
+         *
+         * The notion of socket port is not used with such protocols so the
+         * node is not published via Epmd.
+         */
+        Acceptor(final OtpLocalNode node) throws IOException {
+            // Set acceptorPort arbitrarily to 0, it won't be used later on.
+            acceptorPort = 0;
+
+            // The local node is passed as the identifier to use for incoming
+            // connections.
+            sock = createServerTransport(node);
+
+            setDaemon(true);
+            setName("acceptor");
+            start();
+        }
+
         private boolean publishPort() throws IOException {
             if (getEpmd() != null) {
                 return false; // already published
@@ -786,10 +847,16 @@ public class OtpNode extends OtpLocalNode {
         }
 
         private void unPublishPort() {
-            // unregister with epmd
+            // The notion of socket port is not used with such an alternative
+            // distribution protocol so the node was not registered with Epmd.
+            if (transportFactory instanceof OtpGenericTransportFactory) {
+                return;
+            }
+
+            // Unregister the node from Epmd
             OtpEpmd.unPublishPort(OtpNode.this);
 
-            // close the local descriptor (if we have one)
+            // Close the local descriptor (if we have one)
             closeSock(epmd);
             epmd = null;
         }
@@ -875,7 +942,7 @@ public class OtpNode extends OtpLocalNode {
                 }
             } // while
 
-            // if we have exited loop we must do this too
+            // If we have exited loop we must do this too
             unPublishPort();
         }
     }
diff --git a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpSelf.java b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpSelf.java
index 2ad4b02256..9410aba9a5 100644
--- a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpSelf.java
+++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpSelf.java
@@ -196,12 +196,20 @@ public class OtpSelf extends OtpLocalNode {
             final OtpTransportFactory transportFactory) throws IOException {
         super(node, cookie, transportFactory);
 
-        sock = createServerTransport(port);
+        if (transportFactory instanceof OtpGenericTransportFactory) {
+            // For alternative distribution protocols using a transport factory
+            // extending the OtpGenericTransportFactory abstract class, pass the
+            // local node as the identifier to use for incoming connections.
+            sock = createServerTransport(this);
 
-        if (port != 0) {
-            this.port = port;
         } else {
-            this.port = sock.getLocalPort();
+            sock = createServerTransport(port);
+
+            if (port != 0) {
+                this.port = port;
+            } else {
+                this.port = sock.getLocalPort();
+            }
         }
 
         pid = createPid();
diff --git a/lib/jinterface/java_src/com/ericsson/otp/erlang/java_files b/lib/jinterface/java_src/com/ericsson/otp/erlang/java_files
index cfabbe6271..10709e1059 100644
--- a/lib/jinterface/java_src/com/ericsson/otp/erlang/java_files
+++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/java_files
@@ -2,9 +2,9 @@
 
 #
 # %CopyrightBegin%
-# 
-# Copyright Ericsson AB 2000-2016. All Rights Reserved.
-# 
+#
+# Copyright Ericsson AB 2000-2022. All Rights Reserved.
+#
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
@@ -16,7 +16,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-# 
+#
 # %CopyrightEnd%
 #
 
@@ -44,6 +44,7 @@ COMM = \
 	OtpErlangFun \
 	OtpErlangExternalFun \
 	OtpExternal \
+	OtpGenericTransportFactory \
 	OtpInputStream \
 	OtpLocalNode \
 	OtpNodeStatus \