src/org/objectweb/cjdbc/controller/loadbalancer/singledb/SingleDB.java

説明を見る。

package org.objectweb.cjdbc.controller.loadbalancer.singledb;

import java.sql.Connection;
import java.sql.SQLException;

import org.objectweb.cjdbc.common.exceptions.BadConnectionException;
import org.objectweb.cjdbc.common.exceptions.UnreachableBackendException;
import org.objectweb.cjdbc.common.i18n.Translate;
import org.objectweb.cjdbc.common.log.Trace;
import org.objectweb.cjdbc.common.sql.AbstractWriteRequest;
import org.objectweb.cjdbc.common.sql.ParsingGranularities;
import org.objectweb.cjdbc.common.sql.SelectRequest;
import org.objectweb.cjdbc.common.sql.StoredProcedure;
import org.objectweb.cjdbc.common.xml.DatabasesXmlTags;
import org.objectweb.cjdbc.controller.backend.DatabaseBackend;
import org.objectweb.cjdbc.controller.cache.metadata.MetadataCache;
import org.objectweb.cjdbc.controller.connection.AbstractConnectionManager;
import org.objectweb.cjdbc.controller.loadbalancer.AbstractLoadBalancer;
import org.objectweb.cjdbc.controller.requestmanager.RAIDbLevels;
import org.objectweb.cjdbc.controller.requestmanager.TransactionMarkerMetaData;
import org.objectweb.cjdbc.controller.virtualdatabase.ControllerResultSet;
import org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase;

public class SingleDB extends AbstractLoadBalancer
{
  /*
   * How the code is organized ? 1. Member variables 2. Constructor(s) 3.
   * Request handling 4. Transaction handling 5. Backend management 6.
   * Debug/Monitoring
   */

  private DatabaseBackend backend;

  private static Trace    logger = Trace
                                     .getLogger("org.objectweb.cjdbc.controller.loadbalancer.SingleDB");

  /*
   * Constructors
   */

  public SingleDB(VirtualDatabase vdb) throws SQLException
  {
    // We don't need to parse the requests, just send them to the backend
    super(vdb, RAIDbLevels.SingleDB, ParsingGranularities.NO_PARSING);
  }

  /*
   * Request Handling
   */

  public ControllerResultSet execReadRequest(SelectRequest request,
      MetadataCache metadataCache) throws SQLException
  {
    if (backend == null)
      throw new SQLException(Translate.get(
          "loadbalancer.execute.no.backend.available", request.getId()));

    try
    {
      AbstractConnectionManager cm = backend.getConnectionManager(request
          .getLogin());
      if (request.isAutoCommit())
      {
        ControllerResultSet rs = null;
        boolean badConnection;
        do
        {
          badConnection = false;
          // Use a connection just for this request
          Connection c = null;
          try
          {
            c = cm.getConnection();
          }
          catch (UnreachableBackendException e1)
          {
            String backendName = backend.getName();
            logger.error(Translate.get(
                "loadbalancer.backend.disabling.unreachable", backendName));
            backend.disable();
            backend = null;
            throw new SQLException(Translate.get(
                "loadbalancer.backend.unreacheable", backendName));
          }

          // Sanity check
          if (c == null)
            throw new SQLException(Translate.get(
                "loadbalancer.backend.no.connection", backend.getName()));

          // Execute Query
          try
          {
            rs = executeSelectRequestOnBackend(request, backend, c,
                metadataCache);
            cm.releaseConnection(c);
          }
          catch (SQLException e)
          {
            cm.releaseConnection(c);
            throw new SQLException(Translate.get(
                "loadbalancer.request.failed.on.backend", new String[]{
                    request.getSQLShortForm(vdb.getSQLShortFormLength()),
                    backend.getName(), e.getMessage()}));
          }
          catch (BadConnectionException e)
          { // Get rid of the bad connection
            cm.deleteConnection(c);
            badConnection = true;
          }
        }
        while (badConnection);
        return rs;
      }
      else
      {
        long tid = request.getTransactionId();
        // Re-use the connection used by this transaction
        Connection c = cm.retrieveConnection(tid);

        // Sanity check
        if (c == null)
          throw new SQLException(Translate.get(
              "loadbalancer.unable.retrieve.connection", new String[]{
                  String.valueOf(tid), backend.getName()}));

        // Execute Query
        ControllerResultSet rs = null;
        try
        {
          rs = executeSelectRequestOnBackend(request, backend, c, metadataCache);
        }
        catch (SQLException e)
        {
          throw new SQLException(Translate.get(
              "loadbalancer.request.failed.on.backend", new String[]{
                  request.getSQLShortForm(vdb.getSQLShortFormLength()),
                  backend.getName(), e.getMessage()}));
        }
        catch (BadConnectionException e)
        { // Get rid of the bad connection
          cm.deleteConnection(tid);
          throw new SQLException(Translate.get(
              "loadbalancer.connection.failed", new String[]{
                  String.valueOf(tid), backend.getName(), e.getMessage()}));
        }
        return rs;
      }
    }
    catch (RuntimeException e)
    {
      String msg = "Request '"
          + request.getSQLShortForm(vdb.getSQLShortFormLength())
          + "' failed on backend " + backend.getURL() + " (" + e + ")";
      logger.fatal(msg, e);
      throw new SQLException(msg);
    }
  }

