cluster.java

来自「RESIN 3.2 最新源码」· Java 代码 · 共 1,289 行 · 第 1/2 页

/*
 * Copyright (c) 1998-2008 Caucho Technology -- all rights reserved
 *
 * This file is part of Resin(R) Open Source
 *
 * Each copy or derived work must preserve the copyright notice and this
 * notice unmodified.
 *
 * Resin Open Source is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * Resin Open Source 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, or any warranty
 * of NON-INFRINGEMENT.  See the GNU General Public License for more
 * details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Resin Open Source; if not, write to the
 *
 *   Free Software Foundation, Inc.
 *   59 Temple Place, Suite 330
 *   Boston, MA 02111-1307  USA
 *
 * @author Scott Ferguson
 */

package com.caucho.server.cluster;

import com.caucho.config.program.ConfigProgram;
import com.caucho.config.program.ContainerProgram;
import com.caucho.config.ConfigException;
import com.caucho.config.SchemaBean;
import com.caucho.jmx.Jmx;
import com.caucho.lifecycle.StartLifecycleException;
import com.caucho.loader.DynamicClassLoader;
import com.caucho.loader.Environment;
import com.caucho.loader.EnvironmentBean;
import com.caucho.loader.EnvironmentClassLoader;
import com.caucho.loader.EnvironmentListener;
import com.caucho.loader.EnvironmentLocal;
import com.caucho.management.server.ClusterMXBean;
import com.caucho.management.server.EnvironmentMXBean;
import com.caucho.server.port.Port;
import com.caucho.server.resin.Resin;
import com.caucho.util.L10N;
import com.caucho.util.RandomUtil;
import com.caucho.vfs.Path;
import com.caucho.vfs.Vfs;
import com.caucho.webbeans.manager.*;

import javax.management.ObjectName;
import java.util.ArrayList;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.lang.reflect.*;

/**
 * Defines a set of clustered servers.
 */
