From fd76d49c7d7bbed4775818390e47b958ee50f469 Mon Sep 17 00:00:00 2001 From: Dmitriy Kargapolov Date: Sat, 31 Jan 2015 23:18:23 -0500 Subject: jinterface: transport factory implementation Transport factory basic implementation added. This makes possible creating connections between nodes using ssh channels for example. Default transport factory based on standart Socket/ServerSocket classes is provided. Modifications are backward compatible. --- lib/jinterface/doc/src/jinterface_users_guide.xml | 8 ++ .../ericsson/otp/erlang/AbstractConnection.java | 46 ++++---- .../com/ericsson/otp/erlang/AbstractNode.java | 46 +++++++- .../com/ericsson/otp/erlang/OtpConnection.java | 5 +- .../ericsson/otp/erlang/OtpCookedConnection.java | 5 +- .../java_src/com/ericsson/otp/erlang/OtpEpmd.java | 46 +++++--- .../com/ericsson/otp/erlang/OtpLocalNode.java | 30 +++-- .../java_src/com/ericsson/otp/erlang/OtpNode.java | 96 ++++++++++++++-- .../com/ericsson/otp/erlang/OtpOutputStream.java | 13 ++- .../java_src/com/ericsson/otp/erlang/OtpPeer.java | 17 ++- .../java_src/com/ericsson/otp/erlang/OtpSelf.java | 118 +++++++++++++++++++- .../otp/erlang/OtpServerSocketTransport.java | 68 +++++++++++ .../ericsson/otp/erlang/OtpServerTransport.java | 46 ++++++++ .../ericsson/otp/erlang/OtpSocketTransport.java | 89 +++++++++++++++ .../otp/erlang/OtpSocketTransportFactory.java | 56 ++++++++++ .../com/ericsson/otp/erlang/OtpTransport.java | 49 ++++++++ .../ericsson/otp/erlang/OtpTransportFactory.java | 124 +++++++++++++++++++++ .../java_src/com/ericsson/otp/erlang/java_files | 8 +- lib/jinterface/test/jinterface_SUITE.erl | 14 ++- .../test/jinterface_SUITE_data/Makefile.src | 1 + .../TransportFactoryTest.java | 90 +++++++++++++++ 21 files changed, 899 insertions(+), 76 deletions(-) create mode 100644 lib/jinterface/java_src/com/ericsson/otp/erlang/OtpServerSocketTransport.java create mode 100644 lib/jinterface/java_src/com/ericsson/otp/erlang/OtpServerTransport.java create mode 100644 lib/jinterface/java_src/com/ericsson/otp/erlang/OtpSocketTransport.java create mode 100644 lib/jinterface/java_src/com/ericsson/otp/erlang/OtpSocketTransportFactory.java create mode 100644 lib/jinterface/java_src/com/ericsson/otp/erlang/OtpTransport.java create mode 100644 lib/jinterface/java_src/com/ericsson/otp/erlang/OtpTransportFactory.java create mode 100644 lib/jinterface/test/jinterface_SUITE_data/TransportFactoryTest.java (limited to 'lib/jinterface') diff --git a/lib/jinterface/doc/src/jinterface_users_guide.xml b/lib/jinterface/doc/src/jinterface_users_guide.xml index 5dfe5c0c6d..238f90ce38 100644 --- a/lib/jinterface/doc/src/jinterface_users_guide.xml +++ b/lib/jinterface/doc/src/jinterface_users_guide.xml @@ -222,6 +222,14 @@ OtpNode node = new OtpNode("gurka"); a connection may be broken however.

+
+ Transport Factory +

All necessary connections are made using methods of + OtpTransportFactory + interface. Default OtpTransportFactory implementation is based on standard Socket class. + User may provide custom transport factory as needed. See java doc for details.

+
+
Sending and Receiving Messages