  public int execWriteRequest(AbstractWriteRequest request) throws SQLException
  {
    if (backend == null)
      throw new SQLException(Translate.get(
          "loadbalancer.execute.no.backend.available", request.getId()));

    try
    {
      AbstractConnectionManager cm = backend.getConnectionManager(request
          .getLogin());
      if (request.isAutoCommit())
      {
        // We do not execute request outside the already open transactions if we
        // are disabling the backend.
        if (backend.isDisabling())
          throw new SQLException(Translate.get(
              "loadbalancer.backend.is.disabling", new String[]{
                  request.getSQLShortForm(vdb.getSQLShortFormLength()),
                  backend.getName()}));

        // Use a connection just for this request
        Connection c = null;
        try
        {
          c = cm.getConnection();
        }
        catch (UnreachableBackendException e1)
        {
          String backendName = backend.getName();
          logger.error(Translate.get(
              "loadbalancer.backend.disabling.unreachable", backendName));
          backend.disable();
          backend = null;
          throw new SQLException(Translate.get(
              "loadbalancer.backend.unreacheable", backendName));
        }

        // Sanity check
        if (c == null)
          throw new SQLException(Translate.get(
              "loadbalancer.backend.no.connection", backend.getName()));

        // Execute Query
        int result;
        try
        {
          result = executeUpdateRequestOnBackend(request, backend, c);
        }
        catch (Exception e)
        {
          throw new SQLException(Translate.get(
              "loadbalancer.request.failed.on.backend", new String[]{
                  request.getSQLShortForm(vdb.getSQLShortFormLength()),
                  backend.getName(), e.getMessage()}));
        }
        finally
        {
          cm.releaseConnection(c);
        }
        return result;
      }
      else
      { // Re-use the connection used by this transaction
        Connection c = cm.retrieveConnection(request.getTransactionId());

        // Sanity check
        if (c == null)
          throw new SQLException(Translate.get(
              "loadbalancer.unable.retrieve.connection",
              new String[]{String.valueOf(request.getTransactionId()),
                  backend.getName()}));

        // Execute Query
        try
        {
          return executeUpdateRequestOnBackend(request, backend, c);
        }
        catch (Exception e)
        {
          throw new SQLException(Translate.get(
              "loadbalancer.request.failed.on.backend", new String[]{
                  request.getSQLShortForm(vdb.getSQLShortFormLength()),
                  backend.getName(), e.getMessage()}));
        }
      }
    }
    catch (RuntimeException e)
    {
      String msg = Translate.get("loadbalancer.request.failed.on.backend",
          new String[]{request.getSQLShortForm(vdb.getSQLShortFormLength()),
              backend.getName(), e.getMessage()});
      logger.fatal(msg, e);
      throw new SQLException(msg);
    }
  }