public class Cluster
  implements EnvironmentListener, EnvironmentBean, SchemaBean
{
  private static final L10N L = new L10N(Cluster.class);
  private static final Logger log = Logger.getLogger(Cluster.class.getName());

  static protected final EnvironmentLocal<String> _serverIdLocal
    = new EnvironmentLocal<String>("caucho.server-id");

  static protected final EnvironmentLocal<Cluster> _clusterLocal
    = new EnvironmentLocal<Cluster>("caucho.cluster");

  private static final int DECODE[];
  
  private String _id = "";

  private String _serverId = "";

  private EnvironmentClassLoader _classLoader;
  
  private Resin _resin;

  private Path _rootDirectory;

  private ClusterAdmin _admin;

  private ObjectName _objectName;

  private ArrayList<ContainerProgram> _serverDefaultList
    = new ArrayList<ContainerProgram>();

  private ArrayList<Machine> _machineList
    = new ArrayList<Machine>();

  private ArrayList<ClusterServer> _serverList
    = new ArrayList<ClusterServer>();

  private ClusterServer[] _serverArray = new ClusterServer[0];

  private StoreManager _clusterStore;

  private boolean _isDynamicServerEnable = false;

  private ContainerProgram _serverProgram
    = new ContainerProgram();

  private Server _server;

  private long _version;

  private volatile boolean _isClosed;

  public Cluster(Resin resin)
  {
    this();

    _resin = resin;
  }
    
  public Cluster()
  {
    _classLoader = EnvironmentClassLoader.create("cluster:??");

    _clusterLocal.set(this, _classLoader);
  
    Environment.addEnvironmentListener(this, _classLoader);

    WebBeansContainer.create().addSingletonByName(new Var(), "cluster");

    _rootDirectory = Vfs.getPwd();
  }

  /**
   * Returns the currently active local cluster.
   */
  public static Cluster getLocal()
  {
    Cluster cluster = _clusterLocal.get();

    return cluster;
  }

  /**
   * Returns the currently active local cluster.
   */
  public static Cluster getCurrent()
  {
    Cluster cluster = _clusterLocal.get();

    return cluster;
  }

  /**
   * Returns the currently active local cluster.
   */
  public static Cluster getCluster(ClassLoader loader)
  {
    Cluster cluster = _clusterLocal.get(loader);

    return cluster;
  }

  /**
   * Sets the cluster id.
   */
  public void setId(String id)
  {
    if (id == null)
      throw new NullPointerException();
    
    _id = id;

    _classLoader.setId("cluster:" + _id);
  }

  /**
   * Gets the cluster id.
   */
  public String getId()
  {
    return _id;
  }

  /**
   * Returns the owning resin server.
   */
  public Resin getResin()
  {
    return _resin;
  }

  /**
   * Returns the environment class loader.
   */
  public ClassLoader getClassLoader()
  {
    return _classLoader;
  }

  /**
   * Returns the relax schema.
   */
  public String getSchema()
  {
    return "com/caucho/server/resin/cluster.rnc";
  }

  /**
   * Gets the root directory.
   */
  public Path getRootDirectory()
  {
    return _rootDirectory;
  }

  /**
   * Sets the root directory.
   */
  public void setRootDirectory(Path rootDirectory)
  {
    Vfs.setPwd(rootDirectory);
    
    _rootDirectory = rootDirectory;
  }

  /**
   * Enables dynamic servers
   */
  public void setDynamicServerEnable(boolean isEnable)
  {
    _isDynamicServerEnable = isEnable;
  }

  /**
   * Enables dynamic servers
   */
  public boolean isDynamicServerEnable()
  {
    return _isDynamicServerEnable;
  }

  /**
   * Returns the version
   */
  public long getVersion()
  {
    return _version;
  }

  /**
   * Returns the admin.
   */
  public ClusterMXBean getAdmin()
  {
    return _admin;
  }

  /**
   * Finds the first server with the given server-id.
   */
  public ClusterServer findServer(String id)
  {
    for (int i = _serverList.size() - 1; i >= 0; i--) {
      ClusterServer server = _serverList.get(i);

      if (server != null && server.getId().equals(id))
        return server;
    }

    return null;
  }

  /**
   * Finds the first server with the given server-id.
   */
  public ClusterServer findServer(String address, int port)
  {
    for (int i = _serverList.size() - 1; i >= 0; i--) {
      ClusterServer server = _serverList.get(i);

      if (server == null)
	continue;

      ClusterPort clusterPort = server.getClusterPort();
      
      if (clusterPort.getAddress().equals(address)
	  && clusterPort.getPort() == port) {
        return server;
      }
    }

    return null;
  }

  /**
   * Adds a new server to the cluster.
   */
  public void addServerDefault(ContainerProgram program)
  {
    _serverDefaultList.add(program);
  }

  /**
   * Adds a new server to the cluster.
   */
  public Machine createMachine()
  {
    Machine machine = new Machine(this);

    _machineList.add(machine);

    return machine;
  }

  /**
   * Adds a new server to the cluster.
   */
  public ClusterServer createServer()
  {
    Machine machine = createMachine();
  
    return machine.createServer();
  }

  ClusterServer createServer(ClusterServer server)
  {
    server.setIndex(_serverList.size());

    for (int i = 0; i < _serverDefaultList.size(); i++)
      _serverDefaultList.get(i).configure(server);

    return server;
  }

  /**
   * Adds a new server to the cluster.
   */
  public void addServer(ClusterServer server)
    throws ConfigException
  {
    ClusterServer oldServer = findServer(server.getId());

    if (oldServer != null)
      log.warning(L.l("duplicate <server> with id='{0}'",
                      server.getId()));

    _serverList.add(server);
    _serverArray = new ClusterServer[_serverList.size()];
    _serverList.toArray(_serverArray);

    ClusterServer selfServer = getSelfServer();

    if (selfServer == server) {
      WebBeansContainer webBeans = WebBeansContainer.create();
      
      webBeans.addSingletonByName(new ServerVar(server), "server");
    }
  }

  /**
   * Adds a new server to the cluster.
   */
  public void addDynamicServer(String serverId, String address, int port)
    throws ConfigException
  {
    if (! isDynamicServerEnable()) {
      log.warning(this + " forbidden dynamic-server add id=" + serverId
		  + " " + address + ":" + port);
      return;
    }
    
    try {
      ClusterServer oldServer = findServer(serverId);

      if (oldServer != null) {
	throw new ConfigException(L.l("duplicate server with id='{0}'",
				      serverId));
      }
      
      oldServer = findServer(address, port);

      if (oldServer != null) {
	throw new ConfigException(L.l("duplicate server with '{0}:{1}'",
				      address, port));
      }

      ClusterServer server = createServer();
      server.setId(serverId);

      server.setAddress(address);
      server.setPort(port);

      server.setDynamic(true);

      addServer(server);
      
      server.init();

      log.info(this + " add dynamic server " + server);
    } catch (Exception e) {
      throw ConfigException.create(e);
    }
  }

  /**
   * Adds a new server to the cluster.
   */
  public void addDynamicServer(ClusterServer server)
    throws ConfigException
  {
    try {
      synchronized (this) {
	for (ConfigProgram program : _serverDefaultList)
	  program.configure(server);
    
	server.init();
      
	// XXX: default config
	addServer(server);

	_version++;
      }
    } catch (Exception e) {
      throw ConfigException.create(e);
    }

    if (log.isLoggable(Level.FINE))
      log.fine(this + " add dynamic server " + server);
  }

  /**
   * Adds a new server to the cluster.
   */
  public void removeDynamicServer(ClusterServer server)
    throws ConfigException
  {
    if (! isDynamicServerEnable()) {
      log.warning(this + " forbidden dynamic-server remove " + server);
      return;
    }
    
    try {
      synchronized (this) {
	// XXX: default config
	
	_serverList.remove(server);
	_serverArray = new ClusterServer[_serverList.size()];
	_serverList.toArray(_serverArray);

	_version++;
      }
    } catch (Exception e) {
      throw ConfigException.create(e);
    }

    if (log.isLoggable(Level.FINE))
      log.fine(this + " remove dynamic server " + server);
  }

  /**
   * Adds a srun server.
   */
  public ServerConnector findConnector(String address, int port)
  {
    for (int i = _serverList.size() - 1; i >= 0; i--) {
      ClusterServer server = _serverList.get(i);
      ClusterPort clusterPort = server.getClusterPort();

      if (address.equals(clusterPort.getAddress())
	  && port == clusterPort.getPort()) {
	// XXX:
	//return server.getClient();
	return null;
      }
    }

    return null;
  }

  /**
   * Returns the cluster store.
   */
  public StoreManager getStore()
  {
    return _clusterStore;
  }

  /**
   * Sets the cluster store.
   */
  void setStore(StoreManager store)
  {
    _clusterStore = store;
  }

  public StoreManager createJdbcStore()
    throws ConfigException
  {
    if (getStore() != null)
      throw new ConfigException(L.l("multiple jdbc stores are not allowed in a cluster."));

    StoreManager store = null;

    try {
      Class cl = Class.forName("com.caucho.server.cluster.JdbcStoreManager");

      store = (StoreManager) cl.newInstance();

      store.setCluster(this);

      setStore(store);
    } catch (Throwable e) {
      log.log(Level.FINER, e.toString(), e);
    }

    if (store == null)
      throw new ConfigException(L.l("'jdbc' persistent sessions are available in Resin Professional.  See http://www.caucho.com for information and licensing."));

    return store;
  }

  public StoreManager createPrivateFileStore()
    throws ConfigException
  {
    StoreManager store = createFileStore();

    setStore(null);

    return store;
  }

  public StoreManager createFileStore()
    throws ConfigException
  {
    if (getStore() != null)
      throw new ConfigException(L.l("multiple file stores are not allowed in a cluster."));

    StoreManager store = null;

    try {
      Class cl = Class.forName("com.caucho.server.cluster.FileStoreManager");

      store = (StoreManager) cl.newInstance();

      store.setCluster(this);

      setStore(store);
    } catch (Throwable e) {
      log.log(Level.FINER, e.toString(), e);
    }

    if (store == null)
      throw new ConfigException(L.l("'file' persistent sessions are available in Resin Professional.  See http://www.caucho.com for information and licensing."));

    return store;
  }

  public StoreManager createClusterStore()
    throws ConfigException
  {
    if (getStore() != null)
      throw new ConfigException(L.l("multiple cluster stores are not allowed in a cluster."));

    StoreManager store = null;

    try {
      Class cl = Class.forName("com.caucho.server.cluster.ClusterStoreManager");

      store = (StoreManager) cl.newInstance();

      store.setCluster(this);

      setStore(store);
    } catch (Throwable e) {
      log.log(Level.FINER, e.toString(), e);
    }

    if (store == null)
      throw new ConfigException(L.l("'cluster' persistent sessions are available in Resin Professional.  See http://www.caucho.com for information and licensing."));

    return store;
  }

  /**
   * Adds a program.
   */
  public void addBuilderProgram(ConfigProgram program)
  {
    _serverProgram.addProgram(program);
  }

  /**
   * Initializes the cluster.
   */
  public void start()
    throws ConfigException
  {
    String serverId = _serverIdLocal.get();

    if (serverId == null)
      serverId = "";

    ClusterServer self = findServer(serverId);

    if (self != null) {
      _clusterLocal.set(this);
    }
    else if (_clusterLocal.get() == null && _serverList.size() == 0) {
      // if it's the empty cluster, add it
      _clusterLocal.set(this);
    }

    try {
      String name = _id;

      if (name == null)
        name = "";

      ObjectName objectName = Jmx.getObjectName("type=Cluster,name=" + name);

      _admin = new ClusterAdmin(this);

      Jmx.register(_admin, objectName);

      _objectName = objectName;
    } catch (Exception e) {
      log.log(Level.FINER, e.toString(), e);
    }

    for (ClusterServer server : _serverList) {
      try {
        server.init();
      } catch (Exception e) {
        throw ConfigException.create(e);
      }
    }
  }

  /**
   * Returns the server id.
   */
  public static String getServerId()
  {