Messages sent with this package must be instances of 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 1b0fe3e2e6..ab8fa06c1b 100644 --- a/lib/jinterface/java_src/com/ericsson/otp/erlang/AbstractConnection.java +++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/AbstractConnection.java @@ -20,7 +20,7 @@ package com.ericsson.otp.erlang; import java.io.IOException; -import java.net.Socket; +import java.io.OutputStream; import java.util.Random; /** @@ -84,7 +84,7 @@ public abstract class AbstractConnection extends Thread { private volatile boolean done = false; protected boolean connected = false; // connection status - protected Socket socket; // communication channel + protected OtpTransport socket; // communication channel protected OtpPeer peer; // who are we connected to protected OtpLocalNode localNode; // this nodes id String name; // local name of this connection @@ -126,7 +126,7 @@ public abstract class AbstractConnection extends Thread { * Accept an incoming connection from a remote node. Used by * {@link OtpSelf#accept() OtpSelf.accept()} to create a connection based on * data received when handshaking with the peer node, when the remote node - * is the connection intitiator. + * is the connection initiator. * * @exception java.io.IOException * if it was not possible to connect to the peer. @@ -134,20 +134,17 @@ public abstract class AbstractConnection extends Thread { * @exception OtpAuthException * if handshake resulted in an authentication error */ - protected AbstractConnection(final OtpLocalNode self, final Socket s) + protected AbstractConnection(final OtpLocalNode self, final OtpTransport s) throws IOException, OtpAuthException { localNode = self; - peer = new OtpPeer(); + peer = new OtpPeer(self.transportFactory); socket = s; - socket.setTcpNoDelay(true); - traceLevel = defaultLevel; setDaemon(true); if (traceLevel >= handshakeThreshold) { - System.out.println("<- ACCEPT FROM " + s.getInetAddress() + ":" - + s.getPort()); + System.out.println("<- ACCEPT FROM " + s); } // get his info @@ -189,6 +186,8 @@ public abstract class AbstractConnection extends Thread { // 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 if (peer.proto != self.proto || self.distHigh < peer.distLow @@ -523,7 +522,9 @@ public abstract class AbstractConnection extends Thread { // received tick? send tock! if (len == 0) { synchronized (this) { - socket.getOutputStream().write(tock); + OutputStream out = socket.getOutputStream(); + out.write(tock); + out.flush(); } } @@ -837,8 +838,11 @@ public abstract class AbstractConnection extends Thread { } } - header.writeTo(socket.getOutputStream()); - payload.writeTo(socket.getOutputStream()); + // group flush op in favour of possible ssh-tunneled stream + OutputStream out = socket.getOutputStream(); + header.writeTo(out); + payload.writeTo(out); + out.flush(); } catch (final IOException e) { close(); throw e; @@ -859,7 +863,7 @@ public abstract class AbstractConnection extends Thread { + e); } } - header.writeTo(socket.getOutputStream()); + header.writeToAndFlush(socket.getOutputStream()); } catch (final IOException e) { close(); throw e; @@ -913,7 +917,8 @@ public abstract class AbstractConnection extends Thread { } /* this method now throws exception if we don't get full read */ - protected int readSock(final Socket s, final byte[] b) throws IOException { + protected int readSock(final OtpTransport s, final byte[] b) + throws IOException { int got = 0; final int len = b.length; int i; @@ -980,8 +985,7 @@ public abstract class AbstractConnection extends Thread { protected void doConnect(final int port) throws IOException, OtpAuthException { try { - socket = new Socket(peer.host(), port); - socket.setTcpNoDelay(true); + socket = peer.createTransport(peer.host(), port); if (traceLevel >= handshakeThreshold) { System.out.println("-> MD5 CONNECT TO " + peer.host() + ":" @@ -1077,7 +1081,7 @@ public abstract class AbstractConnection extends Thread { obuf.write4BE(aflags); obuf.write(str.getBytes()); - obuf.writeTo(socket.getOutputStream()); + obuf.writeToAndFlush(socket.getOutputStream()); if (traceLevel >= handshakeThreshold) { System.out.println("-> " + "HANDSHAKE sendName" + " flags=" @@ -1098,7 +1102,7 @@ public abstract class AbstractConnection extends Thread { obuf.write4BE(challenge); obuf.write(str.getBytes()); - obuf.writeTo(socket.getOutputStream()); + obuf.writeToAndFlush(socket.getOutputStream()); if (traceLevel >= handshakeThreshold) { System.out.println("-> " + "HANDSHAKE sendChallenge" + " flags=" @@ -1232,7 +1236,7 @@ public abstract class AbstractConnection extends Thread { obuf.write1(ChallengeReply); obuf.write4BE(challenge); obuf.write(digest); - obuf.writeTo(socket.getOutputStream()); + obuf.writeToAndFlush(socket.getOutputStream()); if (traceLevel >= handshakeThreshold) { System.out.println("-> " + "HANDSHAKE sendChallengeReply" @@ -1294,7 +1298,7 @@ public abstract class AbstractConnection extends Thread { obuf.write1(ChallengeAck); obuf.write(digest); - obuf.writeTo(socket.getOutputStream()); + obuf.writeToAndFlush(socket.getOutputStream()); if (traceLevel >= handshakeThreshold) { System.out.println("-> " + "HANDSHAKE sendChallengeAck" @@ -1341,7 +1345,7 @@ public abstract class AbstractConnection extends Thread { obuf.write1(ChallengeStatus); obuf.write(status.getBytes()); - obuf.writeTo(socket.getOutputStream()); + obuf.writeToAndFlush(socket.getOutputStream()); if (traceLevel >= handshakeThreshold) { System.out.println("-> " + "HANDSHAKE sendStatus" + " status=" 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 6f07d8171e..0a33984b31 100644 --- a/lib/jinterface/java_src/com/ericsson/otp/erlang/AbstractNode.java +++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/AbstractNode.java @@ -64,13 +64,14 @@ import java.net.UnknownHostException; * instead. *

*/ -public class AbstractNode { +public class AbstractNode implements OtpTransportFactory { static String localHost = null; String node; String host; String alive; String cookie; static String defaultCookie = null; + final OtpTransportFactory transportFactory; // Node types static final int NTYPE_R6 = 110; // 'n' post-r5, all nodes @@ -146,21 +147,41 @@ public class AbstractNode { } } - protected AbstractNode() { + protected AbstractNode(final OtpTransportFactory transportFactory) { + this.transportFactory = transportFactory; } /** - * Create a node with the given name and the default cookie. + * Create a node with the given name and default cookie and transport + * factory. */ protected AbstractNode(final String node) { - this(node, defaultCookie); + this(node, defaultCookie, new OtpSocketTransportFactory()); } /** - * Create a node with the given name and cookie. + * Create a node with the given name, transport factory and the default + * cookie. + */ + protected AbstractNode(final String node, + final OtpTransportFactory transportFactory) { + this(node, defaultCookie, transportFactory); + } + + /** + * Create a node with the given name, cookie and default transport factory. */ protected AbstractNode(final String name, final String cookie) { + this(name, cookie, new OtpSocketTransportFactory()); + } + + /** + * Create a node with the given name, cookie and transport factory. + */ + protected AbstractNode(final String name, final String cookie, + final OtpTransportFactory transportFactory) { this.cookie = cookie; + this.transportFactory = transportFactory; final int i = name.indexOf('@', 0); if (i < 0) { @@ -268,4 +289,19 @@ public class AbstractNode { } return home; } + + public OtpTransport createTransport(final String addr, final int port) + throws IOException { + return transportFactory.createTransport(addr, port); + } + + public OtpTransport createTransport(final InetAddress addr, final int port) + throws IOException { + return transportFactory.createTransport(addr, port); + } + + public OtpServerTransport createServerTransport(final int port) + throws IOException { + return transportFactory.createServerTransport(port); + } } diff --git a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpConnection.java b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpConnection.java index 2c9b7766bc..af0926f939 100644 --- a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpConnection.java +++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpConnection.java @@ -19,7 +19,6 @@ package com.ericsson.otp.erlang; import java.io.IOException; -import java.net.Socket; /** * Maintains a connection between a Java process and a remote Erlang, Java or C @@ -63,8 +62,8 @@ public class OtpConnection extends AbstractConnection { * error */ // package scope - OtpConnection(final OtpSelf self, final Socket s) throws IOException, - OtpAuthException { + OtpConnection(final OtpSelf self, final OtpTransport s) + throws IOException, OtpAuthException { super(self, s); this.self = self; queue = new GenericQueue(); diff --git a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpCookedConnection.java b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpCookedConnection.java index 4d80f61d52..b0e3e81fca 100644 --- a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpCookedConnection.java +++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpCookedConnection.java @@ -19,7 +19,6 @@ package com.ericsson.otp.erlang; import java.io.IOException; -import java.net.Socket; /** *

@@ -78,8 +77,8 @@ public class OtpCookedConnection extends AbstractConnection { * error */ // package scope - OtpCookedConnection(final OtpNode self, final Socket s) throws IOException, - OtpAuthException { + OtpCookedConnection(final OtpNode self, final OtpTransport s) + throws IOException, OtpAuthException { super(self, s); this.self = self; links = new Links(25); diff --git a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpEpmd.java b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpEpmd.java index 796babee1b..6c7c8fe951 100644 --- a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpEpmd.java +++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpEpmd.java @@ -21,13 +21,12 @@ package com.ericsson.otp.erlang; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.net.InetAddress; -import java.net.Socket; /** * Provides methods for registering, unregistering and looking up nodes with the * Erlang portmapper daemon (Epmd). For each registered node, Epmd maintains * information about the port on which incoming connections are accepted, as - * well as which versions of the Erlang communication protocolt the node + * well as which versions of the Erlang communication protocol the node * supports. * *

@@ -136,7 +135,7 @@ public class OtpEpmd { */ public static boolean publishPort(final OtpLocalNode node) throws IOException { - Socket s = null; + OtpTransport s = null; s = r4_publish(node); @@ -156,16 +155,16 @@ public class OtpEpmd { * This method does not report any failures. */ public static void unPublishPort(final OtpLocalNode node) { - Socket s = null; + OtpTransport s = null; try { - s = new Socket((String) null, EpmdPort.get()); + s = node.createTransport((String) null, EpmdPort.get()); @SuppressWarnings("resource") final OtpOutputStream obuf = new OtpOutputStream(); obuf.write2BE(node.alive().length() + 1); obuf.write1(stopReq); obuf.writeN(node.alive().getBytes()); - obuf.writeTo(s.getOutputStream()); + obuf.writeToAndFlush(s.getOutputStream()); // don't even wait for a response (is there one?) if (traceLevel >= traceThreshold) { System.out.println("-> UNPUBLISH " + node + " port=" @@ -187,12 +186,12 @@ public class OtpEpmd { private static int r4_lookupPort(final AbstractNode node) throws IOException { int port = 0; - Socket s = null; + OtpTransport s = null; try { @SuppressWarnings("resource") final OtpOutputStream obuf = new OtpOutputStream(); - s = new Socket(node.host(), EpmdPort.get()); + s = node.createTransport(node.host(), EpmdPort.get()); // build and send epmd request // length[2], tag[1], alivename[n] (length = n+1) @@ -201,7 +200,7 @@ public class OtpEpmd { obuf.writeN(node.alive().getBytes()); // send request - obuf.writeTo(s.getOutputStream()); + obuf.writeToAndFlush(s.getOutputStream()); if (traceLevel >= traceThreshold) { System.out.println("-> LOOKUP (r4) " + node); @@ -242,7 +241,7 @@ public class OtpEpmd { System.out.println("<- (no response)"); } throw new IOException("Nameserver not responding on " + node.host() - + " when looking up " + node.alive()); + + " when looking up " + node.alive(), e); } catch (final OtpErlangDecodeException e) { if (traceLevel >= traceThreshold) { System.out.println("<- (invalid response)"); @@ -276,14 +275,14 @@ public class OtpEpmd { * fatal. If we manage to successfully communicate with an r4 epmd, we * return either the socket, or null, depending on the result. */ - private static Socket r4_publish(final OtpLocalNode node) + private static OtpTransport r4_publish(final OtpLocalNode node) throws IOException { - Socket s = null; + OtpTransport s = null; try { @SuppressWarnings("resource") final OtpOutputStream obuf = new OtpOutputStream(); - s = new Socket((String) null, EpmdPort.get()); + s = node.createTransport((String) null, EpmdPort.get()); obuf.write2BE(node.alive().length() + 13); @@ -301,7 +300,7 @@ public class OtpEpmd { obuf.write2BE(0); // No extra // send request - obuf.writeTo(s.getOutputStream()); + obuf.writeToAndFlush(s.getOutputStream()); if (traceLevel >= traceThreshold) { System.out.println("-> PUBLISH (r4) " + node + " port=" @@ -356,23 +355,34 @@ public class OtpEpmd { } public static String[] lookupNames() throws IOException { - return lookupNames(InetAddress.getByName(null)); + return lookupNames(InetAddress.getByName(null), + new OtpSocketTransportFactory()); + } + + public static String[] lookupNames( + final OtpTransportFactory transportFactory) throws IOException { + return lookupNames(InetAddress.getByName(null), transportFactory); } public static String[] lookupNames(final InetAddress address) throws IOException { - Socket s = null; + return lookupNames(address, new OtpSocketTransportFactory()); + } + + public static String[] lookupNames(final InetAddress address, + final OtpTransportFactory transportFactory) throws IOException { + OtpTransport s = null; try { @SuppressWarnings("resource") final OtpOutputStream obuf = new OtpOutputStream(); try { - s = new Socket(address, EpmdPort.get()); + s = transportFactory.createTransport(address, EpmdPort.get()); obuf.write2BE(1); obuf.write1(names4req); // send request - obuf.writeTo(s.getOutputStream()); + obuf.writeToAndFlush(s.getOutputStream()); if (traceLevel >= traceThreshold) { System.out.println("-> NAMES (r4) "); diff --git a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpLocalNode.java b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpLocalNode.java index b996ba6f6c..dd1d299297 100644 --- a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpLocalNode.java +++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpLocalNode.java @@ -29,12 +29,7 @@ public class OtpLocalNode extends AbstractNode { private int refId[]; protected int port; - protected java.net.Socket epmd; - - protected OtpLocalNode() { - super(); - init(); - } + protected OtpTransport epmd; /** * Create a node with the given name and the default cookie. @@ -44,6 +39,16 @@ public class OtpLocalNode extends AbstractNode { init(); } + /** + * Create a node with the given name, transport factory and the default + * cookie. + */ + protected OtpLocalNode(final String node, + final OtpTransportFactory transportFactory) { + super(node, transportFactory); + init(); + } + /** * Create a node with the given name and cookie. */ @@ -52,6 +57,15 @@ public class OtpLocalNode extends AbstractNode { init(); } + /** + * Create a node with the given name, cookie and transport factory. + */ + protected OtpLocalNode(final String node, final String cookie, + final OtpTransportFactory transportFactory) { + super(node, cookie, transportFactory); + init(); + } + private void init() { serial = 0; pidCount = 1; @@ -77,7 +91,7 @@ public class OtpLocalNode extends AbstractNode { * @param s * The socket connecting this node to Epmd. */ - protected void setEpmd(final java.net.Socket s) { + protected void setEpmd(final OtpTransport s) { epmd = s; } @@ -86,7 +100,7 @@ public class OtpLocalNode extends AbstractNode { * * @return The socket connecting this node to Epmd. */ - protected java.net.Socket getEpmd() { + protected OtpTransport getEpmd() { return epmd; } 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 d5edd135cf..7512d34c21 100644 --- a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpNode.java +++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpNode.java @@ -20,8 +20,6 @@ package com.ericsson.otp.erlang; import java.io.IOException; import java.lang.ref.WeakReference; -import java.net.ServerSocket; -import java.net.Socket; import java.util.Collection; import java.util.Enumeration; import java.util.Hashtable; @@ -97,7 +95,39 @@ public class OtpNode extends OtpLocalNode { * */ public OtpNode(final String node) throws IOException { - this(node, defaultCookie, 0); + super(node); + + init(0); + } + + /** + *

+ * Create a node using the default cookie. The default cookie is found by + * reading the first line of the .erlang.cookie file in the user's home + * directory. The home directory is obtained from the System property + * "user.home". + *

+ * + *

+ * If the file does not exist, an empty string is used. This method makes no + * attempt to create the file. + *

+ * + * @param node + * the name of this node. + * + * @param transportFactory + * the transport factory to use when creating connections. + * + * @exception IOException + * if communication could not be initialized. + * + */ + public OtpNode(final String node, + final OtpTransportFactory transportFactory) throws IOException { + super(node, transportFactory); + + init(0); } /** @@ -118,6 +148,28 @@ public class OtpNode extends OtpLocalNode { this(node, cookie, 0); } + /** + * Create a node. + * + * @param node + * the name of this node. + * + * @param cookie + * the authorization cookie that will be used by this node when + * it communicates with other nodes. + * + * @param transportFactory + * the transport factory to use when creating connections. + * + * @exception IOException + * if communication could not be initialized. + * + */ + public OtpNode(final String node, final String cookie, + final OtpTransportFactory transportFactory) throws IOException { + this(node, cookie, 0, transportFactory); + } + /** * Create a node. * @@ -143,6 +195,34 @@ public class OtpNode extends OtpLocalNode { init(port); } + /** + * Create a node. + * + * @param node + * the name of this node. + * + * @param cookie + * the authorization cookie that will be used by this node when + * it communicates with other nodes. + * + * @param port + * the port number you wish to use for incoming connections. + * Specifying 0 lets the system choose an available port. + * + * @param transportFactory + * the transport factory to use when creating connections. + * + * @exception IOException + * if communication could not be initialized. + * + */ + public OtpNode(final String node, final String cookie, final int port, + final OtpTransportFactory transportFactory) throws IOException { + super(node, cookie, transportFactory); + + init(port); + } + private synchronized void init(final int aport) throws IOException { if (!initDone) { connections = new Hashtable(17, @@ -681,12 +761,12 @@ public class OtpNode extends OtpLocalNode { * this thread simply listens for incoming connections */ public class Acceptor extends Thread { - private final ServerSocket sock; + private final OtpServerTransport sock; private final int acceptorPort; private volatile boolean done = false; Acceptor(final int port) throws IOException { - sock = new ServerSocket(port); + sock = createServerTransport(port); acceptorPort = sock.getLocalPort(); OtpNode.this.port = acceptorPort; @@ -720,7 +800,7 @@ public class OtpNode extends OtpLocalNode { localStatus(node, false, null); } - private void closeSock(final ServerSocket s) { + private void closeSock(final OtpServerTransport s) { try { if (s != null) { s.close(); @@ -729,7 +809,7 @@ public class OtpNode extends OtpLocalNode { } } - private void closeSock(final Socket s) { + private void closeSock(final OtpTransport s) { try { if (s != null) { s.close(); @@ -744,7 +824,7 @@ public class OtpNode extends OtpLocalNode { @Override public void run() { - Socket newsock = null; + OtpTransport newsock = null; OtpCookedConnection conn = null; localStatus(node, true, null); diff --git a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpOutputStream.java b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpOutputStream.java index b8493b57ff..2ec583ff5c 100644 --- a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpOutputStream.java +++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpOutputStream.java @@ -21,6 +21,7 @@ package com.ericsson.otp.erlang; // import java.io.OutputStream; import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.io.OutputStream; import java.io.UnsupportedEncodingException; import java.math.BigDecimal; import java.math.BigInteger; @@ -202,6 +203,16 @@ public class OtpOutputStream extends ByteArrayOutputStream { super.count += len; } + @Override + public synchronized void writeTo(OutputStream out) throws IOException { + super.writeTo(out); + } + + public synchronized void writeToAndFlush(OutputStream out) throws IOException { + super.writeTo(out); + out.flush(); + } + /** * Write the low byte of a value to the stream. * @@ -887,7 +898,7 @@ public class OtpOutputStream extends ByteArrayOutputStream { if (oos.size() < 5) { // fast path for small terms try { - oos.writeTo(this); + oos.writeToAndFlush(this); // if the term is written as a compressed term, the output // stream is closed, so we do this here, too close(); diff --git a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpPeer.java b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpPeer.java index 2c79c04247..cb09b40f47 100644 --- a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpPeer.java +++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpPeer.java @@ -32,8 +32,8 @@ public class OtpPeer extends AbstractNode { * common protocol version we both support */ - OtpPeer() { - super(); + OtpPeer(final OtpTransportFactory transportFactory) { + super(transportFactory); } /** @@ -46,6 +46,19 @@ public class OtpPeer extends AbstractNode { super(node); } + /** + * Create a peer node with custom transport factory. + * + * @param node + * the name of the node. + * @param transportFactory + * custom transport factory + */ + public OtpPeer(final String node, final OtpTransportFactory + transportFactory) { + super(node, transportFactory); + } + /** * Create a connection to a remote node. * 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 166dac5701..5b9d13ad81 100644 --- a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpSelf.java +++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpSelf.java @@ -19,8 +19,6 @@ package com.ericsson.otp.erlang; import java.io.IOException; -import java.net.ServerSocket; -import java.net.Socket; import java.net.UnknownHostException; /** @@ -48,7 +46,7 @@ import java.net.UnknownHostException; * */ public class OtpSelf extends OtpLocalNode { - private final ServerSocket sock; + private final OtpServerTransport sock; private final OtpErlangPid pid; /** @@ -67,11 +65,42 @@ public class OtpSelf extends OtpLocalNode { * @param node * the name of this node. * + * @exception IOException + * in case of server transport failure + * */ public OtpSelf(final String node) throws IOException { this(node, defaultCookie, 0); } + /** + *

+ * Create a self node using the default cookie and custom transport factory. + * The default cookie is found by reading the first line of the + * .erlang.cookie file in the user's home directory. The home directory is + * obtained from the System property "user.home". + *

+ * + *

+ * If the file does not exist, an empty string is used. This method makes no + * attempt to create the file. + *

+ * + * @param node + * the name of this node. + * + * @param transportFactory + * the transport factory to use when creating connections. + * + * @exception IOException + * in case of server transport failure + * + */ + public OtpSelf(final String node, + final OtpTransportFactory transportFactory) throws IOException { + this(node, defaultCookie, 0, transportFactory); + } + /** * Create a self node. * @@ -81,16 +110,95 @@ public class OtpSelf extends OtpLocalNode { * @param cookie * the authorization cookie that will be used by this node when * it communicates with other nodes. + * + * @exception IOException + * in case of server transport failure */ public OtpSelf(final String node, final String cookie) throws IOException { this(node, cookie, 0); } + /** + * Create a self node. + * + * @param node + * the name of this node. + * + * @param cookie + * the authorization cookie that will be used by this node when + * it communicates with other nodes. + * + * @param transportFactory + * the transport factory to use when creating connections. + * + * @exception IOException + * in case of server transport failure + */ + public OtpSelf(final String node, final String cookie, + final OtpTransportFactory transportFactory) throws IOException { + this(node, cookie, 0, transportFactory); + } + + /** + * Create a self node. + * + * @param node + * the name of this node. + * + * @param cookie + * the authorization cookie that will be used by this node when + * it communicates with other nodes. + * + * @param port + * the port number you wish to use for incoming connections. + * Specifying 0 lets the system choose an available port. + * + * @param transportFactory + * the transport factory to use when creating connections. + * + * @exception IOException + * in case of server transport failure + */ public OtpSelf(final String node, final String cookie, final int port) throws IOException { super(node, cookie); - sock = new ServerSocket(port); + sock = createServerTransport(port); + + if (port != 0) { + this.port = port; + } else { + this.port = sock.getLocalPort(); + } + + pid = createPid(); + } + + /** + * Create a self node. + * + * @param node + * the name of this node. + * + * @param cookie + * the authorization cookie that will be used by this node when + * it communicates with other nodes. + * + * @param port + * the port number you wish to use for incoming connections. + * Specifying 0 lets the system choose an available port. + * + * @param transportFactory + * the transport factory to use when creating connections. + * + * @exception IOException + * in case of server transport failure + */ + public OtpSelf(final String node, final String cookie, final int port, + final OtpTransportFactory transportFactory) throws IOException { + super(node, cookie, transportFactory); + + sock = createServerTransport(port); if (port != 0) { this.port = port; @@ -179,7 +287,7 @@ public class OtpSelf extends OtpLocalNode { * authorized to connect. */ public OtpConnection accept() throws IOException, OtpAuthException { - Socket newsock = null; + OtpTransport newsock = null; while (true) { try { diff --git a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpServerSocketTransport.java b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpServerSocketTransport.java new file mode 100644 index 0000000000..0e25b6bfb7 --- /dev/null +++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpServerSocketTransport.java @@ -0,0 +1,68 @@ +/* + * %CopyrightBegin% + * + * Copyright Ericsson AB 2015. All Rights Reserved. + * + * The contents of this file are subject to the Erlang Public License, + * Version 1.1, (the "License"); you may not use this file except in + * compliance with the License. You should have received a copy of the + * Erlang Public License along with this software. If not, it can be + * retrieved online at http://www.erlang.org/. + * + * Software distributed under the License is distributed on an "AS IS" + * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See + * the License for the specific language governing rights and limitations + * under the License. + * + * %CopyrightEnd% + */ + +package com.ericsson.otp.erlang; + +import java.io.IOException; +import java.net.ServerSocket; +import java.net.Socket; + +/** + * Default socket-based server transport + * + * @author Dmitriy Kargapolov + */ +public class OtpServerSocketTransport implements OtpServerTransport { + + /** + * Underlying server socket + */ + private final ServerSocket socket; + + /** + * @see ServerSocket#ServerSocket(int) + */ + public OtpServerSocketTransport(final int port) throws IOException { + socket = new ServerSocket(port); + } + + /** + * @see ServerSocket#getLocalPort() + */ + public int getLocalPort() { + return socket.getLocalPort(); + } + + /** + * @see ServerSocket#accept() + */ + public OtpTransport accept() throws IOException { + final Socket sock = socket.accept(); + sock.setTcpNoDelay(true); + return new OtpSocketTransport(sock); + } + + /** + * @see ServerSocket#close() + */ + public void close() throws IOException { + socket.close(); + } + +} diff --git a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpServerTransport.java b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpServerTransport.java new file mode 100644 index 0000000000..4d31380bee --- /dev/null +++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpServerTransport.java @@ -0,0 +1,46 @@ +/* + * %CopyrightBegin% + * + * Copyright Ericsson AB 2015. All Rights Reserved. + * + * The contents of this file are subject to the Erlang Public License, + * Version 1.1, (the "License"); you may not use this file except in + * compliance with the License. You should have received a copy of the + * Erlang Public License along with this software. If not, it can be + * retrieved online at http://www.erlang.org/. + * + * Software distributed under the License is distributed on an "AS IS" + * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See + * the License for the specific language governing rights and limitations + * under the License. + * + * %CopyrightEnd% + */ + +package com.ericsson.otp.erlang; + +import java.io.IOException; +import java.net.ServerSocket; + +/** + * Server-side connection-oriented transport interface. + * + * @author Dmitriy Kargapolov + */ +public interface OtpServerTransport { + + /** + * @see ServerSocket#getLocalPort() + */ + int getLocalPort(); + + /** + * @see ServerSocket#accept() + */ + OtpTransport accept() throws IOException; + + /** + * @see ServerSocket#close() + */ + void close() throws IOException; +} diff --git a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpSocketTransport.java b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpSocketTransport.java new file mode 100644 index 0000000000..f690ab59ed --- /dev/null +++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpSocketTransport.java @@ -0,0 +1,89 @@ +/* + * %CopyrightBegin% + * + * Copyright Ericsson AB 2015. All Rights Reserved. + * + * The contents of this file are subject to the Erlang Public License, + * Version 1.1, (the "License"); you may not use this file except in + * compliance with the License. You should have received a copy of the + * Erlang Public License along with this software. If not, it can be + * retrieved online at http://www.erlang.org/. + * + * Software distributed under the License is distributed on an "AS IS" + * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See + * the License for the specific language governing rights and limitations + * under the License. + * + * %CopyrightEnd% + */ + +package com.ericsson.otp.erlang; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.Socket; +import java.net.UnknownHostException; + +/** + * Default socket-based client transport + * + * @author Dmitriy Kargapolov + */ +public class OtpSocketTransport implements OtpTransport { + + /** + * Underlying socket + */ + private final Socket socket; + + /** + * @see Socket#Socket(String, int) + */ + public OtpSocketTransport(final String addr, final int port) + throws UnknownHostException, IOException { + socket = new Socket(addr, port); + socket.setTcpNoDelay(true); + } + + /** + * @see Socket#Socket(InetAddress, int) + */ + public OtpSocketTransport(final InetAddress addr, final int port) + throws UnknownHostException, IOException { + socket = new Socket(addr, port); + socket.setTcpNoDelay(true); + } + + /** + * Socket wrapping constructor + * + * @param s + * socket to wrap + */ + public OtpSocketTransport(final Socket s) { + socket = s; + } + + /** + * @see Socket#getInputStream() + */ + public InputStream getInputStream() throws IOException { + return socket.getInputStream(); + } + + /** + * @see Socket#getOutputStream() + */ + public OutputStream getOutputStream() throws IOException { + return socket.getOutputStream(); + } + + /** + * @see Socket#close() + */ + public void close() throws IOException { + socket.close(); + } +} diff --git a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpSocketTransportFactory.java b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpSocketTransportFactory.java new file mode 100644 index 0000000000..f6b5bfc86d --- /dev/null +++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpSocketTransportFactory.java @@ -0,0 +1,56 @@ +/* + * %CopyrightBegin% + * + * Copyright Ericsson AB 2015. All Rights Reserved. + * + * The contents of this file are subject to the Erlang Public License, + * Version 1.1, (the "License"); you may not use this file except in + * compliance with the License. You should have received a copy of the + * Erlang Public License along with this software. If not, it can be + * retrieved online at http://www.erlang.org/. + * + * Software distributed under the License is distributed on an "AS IS" + * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See + * the License for the specific language governing rights and limitations + * under the License. + * + * %CopyrightEnd% + */ + +package com.ericsson.otp.erlang; + +import java.io.IOException; +import java.net.InetAddress; + +/** + * Default socket-based transport factory + * + * @author Dmitriy Kargapolov + */ +public class OtpSocketTransportFactory implements OtpTransportFactory { + + /** + * @see OtpTransportFactory#createTransport(String, int) + */ + public OtpTransport createTransport(final String addr, final int port) + throws IOException { + return new OtpSocketTransport(addr, port); + } + + /** + * @see OtpTransportFactory#createTransport(InetAddress, int) + */ + public OtpTransport createTransport(final InetAddress addr, final int port) + throws IOException { + return new OtpSocketTransport(addr, port); + } + + /** + * @see OtpTransportFactory#createServerTransport(int) + */ + public OtpServerTransport createServerTransport(final int port) + throws IOException { + return new OtpServerSocketTransport(port); + } + +} diff --git a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpTransport.java b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpTransport.java new file mode 100644 index 0000000000..51c62d9ef0 --- /dev/null +++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpTransport.java @@ -0,0 +1,49 @@ +/* + * %CopyrightBegin% + * + * Copyright Ericsson AB 2015. All Rights Reserved. + * + * The contents of this file are subject to the Erlang Public License, + * Version 1.1, (the "License"); you may not use this file except in + * compliance with the License. You should have received a copy of the + * Erlang Public License along with this software. If not, it can be + * retrieved online at http://www.erlang.org/. + * + * Software distributed under the License is distributed on an "AS IS" + * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See + * the License for the specific language governing rights and limitations + * under the License. + * + * %CopyrightEnd% + */ + +package com.ericsson.otp.erlang; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.Socket; + +/** + * Client-side connection-oriented transport interface. + * + * @author Dmitriy Kargapolov + */ +public interface OtpTransport { + + /** + * @see Socket#getInputStream() + */ + public abstract InputStream getInputStream() throws IOException; + + /** + * @see Socket#getOutputStream() + */ + public abstract OutputStream getOutputStream() throws IOException; + + /** + * @see Socket#close() + */ + public abstract void close() throws IOException; + +} diff --git a/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpTransportFactory.java b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpTransportFactory.java new file mode 100644 index 0000000000..bd404daea5 --- /dev/null +++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/OtpTransportFactory.java @@ -0,0 +1,124 @@ +/* + * %CopyrightBegin% + * + * Copyright Ericsson AB 2015. All Rights Reserved. + * + * The contents of this file are subject to the Erlang Public License, + * Version 1.1, (the "License"); you may not use this file except in + * compliance with the License. You should have received a copy of the + * Erlang Public License along with this software. If not, it can be + * retrieved online at http://www.erlang.org/. + * + * Software distributed under the License is distributed on an "AS IS" + * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See + * the License for the specific language governing rights and limitations + * under the License. + * + * %CopyrightEnd% + */ + +package com.ericsson.otp.erlang; + +import java.io.IOException; +import java.net.InetAddress; + +/** + * Factory class used to create client- and server-side transport instances. One + * static instance of class implementing this interface is created when program + * loaded. Default implementation used is {@link OtpSocketTransportFactory}. + * JInterface user can specify custom transport factory implementing this + * interface in the following ways: + *
+ *
defining static class as internal to class holding main() method
+ *
In the systems, where main class can be retrieved with + * System.getProperty("sun.java.command"), user can define static + * class OtpErlangSystemTuner internal to the main class, providing at + * least one static method with the name getOtpTransportFactory, with no + * parameters, returning object of class implementing + * OtpTransportFactory, for example: + * + * 
+ * 
+ * public class MyMainClass {
+ * 
+ *     public static class OtpErlangSystemTuner {
+ *         ...
+ *         public static OtpTransportFactory getOtpTransportFactory() {
+ *             return new MyTransportFactory();
+ *         }
+ *     }
+ * 
+ *     public static class MyTransportFactory implements OtpTransportFactory {
+ *         ...
+ *     }
+ * 
+ *     public static void main(String[] args) {
+ *         ...
+ *     }
+ * }
+ * 
+ * 
+ *
+ * + *
+ * + *
specifying factory class in the system properties
+ *
User-defined transport factory class may be specified via system property + * OtpTransportFactory, for example: + * + * 
+ * 
+ * package com.my.company;
+ * 
+ * public static class MyTransportFactory implements OtpTransportFactory {
+ *     ...
+ * }
+ *
+ * + * In such case program may be run with + * -DOtpTransportFactory=com.my.company.MyTransportFactory, or other way of + * setting system property before execution of static initializers may be + * used.
+ *
+ * + * @author Dmitriy Kargapolov + */ +public interface OtpTransportFactory { + + /** + * Create instance of {@link OtpTransport} + * + * @param addr + * host name or IP address string + * @param port + * port number + * @return new socket object + * @throws IOException + */ + public abstract OtpTransport createTransport(String addr, int port) + throws IOException; + + /** + * Create instance of {@link OtpTransport} + * + * @param addr + * peer address + * @param port + * port number + * @return new socket object + * @throws IOException + */ + public abstract OtpTransport createTransport(InetAddress addr, int port) + throws IOException; + + /** + * Create instance of {@link OtpServerTransport} + * + * @param port + * port number to listen on + * @return new socket object + * @throws IOException + */ + public OtpServerTransport createServerTransport(int port) + throws IOException; +} 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 62fa7f990e..a0f19bc1aa 100644 --- a/lib/jinterface/java_src/com/ericsson/otp/erlang/java_files +++ b/lib/jinterface/java_src/com/ericsson/otp/erlang/java_files @@ -53,7 +53,13 @@ COMM = \ OtpOutputStream \ OtpPeer \ OtpSelf \ - OtpServer + OtpServer \ + OtpServerSocketTransport \ + OtpServerTransport \ + OtpSocketTransport \ + OtpSocketTransportFactory \ + OtpTransport \ + OtpTransportFactory ERL = \ OtpErlangAtom \ diff --git a/lib/jinterface/test/jinterface_SUITE.erl b/lib/jinterface/test/jinterface_SUITE.erl index 00abc97ff5..3743e763c5 100644 --- a/lib/jinterface/test/jinterface_SUITE.erl +++ b/lib/jinterface/test/jinterface_SUITE.erl @@ -22,7 +22,8 @@ init_per_suite/1, end_per_suite/1, init_per_testcase/2, end_per_testcase/2]). --export([nodename/1, register_and_whereis/1, get_names/1, boolean_atom/1, +-export([transport_factory/1, + nodename/1, register_and_whereis/1, get_names/1, boolean_atom/1, node_ping/1, mbox_ping/1, java_erlang_send_receive/1, java_internal_send_receive_same_node/1, @@ -103,6 +104,7 @@ end_per_group(_GroupName, Config) -> fundamental() -> [ + transport_factory, % TransportFactoryTest.java nodename, % Nodename.java register_and_whereis, % RegisterAndWhereis.java get_names, % GetNames.java @@ -200,6 +202,16 @@ end_per_testcase(_Case,Config) -> %%%----------------------------------------------------------------- %%% TEST CASES +%%%----------------------------------------------------------------- +transport_factory(doc) -> + ["TransportFactoryTest.java: Test custom OTP Transport Factory"]; +transport_factory(suite) -> + []; +transport_factory(Config) when is_list(Config) -> + ok = jitu:java(?config(java, Config), + ?config(data_dir, Config), + "TransportFactoryTest"). + %%%----------------------------------------------------------------- nodename(doc) -> ["Nodename.java: " diff --git a/lib/jinterface/test/jinterface_SUITE_data/Makefile.src b/lib/jinterface/test/jinterface_SUITE_data/Makefile.src index cd68f1ead5..eca043913e 100644 --- a/lib/jinterface/test/jinterface_SUITE_data/Makefile.src +++ b/lib/jinterface/test/jinterface_SUITE_data/Makefile.src @@ -38,6 +38,7 @@ JINTERFACE_CLASSPATH = @jinterface_classpath@ CLASSPATH = .@PS@$(JINTERFACE_CLASSPATH)@PS@ JAVA_FILES = \ + TransportFactoryTest.java \ Nodename.java \ RegisterAndWhereis.java \ GetNames.java \ diff --git a/lib/jinterface/test/jinterface_SUITE_data/TransportFactoryTest.java b/lib/jinterface/test/jinterface_SUITE_data/TransportFactoryTest.java new file mode 100644 index 0000000000..367e28a512 --- /dev/null +++ b/lib/jinterface/test/jinterface_SUITE_data/TransportFactoryTest.java @@ -0,0 +1,90 @@ +/* + * %CopyrightBegin% + * + * Copyright Ericsson AB 2015. All Rights Reserved. + * + * The contents of this file are subject to the Erlang Public License, + * Version 1.1, (the "License"); you may not use this file except in + * compliance with the License. You should have received a copy of the + * Erlang Public License along with this software. If not, it can be + * retrieved online at http://www.erlang.org/. + * + * Software distributed under the License is distributed on an "AS IS" + * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See + * the License for the specific language governing rights and limitations + * under the License. + * + * %CopyrightEnd% + */ + +import java.io.IOException; +import java.net.InetAddress; + +import com.ericsson.otp.erlang.OtpSelf; +import com.ericsson.otp.erlang.OtpServerTransport; +import com.ericsson.otp.erlang.OtpSocketTransportFactory; +import com.ericsson.otp.erlang.OtpTransport; +import com.ericsson.otp.erlang.OtpTransportFactory; + +/** + * @author Dmitriy Kargapolov + */ +public class TransportFactoryTest { + + /** + * example of custom transport factory wrapping default one + */ + public static class TransportFactory implements OtpTransportFactory { + + OtpSocketTransportFactory tf = new OtpSocketTransportFactory(); + + public OtpTransport createTransport(final String addr, final int port) + throws IOException { + clientOk = true; + System.out.println("creating transport to " + addr + ", " + port); + return tf.createTransport(addr, port); + } + + public OtpTransport createTransport(final InetAddress addr, + final int port) throws IOException { + clientOk = true; + System.out.println("creating transport to " + addr + ", " + port); + return tf.createTransport(addr, port); + } + + public OtpServerTransport createServerTransport(final int port) + throws IOException { + serverOk = true; + System.out.println("creating server transport to " + port); + return tf.createServerTransport(port); + } + + } + + static boolean serverOk = false; + static boolean clientOk = false; + + public static void main(final String[] args) throws IOException { + + // check server transport + final OtpSelf self = new OtpSelf("local", new TransportFactory()); + if (!serverOk) { + fail("custom server transport was not created"); + } + System.out.println("accepting connections on " + self.port()); + + // check client transport + try { + self.publishPort(); + } catch (final Exception e) { + } + if (!clientOk) { + fail("custom client transport was not created"); + } + } + + private static void fail(final String string) { + System.err.println(string); + System.exit(1); + } +} -- cgit v1.2.3