  public ControllerResultSet execWriteRequestWithKeys(
      AbstractWriteRequest request, MetadataCache metadataCache)
      throws SQLException
  {
    if (backend == null)
      throw new SQLException(Translate.get(
          "loadbalancer.execute.no.backend.available", request.getId()));

    if (!backend.getDriverCompliance().supportGetGeneratedKeys())
      throw new SQLException(Translate.get(
          "loadbalancer.backend.autogeneratedkeys.unsupported", backend
              .getName()));

    try
    {
      AbstractConnectionManager cm = backend.getConnectionManager(request
          .getLogin());
      if (request.isAutoCommit())
      {
        // We do not execute request outside the already open transactions if we
        // are disabling the backend.
        if (backend.isDisabling())
          throw new SQLException(Translate.get(
              "loadbalancer.backend.is.disabling", new String[]{
                  request.getSQLShortForm(vdb.getSQLShortFormLength()),
                  backend.getName()}));

        // Use a connection just for this request
        Connection c = null;
        try
        {
          c = cm.getConnection();
        }
        catch (UnreachableBackendException e1)
        {
          String backendName = backend.getName();
          logger.error(Translate.get(
              "loadbalancer.backend.disabling.unreachable", backendName));
          backend.disable();
          backend = null;
          throw new SQLException(Translate.get(
              "loadbalancer.backend.unreacheable", backendName));
        }

        // Sanity check
        if (c == null)
          throw new SQLException(Translate.get(
              "loadbalancer.backend.no.connection", backend.getName()));

        // Execute Query
        ControllerResultSet result;
        try
        {
          result = new ControllerResultSet(request,
              executeUpdateRequestOnBackendWithKeys(request, backend, c),
              metadataCache);
        }
        catch (Exception e)
        {
          throw new SQLException(Translate.get(
              "loadbalancer.request.failed.on.backend", new String[]{
                  request.getSQLShortForm(vdb.getSQLShortFormLength()),
                  backend.getName(), e.getMessage()}));
        }
        finally
        {
          cm.releaseConnection(c);
        }
        return result;
      }
      else
      {
        // Re-use the connection used by this transaction
        Connection c = cm.retrieveConnection(request.getTransactionId());

        // Sanity check
        if (c == null)
          throw new SQLException(Translate.get(
              "loadbalancer.unable.retrieve.connection",
              new String[]{String.valueOf(request.getTransactionId()),
                  backend.getName()}));

        // Execute Query
        try
        {
          return new ControllerResultSet(request,
              executeUpdateRequestOnBackendWithKeys(request, backend, c),
              metadataCache);
        }
        catch (Exception e)
        {
          throw new SQLException(Translate.get(
              "loadbalancer.request.failed.on.backend", new String[]{
                  request.getSQLShortForm(vdb.getSQLShortFormLength()),
                  backend.getName(), e.getMessage()}));
        }
        finally
        {
          backend.removePendingRequest(request);
        }
      }
    }
    catch (RuntimeException e)
    {
      String msg = Translate.get("loadbalancer.request.failed.on.backend",
          new String[]{request.getSQLShortForm(vdb.getSQLShortFormLength()),
              backend.getName(), e.getMessage()});
      logger.fatal(msg, e);
      throw new SQLException(msg);
    }
  }

  public ControllerResultSet execReadOnlyReadStoredProcedure(
      StoredProcedure proc, MetadataCache metadataCache) throws SQLException
  {
    return execReadStoredProcedure(proc, metadataCache);
  }

  public ControllerResultSet execReadStoredProcedure(StoredProcedure proc,
      MetadataCache metadataCache) throws SQLException
  {
    if (backend == null)
      throw new SQLException(
          "No available backend to execute stored procedure " + proc.getId());

    try
    {
      AbstractConnectionManager cm = backend.getConnectionManager(proc
          .getLogin());
      if (proc.isAutoCommit())
      { // Use a connection just for this request
        Connection c = null;
        try
        {
          c = cm.getConnection();
        }
        catch (UnreachableBackendException e1)
        {
          String backendName = backend.getName();
          logger.error(Translate.get(
              "loadbalancer.backend.disabling.unreachable", backendName));
          backend.disable();
          backend = null;
          throw new SQLException(Translate.get(
              "loadbalancer.backend.unreacheable", backendName));
        }

        // Sanity check
        if (c == null)
          throw new SQLException(Translate.get(
              "loadbalancer.backend.no.connection", backend.getName()));

        // Execute Query
        ControllerResultSet rs = null;
        try
        {
          rs = AbstractLoadBalancer.executeReadStoredProcedureOnBackend(proc,
              backend, c, metadataCache);
        }
        catch (Exception e)
        {
          throw new SQLException(Translate.get(
              "loadbalancer.storedprocedure.failed.on.backend", new String[]{
                  proc.getSQLShortForm(vdb.getSQLShortFormLength()),
                  backend.getName(), e.getMessage()}));
        }
        finally
        {
          cm.releaseConnection(c);
        }
        return rs;
      }
      else
      { // Re-use the connection used by this transaction
        Connection c = cm.retrieveConnection(proc.getTransactionId());

        // Sanity check
        if (c == null)
          throw new SQLException(Translate.get(
              "loadbalancer.unable.retrieve.connection", new String[]{
                  String.valueOf(proc.getTransactionId()), backend.getName()}));

        // Execute Query
        try
        {
          return AbstractLoadBalancer.executeReadStoredProcedureOnBackend(proc,
              backend, c, metadataCache);
        }
        catch (Exception e)
        {
          throw new SQLException(Translate.get(
              "loadbalancer.storedprocedure.failed.on.backend", new String[]{
                  proc.getSQLShortForm(vdb.getSQLShortFormLength()),
                  backend.getName(), e.getMessage()}));
        }
      }
    }
    catch (RuntimeException e)
    {
      String msg = Translate.get(
          "loadbalancer.storedprocedure.failed.on.backend", new String[]{
              proc.getSQLShortForm(vdb.getSQLShortFormLength()),
              backend.getName(), e.getMessage()});
      logger.fatal(msg, e);
      throw new SQLException(msg);
    }
  }

