/*
* JBoss, Home of Professional Open Source.
* Copyright 2006, Red Hat Middleware LLC, 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.ha.jndi;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.UndeclaredThrowableException;
import java.net.DatagramPacket;
import java.net.InetAddress;
import java.net.MulticastSocket;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.UnknownHostException;
import java.rmi.MarshalledObject;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import javax.management.ObjectInstance;
import javax.management.ObjectName;
import javax.management.Query;
import javax.management.QueryExp;
import javax.net.ServerSocketFactory;
import org.jboss.ha.framework.interfaces.HAPartition;
import org.jboss.ha.framework.server.ClusterPartition;
import org.jboss.ha.framework.server.ClusterPartitionMBean;
import org.jboss.invocation.Invocation;
import org.jboss.invocation.MarshalledInvocation;
import org.jboss.logging.Logger;
import org.jboss.mx.util.MBeanProxyExt;
import org.jboss.system.ServiceMBeanSupport;
import org.jboss.system.server.ServerConfigUtil;
import org.jboss.util.threadpool.BasicThreadPool;
import org.jboss.util.threadpool.BasicThreadPoolMBean;
import org.jboss.util.threadpool.ThreadPool;
import org.jnp.interfaces.Naming;
import org.jnp.interfaces.NamingContext;
/**
* Management Bean for the protocol independent HA-JNDI service. This allows the
* naming service transport layer to be provided by a detached invoker service
* like JRMPInvokerHA + ProxyFactoryHA.
* @author <a href="mailto:bill@burkecentral.com">Bill Burke</a>
* @author <a href="mailto:sacha.labourey@cogito-info.ch">Sacha Labourey</a>
* @author Scott.Stark@jboss.org
* @version $Revision: 57188 $
*/
public class DetachedHANamingService
extends ServiceMBeanSupport
implements DetachedHANamingServiceMBean
{
// Constants -----------------------------------------------------
// Attributes ----------------------------------------------------
/**
* The jnp server socket through which the HAJNDI stub is vended
*/
protected ServerSocket bootstrapSocket;
/**
* The Naming interface server implementation
*/
protected HAJNDI theServer;
/**
* The mapping from the long method hash to the Naming Method
*/
protected Map marshalledInvocationMapping;
/**
* The protocol stub returned to clients by the bootstrap lookup
*/
protected Naming stub;
/**
* The HAPartition used for the state transfer service
*/
protected HAPartition partition;
/**
* The ClusterPartition with which we are associated.
*/
protected ClusterPartitionMBean clusterPartition;
/**
* The partition name used to lookup the HAPartition binding
*/
protected String partitionName = ServerConfigUtil.getDefaultPartitionName();
/**
* The proxy factory service that generates the Naming stub
*/
private ObjectName proxyFactory;
/**
* The interface to bind to. This is useful for multi-homed hosts that want
* control over which interfaces accept connections.
*/
protected InetAddress bindAddress;
/**
* The bootstrapSocket listen queue depth
*/
protected int backlog = 50;
/**
* The jnp protocol listening port. The default is 1100, the same as the RMI
* registry default port.
*/
protected int port = 1100;
/**
* The autodiscovery multicast group
*/
protected String adGroupAddress = NamingContext.DEFAULT_DISCOVERY_GROUP_ADDRESS;
/**
* The autodiscovery port
*/
protected int adGroupPort = NamingContext.DEFAULT_DISCOVERY_GROUP_PORT;
/**
* The interface to bind the Multicast socket for autodiscovery to
*/
protected InetAddress discoveryBindAddress;
/** The runable task for discovery request packets */
protected AutomaticDiscovery autoDiscovery = null;
/** A flag indicating if autodiscovery should be disabled */
protected boolean discoveryDisabled = false;
/** The autodiscovery Multicast reply TTL */
protected int autoDiscoveryTTL = 16;
/**
* An optional custom server socket factory for the bootstrap lookup
*/
protected ServerSocketFactory jnpServerSocketFactory;
/**
* The class name of the optional custom JNP server socket factory
*/
protected String jnpServerSocketFactoryName;
/**
* The thread pool used to handle jnp stub lookup requests
*/
protected ThreadPool lookupPool;
// Public --------------------------------------------------------
public DetachedHANamingService()
{
// for JMX
}
/**
* Expose the Naming service interface mapping as a read-only attribute
* @return A Map<Long hash, Method> of the Naming interface
* @jmx:managed-attribute
*/
public Map getMethodMap()
{
return marshalledInvocationMapping;
}
public ClusterPartitionMBean getClusterPartition()
{
return clusterPartition;
}
public void setClusterPartition(ClusterPartitionMBean clusterPartition)
{
this.clusterPartition = clusterPartition;
}
public String getPartitionName()
{
return partitionName;
}
public void setPartitionName(final String partitionName)
{
this.partitionName = partitionName;
}
public ObjectName getProxyFactoryObjectName()
{
return proxyFactory;
}
public void setProxyFactoryObjectName(ObjectName proxyFactory)
{
this.proxyFactory = proxyFactory;
}
public void setPort(int p)
{
port = p;
}
public int getPort()
{
return port;
}
public String getBindAddress()
{
String address = null;
if (bindAddress != null)
address = bindAddress.getHostAddress();
return address;
}
public void setBindAddress(String host) throws java.net.UnknownHostException
{
bindAddress = InetAddress.getByName(host);
}
public int getBacklog()
{
return backlog;
}
public void setBacklog(int backlog)
{
if (backlog <= 0)
backlog = 50;
this.backlog = backlog;
}
public void setDiscoveryDisabled(boolean disable)
{
this.discoveryDisabled = disable;
}
public boolean getDiscoveryDisabled()
{
return this.discoveryDisabled;
}
public String getAutoDiscoveryAddress()
{
return this.adGroupAddress;
}
public void setAutoDiscoveryAddress(String adAddress)
{
this.adGroupAddress = adAddress;
}
public int getAutoDiscoveryGroup()
{
return this.adGroupPort;
}
public void setAutoDiscoveryGroup(int adGroup)
{
this.adGroupPort = adGroup;
}
public String getAutoDiscoveryBindAddress()
{
String address = null;
if (discoveryBindAddress != null)
address = discoveryBindAddress.getHostAddress();
return address;
}
public void setAutoDiscoveryBindAddress(String address)
throws UnknownHostException
{
discoveryBindAddress = InetAddress.getByName(address);
}
public int getAutoDiscoveryTTL()
{
return autoDiscoveryTTL;
}
public void setAutoDiscoveryTTL(int ttl)
{
autoDiscoveryTTL = ttl;
}
public void setJNPServerSocketFactory(String factoryClassName)
throws ClassNotFoundException, InstantiationException, IllegalAccessException
{
this.jnpServerSocketFactoryName = factoryClassName;
ClassLoader loader = Thread.currentThread().getContextClassLoader();
Class clazz = loader.loadClass(jnpServerSocketFactoryName);
jnpServerSocketFactory = (ServerSocketFactory) clazz.newInstance();
}
public void setLookupPool(BasicThreadPoolMBean poolMBean)
{
lookupPool = poolMBean.getInstance();
}
public void startService(HAPartition haPartition)
throws Exception
{
this.partition = haPartition;
this.startService();
}
protected void createService()
throws Exception
{
boolean debug = log.isDebugEnabled();
if (this.clusterPartition == null)
{
partition = findHAPartitionWithName(partitionName);
}
else
{
partition = clusterPartition.getHAPartition();
partitionName = partition.getPartitionName();
}
if (partition == null)
throw new IllegalStateException("Cannot find partition '" + partitionName + "'");
if (debug)
log.debug("Initializing HAJNDI server on partition: " + partitionName);
// Start HAJNDI service
theServer = new HAJNDI(partition);
log.debug("initialize HAJNDI");
theServer.init();
// Build the Naming interface method map
HashMap tmpMap = new HashMap(13);
Method[] methods = Naming.class.getMethods();
for (int m = 0; m < methods.length; m++)
{
Method method = methods[m];
Long hash = new Long(MarshalledInvocation.calculateHash(method));
tmpMap.put(hash, method);
}
marshalledInvocationMapping = Collections.unmodifiableMap(tmpMap);
// share instance for in-vm discovery
NamingContext.setHANamingServerForPartition(partitionName, theServer);
}
protected void startService()
throws Exception
{
log.debug("Obtaining the transport proxy");
stub = this.getNamingProxy();
this.theServer.setHAStub(stub);
if (port >= 0)
{
log.debug("Starting HAJNDI bootstrap listener");
initBootstrapListener();
}
// Automatic Discovery for unconfigured clients
if (adGroupAddress != null && discoveryDisabled == false)
{
try
{
autoDiscovery = new AutomaticDiscovery();
autoDiscovery.start();
lookupPool.run(autoDiscovery);
}
catch (Exception e)
{
log.warn("Failed to start AutomaticDiscovery", e);
}
}
}
protected void stopService() throws Exception
{
// un-share instance for in-vm discovery
NamingContext.removeHANamingServerForPartition(partitionName);
// Stop listener
ServerSocket s = bootstrapSocket;
bootstrapSocket = null;
if (s != null)
{
log.debug("Closing the bootstrap listener");
s.close();
}
// Stop HAJNDI service
log.debug("Stopping the HAJNDI service");
theServer.stop();
log.debug("Stopping AutomaticDiscovery");
if (autoDiscovery != null && discoveryDisabled == false)
autoDiscovery.stop();
}
protected void destroyService() throws Exception
{
log.debug("Destroying the HAJNDI service");
theServer.destroy();
}
/**
* Expose the Naming service via JMX to invokers.
* @param invocation A pointer to the invocation object
* @return Return value of method invocation.
* @throws Exception Failed to invoke method.
* @jmx:managed-operation
*/
public Object invoke(Invocation invocation) throws Exception
{
// Set the method hash to Method mapping
if (invocation instanceof MarshalledInvocation)
{
MarshalledInvocation mi = (MarshalledInvocation) invocation;
mi.setMethodMap(marshalledInvocationMapping);
}
// Invoke the Naming method via reflection
Method method = invocation.getMethod();
Object[] args = invocation.getArguments();
Object value = null;
try
{
value = method.invoke(theServer, args);
}
catch (InvocationTargetException e)
{
Throwable t = e.getTargetException();
if (t instanceof Exception)
throw (Exception) t;
else
throw new UndeclaredThrowableException(t, method.toString());
}
return value;
}
/**
* Bring up the bootstrap lookup port for obtaining the naming service proxy
*/
protected void initBootstrapListener()
{
// Start listener
try
{
// Get the default ServerSocketFactory is one was not specified
if (jnpServerSocketFactory == null)
jnpServerSocketFactory = ServerSocketFactory.getDefault();
bootstrapSocket = jnpServerSocketFactory.createServerSocket(port, backlog, bindAddress);
// If an anonymous port was specified get the actual port used
if (port == 0)
port = bootstrapSocket.getLocalPort();
String msg = "Started ha-jndi bootstrap jnpPort=" + port
+ ", backlog=" + backlog + ", bindAddress=" + bindAddress;
log.info(msg);
}
catch (IOException e)
{
log.error("Could not start on port " + port, e);
}
if (lookupPool == null)
lookupPool = new BasicThreadPool("HANamingBootstrap Pool");
AcceptHandler handler = new AcceptHandler();
lookupPool.run(handler);
}
// Protected -----------------------------------------------------
protected HAPartition findHAPartitionWithName(String name) throws Exception
{
HAPartition result = null;
QueryExp exp = Query.and(Query.eq(Query.classattr(),
Query.value(ClusterPartition.class.getName())),
Query.match(Query.attr("PartitionName"),
Query.value(name)));
Set mbeans = this.getServer().queryMBeans(null, exp);
if (mbeans != null && mbeans.size() > 0)
{
ObjectInstance inst = (ObjectInstance) (mbeans.iterator().next());
ClusterPartitionMBean cp = (ClusterPartitionMBean) MBeanProxyExt.create(ClusterPartitionMBean.class,
inst.getObjectName(),
this.getServer());
result = cp.getHAPartition();
}
return result;
}
/**
* Get the Naming proxy for the transport. This version looks up the
* proxyFactory service Proxy attribute. Subclasses can override this to set
* the proxy another way.
* @return The Naming proxy for the protocol used with the HAJNDI service
*/
protected Naming getNamingProxy() throws Exception
{
Naming proxy = (Naming) server.getAttribute(proxyFactory, "Proxy");
return proxy;
}
// Private -------------------------------------------------------
private class AutomaticDiscovery
implements Runnable
{
protected Logger log = Logger.getLogger(AutomaticDiscovery.class);
/** The socket for auto discovery requests */
protected MulticastSocket socket = null;
/** The ha-jndi addres + ':' + port string */
protected byte[] ipAddress = null;
/** The multicast group address */
protected InetAddress group = null;
protected boolean stopping = false;
// Thread that is executing the run() method
protected Thread receiverThread = null;
protected boolean receiverStopped = true;
public AutomaticDiscovery() throws Exception
{
}
public void start() throws Exception
{
stopping = false;
// Use the jndi bind address if there is no discovery address
if (discoveryBindAddress == null)
discoveryBindAddress = bindAddress;
socket = new MulticastSocket(adGroupPort);
// If there is a bind address valid, set the socket interface to it
if (discoveryBindAddress != null && discoveryBindAddress.isAnyLocalAddress() == false)
{
socket.setInterface(discoveryBindAddress);
}
socket.setTimeToLive(autoDiscoveryTTL);
group = InetAddress.getByName(adGroupAddress);
socket.joinGroup(group);
String address = getBindAddress();
/* An INADDR_ANY (0.0.0.0 || null) address is useless as the value
sent to a remote client so check for this and use the local host
address instead.
*/
if (address == null || address.equals("0.0.0.0"))
{
address = InetAddress.getLocalHost().getHostAddress();
}
ipAddress = (address + ":" + port).getBytes();
log.info("Listening on " + socket.getInterface() + ":" + socket.getLocalPort()
+ ", group=" + adGroupAddress
+ ", HA-JNDI address=" + new String(ipAddress));
}
public void stop()
{
try
{
stopping = true;
// JBAS-2834 -- try to stop the receiverThread
if (receiverThread != null
&& receiverThread != Thread.currentThread()
&& receiverThread.isInterrupted() == false)
{
// Give it a moment to die on its own (unlikely)
receiverThread.join(5);
if (!receiverStopped)
receiverThread.interrupt(); // kill it
}
socket.leaveGroup(group);
socket.close();
}
catch (Exception ex)
{
log.error("Stopping AutomaticDiscovery failed", ex);
}
}
public void run()
{
boolean trace = log.isTraceEnabled();
log.debug("Discovery request thread begin");
// JBAS-2834 Cache a reference to this thread so stop()
// can interrupt it if necessary
receiverThread = Thread.currentThread();
receiverStopped = false;
// Wait for a datagram
while (true)
{
// Stopped by normal means
if (stopping)
break;
try
{
if (trace)
log.trace("HA-JNDI AutomaticDiscovery waiting for queries...");
byte[] buf = new byte[256];
DatagramPacket packet = new DatagramPacket(buf, buf.length);
socket.receive(packet);
if (trace)
log.trace("HA-JNDI AutomaticDiscovery Packet received.");
// Queue the response to the thread pool
DiscoveryRequestHandler handler = new DiscoveryRequestHandler(log,
packet, socket, ipAddress);
lookupPool.run(handler);
if (trace)
log.trace("Queued DiscoveryRequestHandler");
}
catch (Throwable t)
{
if (stopping == false)
log.warn("Ignored error while processing HAJNDI discovery request:", t);
}
}
receiverStopped = true;
log.debug("Discovery request thread end");
}
}
/**
* The class used as the runnable for writing the bootstrap stub
*/
private class DiscoveryRequestHandler implements Runnable
{
private Logger log;
private MulticastSocket socket;
private DatagramPacket packet;
private byte[] ipAddress;
DiscoveryRequestHandler(Logger log, DatagramPacket packet,
MulticastSocket socket, byte[] ipAddress)
{
this.log = log;
this.packet = packet;
this.socket = socket;
this.ipAddress = ipAddress;
}
public void run()
{
boolean trace = log.isTraceEnabled();
if( trace )
log.trace("DiscoveryRequestHandler begin");
// Return the naming server IP address and port to the client
try
{
// See if the discovery is restricted to a particular parition
String requestData = new String(packet.getData()).trim();
if( trace )
log.trace("RequestData: "+requestData);
int colon = requestData.indexOf(':');
if (colon > 0)
{
// Check the partition name
String name = requestData.substring(colon + 1);
if (name.equals(partitionName) == false)
{
log.debug("Ignoring discovery request for partition: " + name);
if( trace )
log.trace("DiscoveryRequestHandler end");
return;
}
}
DatagramPacket p = new DatagramPacket(ipAddress, ipAddress.length,
packet.getAddress(), packet.getPort());
if (trace)
log.trace("Sending AutomaticDiscovery answer: " + new String(ipAddress));
socket.send(p);
if (trace)
log.trace("AutomaticDiscovery answer sent.");
}
catch (IOException ex)
{
log.error("Error writing response", ex);
}
if( trace )
log.trace("DiscoveryRequestHandler end");
}
}
/**
* The class used as the runnable for the bootstrap lookup thread pool.
*/
private class AcceptHandler implements Runnable
{
public void run()
{
boolean trace = log.isTraceEnabled();
while (bootstrapSocket != null)
{
Socket socket = null;
// Accept a connection
try
{
socket = bootstrapSocket.accept();
if( trace )
log.trace("Accepted bootstrap client: "+socket);
BootstrapRequestHandler handler = new BootstrapRequestHandler(socket);
lookupPool.run(handler);
}
catch (IOException e)
{
// Stopped by normal means
if (bootstrapSocket == null)
return;
log.error("Naming accept handler stopping", e);
}
catch(Throwable e)
{
log.error("Unexpected exception during accept", e);
}
}
}
}
/**
* The class used as the runnable for writing the bootstrap stub
*/
private class BootstrapRequestHandler implements Runnable
{
private Socket socket;
BootstrapRequestHandler(Socket socket)
{
this.socket = socket;
}
public void run()
{
// Return the naming server stub
try
{
OutputStream os = socket.getOutputStream();
ObjectOutputStream out = new ObjectOutputStream(os);
MarshalledObject replyStub = new MarshalledObject(stub);
out.writeObject(replyStub);
out.close();
}
catch (IOException ex)
{
log.debug("Error writing response to " + socket, ex);
}
finally
{
try
{
socket.close();
}
catch (IOException e)
{
}
}
}
}
}
|
