/*
* %CopyrightBegin%
*
* Copyright Ericsson AB 2000-2012. 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.lang.ref.WeakReference;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Collection;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Iterator;
/**
* <p>
* Represents a local OTP node. This class is used when you do not wish to
* manage connections yourself - outgoing connections are established as needed,
* and incoming connections accepted automatically. This class supports the use
* of a mailbox API for communication, while management of the underlying
* communication mechanism is automatic and hidden from the application
* programmer.
* </p>
*
* <p>
* Once an instance of this class has been created, obtain one or more mailboxes
* in order to send or receive messages. The first message sent to a given node
* will cause a connection to be set up to that node. Any messages received will
* be delivered to the appropriate mailboxes.
* </p>
*
* <p>
* To shut down the node, call {@link #close close()}. This will prevent the
* node from accepting additional connections and it will cause all existing
* connections to be closed. Any unread messages in existing mailboxes can still
* be read, however no new messages will be delivered to the mailboxes.
* </p>
*
* <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.
* </p>
*/
public class OtpNode extends OtpLocalNode {
private boolean initDone = false;
// thread to manage incoming connections
private Acceptor acceptor = null;
// keep track of all connections
Hashtable<String, OtpCookedConnection> connections = null;
// keep track of all mailboxes
Mailboxes mboxes = null;
// handle status changes
OtpNodeStatus handler;
// flags
private int connFlags = 0;
/**
* <p>
* 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".
* </p>
*
* <p>
* If the file does not exist, an empty string is used. This method makes no
* attempt to create the file.
* </p>
*
* @param node
* the name of this node.
*
* @exception IOException
* if communication could not be initialized.
*
*/
public OtpNode(final String node) throws IOException {
this(node, defaultCookie, 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.
*
* @exception IOException
* if communication could not be initialized.
*
*/
public OtpNode(final String node, final String cookie) throws IOException {
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 port
* the port number you wish to use for incoming connections.
* Specifying 0 lets the system choose an available port.
*
* @exception IOException
* if communication could not be initialized.
*
*/
public OtpNode(final String node, final String cookie, final int port)
throws IOException {
super(node, cookie);
init(port);
}
private synchronized void init(final int aport) throws IOException {
if (!initDone) {
connections = new Hashtable<String, OtpCookedConnection>(17,
(float) 0.95);
mboxes = new Mailboxes();
acceptor = new Acceptor(aport);
initDone = true;
}
}
/**
* Close the node. Unpublish the node from Epmd (preventing new connections)
* and close all existing connections.
*/
public synchronized void close() {
acceptor.quit();
OtpCookedConnection conn;
final Collection<OtpCookedConnection> coll = connections.values();
final Iterator<OtpCookedConnection> it = coll.iterator();
mboxes.clear();
while (it.hasNext()) {
conn = it.next();
it.remove();
conn.close();
}
initDone = false;
}
@Override
protected void finalize() {
close();
}
/**
* Create an unnamed {@link OtpMbox mailbox} that can be used to send and
* receive messages with other, similar mailboxes and with Erlang processes.
* Messages can be sent to this mailbox by using its associated
* {@link OtpMbox#self() pid}.
*
* @return a mailbox.
*/
public OtpMbox createMbox() {
return mboxes.create();
}
/**
* Close the specified mailbox with reason 'normal'.
*
* @param mbox
* the mailbox to close.
*
* <p>
* After this operation, the mailbox will no longer be able to
* receive messages. Any delivered but as yet unretrieved
* messages can still be retrieved however.
* </p>
*
* <p>
* If there are links from the mailbox to other
* {@link OtpErlangPid pids}, they will be broken when this
* method is called and exit signals with reason 'normal' will be
* sent.
* </p>
*
*/
public void closeMbox(final OtpMbox mbox) {
closeMbox(mbox, new OtpErlangAtom("normal"));
}
/**
* Close the specified mailbox with the given reason.
*
* @param mbox
* the mailbox to close.
* @param reason
* an Erlang term describing the reason for the termination.
*
* <p>
* After this operation, the mailbox will no longer be able to
* receive messages. Any delivered but as yet unretrieved
* messages can still be retrieved however.
* </p>
*
* <p>
* If there are links from the mailbox to other
* {@link OtpErlangPid pids}, they will be broken when this
* method is called and exit signals with the given reason will
* be sent.
* </p>
*
*/
public void closeMbox(final OtpMbox mbox, final OtpErlangObject reason) {
if (mbox != null) {
mboxes.remove(mbox);
mbox.name = null;
mbox.breakLinks(reason);
}
}
/**
* Create an named mailbox that can be used to send and receive messages
* with other, similar mailboxes and with Erlang processes. Messages can be
* sent to this mailbox by using its registered name or the associated
* {@link OtpMbox#self() pid}.
*
* @param name
* a name to register for this mailbox. The name must be unique
* within this OtpNode.
*
* @return a mailbox, or null if the name was already in use.
*
*/
public OtpMbox createMbox(final String name) {
return mboxes.create(name);
}
/**
* <p>
* Register or remove a name for the given mailbox. Registering a name for a
* mailbox enables others to send messages without knowing the
* {@link OtpErlangPid pid} of the mailbox. A mailbox can have at most one
* name; if the mailbox already had a name, calling this method will
* supercede that name.
* </p>
*
* @param name
* the name to register for the mailbox. Specify null to
* unregister the existing name from this mailbox.
*
* @param mbox
* the mailbox to associate with the name.
*
* @return true if the name was available, or false otherwise.
*/
public boolean registerName(final String name, final OtpMbox mbox) {
return mboxes.register(name, mbox);
}
/**
* Get a list of all known registered names on this node.
*
* @return an array of Strings, containins all known registered names on
* this node.
*/
public String[] getNames() {
return mboxes.names();
}
/**
* Determine the {@link OtpErlangPid pid} corresponding to a registered name
* on this node.
*
* @return the {@link OtpErlangPid pid} corresponding to the registered
* name, or null if the name is not known on this node.
*/
public OtpErlangPid whereis(final String name) {
final OtpMbox m = mboxes.get(name);
if (m != null) {
return m.self();
}
return null;
}
/**
* Register interest in certain system events. The {@link OtpNodeStatus
* OtpNodeStatus} handler object contains callback methods, that will be
* called when certain events occur.
*
* @param ahandler
* the callback object to register. To clear the handler, specify
* null as the handler to use.
*
*/
public synchronized void registerStatusHandler(final OtpNodeStatus ahandler) {
this.handler = ahandler;
}
/**
* <p>
* Determine if another node is alive. This method has the side effect of
* setting up a connection to the remote node (if possible). Only a single
* outgoing message is sent; the timeout is how long to wait for a response.
* </p>
*
* <p>
* Only a single attempt is made to connect to the remote node, so for
* example it is not possible to specify an extremely long timeout and
* expect to be notified when the node eventually comes up. If you wish to
* wait for a remote node to be started, the following construction may be
* useful:
* </p>
*
* <pre>
* // ping every 2 seconds until positive response
* while (!me.ping(him, 2000))
* ;
* </pre>
*
* @param anode
* the name of the node to ping.
*
* @param timeout
* the time, in milliseconds, to wait for response before
* returning false.
*
* @return true if the node was alive and the correct ping response was
* returned. false if the correct response was not returned on time.
*/
/*
* internal info about the message formats...
*
* the request: -> REG_SEND {6,#Pid<[email protected]>,'',net_kernel}
* {'$gen_call',{#Pid<[email protected]>,#Ref<[email protected]>},{is_auth,bingo@aule}}
*
* the reply: <- SEND {2,'',#Pid<[email protected]>} {#Ref<[email protected]>,yes}
*/
public boolean ping(final String anode, final long timeout) {
if (anode.equals(this.node)) {
return true;
} else if (anode.indexOf('@', 0) < 0
&& anode.equals(this.node
.substring(0, this.node.indexOf('@', 0)))) {
return true;
}
// other node
OtpMbox mbox = null;
try {
mbox = createMbox();
mbox.send("net_kernel", anode, getPingTuple(mbox));
final OtpErlangObject reply = mbox.receive(timeout);
final OtpErlangTuple t = (OtpErlangTuple) reply;
final OtpErlangAtom a = (OtpErlangAtom) t.elementAt(1);
return "yes".equals(a.atomValue());
} catch (final Exception e) {
} finally {
closeMbox(mbox);
}
return false;
}
/* create the outgoing ping message */
private OtpErlangTuple getPingTuple(final OtpMbox mbox) {
final OtpErlangObject[] ping = new OtpErlangObject[3];
final OtpErlangObject[] pid = new OtpErlangObject[2];
final OtpErlangObject[] anode = new OtpErlangObject[2];
pid[0] = mbox.self();
pid[1] = createRef();
anode[0] = new OtpErlangAtom("is_auth");
anode[1] = new OtpErlangAtom(node());
ping[0] = new OtpErlangAtom("$gen_call");
ping[1] = new OtpErlangTuple(pid);
ping[2] = new OtpErlangTuple(anode);
return new OtpErlangTuple(ping);
}
/*
* this method simulates net_kernel only for the purpose of replying to
* pings.
*/
private boolean netKernel(final OtpMsg m) {
OtpMbox mbox = null;
try {
final OtpErlangTuple t = (OtpErlangTuple) m.getMsg();
final OtpErlangTuple req = (OtpErlangTuple) t.elementAt(1); // actual
// request
final OtpErlangPid pid = (OtpErlangPid) req.elementAt(0); // originating
// pid
final OtpErlangObject[] pong = new OtpErlangObject[2];
pong[0] = req.elementAt(1); // his #Ref
pong[1] = new OtpErlangAtom("yes");
mbox = createMbox();
mbox.send(pid, new OtpErlangTuple(pong));
return true;
} catch (final Exception e) {
} finally {
closeMbox(mbox);
}
return false;
}
/*
* OtpCookedConnection delivers messages here return true if message was
* delivered successfully, or false otherwise.
*/
boolean deliver(final OtpMsg m) {
OtpMbox mbox = null;
try {
final int t = m.type();
if (t == OtpMsg.regSendTag) {
final String name = m.getRecipientName();
/* special case for netKernel requests */
if (name.equals("net_kernel")) {
return netKernel(m);
}
mbox = mboxes.get(name);
} else {
mbox = mboxes.get(m.getRecipientPid());
}
if (mbox == null) {
return false;
}
mbox.deliver(m);
} catch (final Exception e) {
return false;
}
return true;
}
/*
* OtpCookedConnection delivers errors here, we send them on to the handler
* specified by the application
*/
void deliverError(final OtpCookedConnection conn, final Exception e) {
removeConnection(conn);
remoteStatus(conn.name, false, e);
}
/*
* find or create a connection to the given node
*/
OtpCookedConnection getConnection(final String anode) {
OtpPeer peer = null;
OtpCookedConnection conn = null;
synchronized (connections) {
// first just try looking up the name as-is
conn = connections.get(anode);
if (conn == null) {
// in case node had no '@' add localhost info and try again
peer = new OtpPeer(anode);
conn = connections.get(peer.node());
if (conn == null) {
try {
conn = new OtpCookedConnection(this, peer);
conn.setFlags(connFlags);
addConnection(conn);
} catch (final Exception e) {
/* false = outgoing */
connAttempt(peer.node(), false, e);
}
}
}
return conn;
}
}
void addConnection(final OtpCookedConnection conn) {
if (conn != null && conn.name != null) {
connections.put(conn.name, conn);
remoteStatus(conn.name, true, null);
}
}
private void removeConnection(final OtpCookedConnection conn) {
if (conn != null && conn.name != null) {
connections.remove(conn.name);
}
}
/* use these wrappers to call handler functions */
private synchronized void remoteStatus(final String anode, final boolean up,
final Object info) {
if (handler == null) {
return;
}
try {
handler.remoteStatus(anode, up, info);
} catch (final Exception e) {
}
}
synchronized void localStatus(final String anode, final boolean up,
final Object info) {
if (handler == null) {
return;
}
try {
handler.localStatus(anode, up, info);
} catch (final Exception e) {
}
}
synchronized void connAttempt(final String anode, final boolean incoming,
final Object info) {
if (handler == null) {
return;
}
try {
handler.connAttempt(anode, incoming, info);
} catch (final Exception e) {
}
}
/*
* this class used to wrap the mailbox hashtables so we can use weak
* references
*/
public class Mailboxes {
// mbox pids here
private Hashtable<OtpErlangPid, WeakReference<OtpMbox>> byPid = null;
// mbox names here
private Hashtable<String, WeakReference<OtpMbox>> byName = null;
public Mailboxes() {
byPid = new Hashtable<OtpErlangPid, WeakReference<OtpMbox>>(17,
(float) 0.95);
byName = new Hashtable<String, WeakReference<OtpMbox>>(17,
(float) 0.95);
}
public OtpMbox create(final String name) {
OtpMbox m = null;
synchronized (byName) {
if (get(name) != null) {
return null;
}
final OtpErlangPid pid = createPid();
m = new OtpMbox(OtpNode.this, pid, name);
byPid.put(pid, new WeakReference<OtpMbox>(m));
byName.put(name, new WeakReference<OtpMbox>(m));
}
return m;
}
public OtpMbox create() {
final OtpErlangPid pid = createPid();
final OtpMbox m = new OtpMbox(OtpNode.this, pid);
byPid.put(pid, new WeakReference<OtpMbox>(m));
return m;
}
public void clear() {
byPid.clear();
byName.clear();
}
public String[] names() {
String allnames[] = null;
synchronized (byName) {
final int n = byName.size();
final Enumeration<String> keys = byName.keys();
allnames = new String[n];
int i = 0;
while (keys.hasMoreElements()) {
allnames[i++] = keys.nextElement();
}
}
return allnames;
}
public boolean register(final String name, final OtpMbox mbox) {
if (name == null) {
if (mbox.name != null) {
byName.remove(mbox.name);
mbox.name = null;
}
} else {
synchronized (byName) {
if (get(name) != null) {
return false;
}
byName.put(name, new WeakReference<OtpMbox>(mbox));
mbox.name = name;
}
}
return true;
}
/*
* look up a mailbox based on its name. If the mailbox has gone out of
* scope we also remove the reference from the hashtable so we don't
* find it again.
*/
public OtpMbox get(final String name) {
final WeakReference<OtpMbox> wr = byName.get(name);
if (wr != null) {
final OtpMbox m = wr.get();
if (m != null) {
return m;
}
byName.remove(name);
}
return null;
}
/*
* look up a mailbox based on its pid. If the mailbox has gone out of
* scope we also remove the reference from the hashtable so we don't
* find it again.
*/
public OtpMbox get(final OtpErlangPid pid) {
final WeakReference<OtpMbox> wr = byPid.get(pid);
if (wr != null) {
final OtpMbox m = wr.get();
if (m != null) {
return m;
}
byPid.remove(pid);
}
return null;
}
public void remove(final OtpMbox mbox) {
byPid.remove(mbox.self);
if (mbox.name != null) {
byName.remove(mbox.name);
}
}
}
/*
* this thread simply listens for incoming connections
*/
public class Acceptor extends Thread {
private final ServerSocket sock;
private final int acceptorPort;
private volatile boolean done = false;
Acceptor(final int port) throws IOException {
sock = new ServerSocket(port);
this.acceptorPort = sock.getLocalPort();
OtpNode.this.port = this.acceptorPort;
setDaemon(true);
setName("acceptor");
publishPort();
start();
}
private boolean publishPort() throws IOException {
if (getEpmd() != null) {
return false; // already published
}
OtpEpmd.publishPort(OtpNode.this);
return true;
}
private void unPublishPort() {
// unregister with epmd
OtpEpmd.unPublishPort(OtpNode.this);
// close the local descriptor (if we have one)
closeSock(epmd);
epmd = null;
}
public void quit() {
unPublishPort();
done = true;
closeSock(sock);
localStatus(node, false, null);
}
private void closeSock(final ServerSocket s) {
try {
if (s != null) {
s.close();
}
} catch (final Exception e) {
}
}
private void closeSock(final Socket s) {
try {
if (s != null) {
s.close();
}
} catch (final Exception e) {
}
}
public int port() {
return acceptorPort;
}
@Override
public void run() {
Socket newsock = null;
OtpCookedConnection conn = null;
localStatus(node, true, null);
accept_loop: while (!done) {
conn = null;
try {
newsock = sock.accept();
} catch (final Exception e) {
// Problem in java1.2.2: accept throws SocketException
// when socket is closed. This will happen when
// acceptor.quit()
// is called. acceptor.quit() will call localStatus(...), so
// we have to check if that's where we come from.
if (!done) {
localStatus(node, false, e);
}
break accept_loop;
}
try {
synchronized (connections) {
conn = new OtpCookedConnection(OtpNode.this, newsock);
conn.setFlags(connFlags);
addConnection(conn);
}
} catch (final OtpAuthException e) {
if (conn != null && conn.name != null) {
connAttempt(conn.name, true, e);
} else {
connAttempt("unknown", true, e);
}
closeSock(newsock);
} catch (final IOException e) {
if (conn != null && conn.name != null) {
connAttempt(conn.name, true, e);
} else {
connAttempt("unknown", true, e);
}
closeSock(newsock);
} catch (final Exception e) {
closeSock(newsock);
closeSock(sock);
localStatus(node, false, e);
break accept_loop;
}
} // while
// if we have exited loop we must do this too
unPublishPort();
}
}
public void setFlags(final int flags) {
this.connFlags = flags;
}
}