  public int execWriteStoredProcedure(StoredProcedure proc) throws SQLException
  {
    if (backend == null)
      throw new SQLException(
          "No available backend to execute stored procedure " + proc.getId());

    try
    {
      AbstractConnectionManager cm = backend.getConnectionManager(proc
          .getLogin());
      if (proc.isAutoCommit())
      { // Use a connection just for this request
        Connection c = null;
        try
        {
          c = cm.getConnection();
        }
        catch (UnreachableBackendException e1)
        {
          String backendName = backend.getName();
          logger.error(Translate.get(
              "loadbalancer.backend.disabling.unreachable", backendName));
          backend.disable();
          backend = null;
          throw new SQLException(Translate.get(
              "loadbalancer.backend.unreacheable", backendName));
        }

        // Sanity check
        if (c == null)
          throw new SQLException(Translate.get(
              "loadbalancer.backend.no.connection", backend.getName()));

        // Execute Query
        int result;
        try
        {
          result = AbstractLoadBalancer.executeWriteStoredProcedureOnBackend(
              proc, backend, c);
        }
        catch (Exception e)
        {
          throw new SQLException(Translate.get(
              "loadbalancer.storedprocedure.failed.on.backend", new String[]{
                  proc.getSQLShortForm(vdb.getSQLShortFormLength()),
                  backend.getName(), e.getMessage()}));
        }
        finally
        {
          cm.releaseConnection(c);
        }
        return result;
      }
      else
      { // Re-use the connection used by this transaction
        Connection c = cm.retrieveConnection(proc.getTransactionId());

        // Sanity check
        if (c == null)
          throw new SQLException(Translate.get(
              "loadbalancer.unable.retrieve.connection", new String[]{
                  String.valueOf(proc.getTransactionId()), backend.getName()}));

        // Execute Query
        try
        {
          return AbstractLoadBalancer.executeWriteStoredProcedureOnBackend(
              proc, backend, c);
        }
        catch (Exception e)
        {
          throw new SQLException(Translate.get(
              "loadbalancer.storedprocedure.failed.on.backend", new String[]{
                  proc.getSQLShortForm(vdb.getSQLShortFormLength()),
                  backend.getName(), e.getMessage()}));
        }
      }
    }
    catch (RuntimeException e)
    {
      String msg = Translate.get(
          "loadbalancer.storedprocedure.failed.on.backend", new String[]{
              proc.getSQLShortForm(vdb.getSQLShortFormLength()),
              backend.getName(), e.getMessage()});
      logger.fatal(msg, e);
      throw new SQLException(msg);
    }
  }

  /*
   * Transaction management
   */

  public void begin(TransactionMarkerMetaData tm) throws SQLException
  {
    if (backend == null)
      throw new SQLException("No available backend to begin transaction "
          + tm.getTransactionId());

    // We do not accept new transactions if we are disabling the backend
    if (backend.isDisabling())
      throw new SQLException(Translate.get("loadbalancer.backend.is.disabling",
          new String[]{"begin transaction " + tm.getTransactionId(),
              backend.getName()}));

    try
    {
      Connection c = backend.getConnectionManager(tm.getLogin()).getConnection(
          tm.getTransactionId());

      if (c == null)
        throw new SQLException(Translate.get(
            "loadbalancer.backend.no.connection", backend.getName()));

      c.setAutoCommit(false);
    }
    catch (Exception e)
    {
      throw new SQLException("Begin of transaction " + tm.getTransactionId()
          + " failed on backend " + backend.getURL() + " (" + e + ")");
    }
  }

  public void commit(TransactionMarkerMetaData tm) throws SQLException
  {
    if (backend == null)
      throw new SQLException("No available backend to commit transaction "
          + tm.getTransactionId());

    try
    {
      AbstractConnectionManager cm = backend
          .getConnectionManager(tm.getLogin());
      Connection c = cm.retrieveConnection(tm.getTransactionId());

      if (c == null)
        throw new SQLException("No connection found for transaction "
            + tm.getTransactionId());

      try
      {
        c.commit();
        c.setAutoCommit(true);
      }
      catch (SQLException e)
      {
        throw new SQLException(Translate.get("loadbalancer.commit.failed",
            new String[]{String.valueOf(tm.getTransactionId()),
                backend.getName(), e.getMessage()}));
      }
      finally
      {
        cm.releaseConnection(tm.getTransactionId());
      }
    }
    catch (RuntimeException e)
    {
      String msg = Translate.get("loadbalancer.commit.failed", new String[]{
          String.valueOf(tm.getTransactionId()), backend.getName(),
          e.getMessage()});
      logger.fatal(msg, e);
      throw new SQLException(msg);
    }
  }

  public void rollback(TransactionMarkerMetaData tm) throws SQLException
  {
    if (backend == null)
      throw new SQLException("No available backend to rollback transaction "
          + tm.getTransactionId());

    try
    {
      AbstractConnectionManager cm = backend
          .getConnectionManager(tm.getLogin());
      Connection c = cm.retrieveConnection(tm.getTransactionId());

      if (c == null)
        throw new SQLException("No connection found for transaction "
            + tm.getTransactionId());

      try
      {
        c.rollback();
        c.setAutoCommit(true);
      }
      catch (SQLException e)
      {
        throw new SQLException(Translate.get("loadbalancer.rollback.failed",
            new String[]{String.valueOf(tm.getTransactionId()),
                backend.getName(), e.getMessage()}));
      }
      finally
      {
        cm.releaseConnection(tm.getTransactionId());
      }
    }
    catch (RuntimeException e)
    {
      String msg = Translate.get("loadbalancer.rollback.failed", new String[]{
          String.valueOf(tm.getTransactionId()), backend.getName(),
          e.getMessage()});
      logger.fatal(msg, e);
      throw new SQLException(msg);
    }
  }

  /*
   * Backends management
   */

  public void enableBackend(DatabaseBackend db, boolean writeEnabled)
      throws SQLException
  {
    if (backend != null)
    {
      if (backend.isReadEnabled())
        throw new SQLException(
            "SingleDB load balancer accepts only one backend and "
                + backend.getName() + " is already enabled. Skipping "
                + db.getName() + " initialization.");
    }
    backend = db;
    logger.info(Translate.get("loadbalancer.backend.enabling", db.getName()));
    if (!backend.isInitialized())
      backend.initializeConnections();
    backend.enableRead();
    if (writeEnabled)
      backend.enableWrite();
  }

  public void disableBackend(DatabaseBackend db) throws SQLException
  {
    if (backend.equals(db))
    {
      logger
          .info(Translate.get("loadbalancer.backend.disabling", db.getName()));
      backend.disable();
      if (backend.isInitialized())
        backend.finalizeConnections();
      backend = null;
    }
    else
    {
      String msg = "Trying to disable a non-existing backend " + db.getName();
      logger.warn(msg);
      throw new SQLException(msg);
    }
  }

  public void setWeight(String name, int w) throws SQLException
  {
    throw new SQLException("Weight is not supported with this load balancer");
  }

  /*
   * Debug/Monitoring
   */

  public String getInformation()
  {
    if (backend == null)
      return "SingleDB Request load balancer: !!!Warning!!! No enabled backend node found\n";
    else
      return "SingleDB Request load balancer using " + backend.getURL() + "\n";
  }

  public String getXmlImpl()
  {
    return "<" + DatabasesXmlTags.ELT_SingleDB + "/>";
  }

}

---