joramadapter.java

一个类似于openJMS分布在ObjectWeb之下的JMS消息中间件。

                              + exc);
    }
    catch (AdminException exc) {
      throw new ResourceException("Problem when handling the JORAM "
                                  + "destinations: " + exc);
    }
  }

  /**
   * Notifies the adapter to deactivate message delivery for a given endpoint.
   */
  public void endpointDeactivation(MessageEndpointFactory endpointFactory,
                                   ActivationSpec spec) {
    if (AdapterTracing.dbgAdapter.isLoggable(BasicLevel.DEBUG))
      AdapterTracing.dbgAdapter.log(BasicLevel.DEBUG,
                                    this + " endpointDeactivation(" + endpointFactory + ", " + spec + ")");
    if (! started || stopped)
      return;

    if (AdapterTracing.dbgAdapter.isLoggable(BasicLevel.DEBUG))
      AdapterTracing.dbgAdapter.log(BasicLevel.DEBUG,
                                    "Deactivating Endpoint on JORAM adapter.");

    ((InboundConsumer) consumers.remove(spec)).close();
  }

  /**
   * Returns XA resources given an array of ActivationSpec instances.
   *
   * @exception IllegalStateException  If the adapter is either not started,
   *                                   or stopped.
   * @exception NotSupportedException  If provided activation parameters
   *                                   are invalid.
   * @exception CommException          If the JORAM server is not reachable.
   * @exception SecurityException      If connecting is not allowed.
   * @exception ResourceException      Generic exception.
   */
  public XAResource[] getXAResources(ActivationSpec[] specs)
    throws ResourceException {
    if (AdapterTracing.dbgAdapter.isLoggable(BasicLevel.DEBUG))
      AdapterTracing.dbgAdapter.log(BasicLevel.DEBUG,
                                    this + " getXAResources(" + specs + ")");

    if (! started)
      throw new IllegalStateException("Non started resource adapter.");
    if (stopped)
      throw new IllegalStateException("Stopped resource adapter.");

    ActivationSpecImpl specImpl;
    String userName;
    String password;
    XAConnectionFactory connectionFactory = null;
    XAConnection connection;
    Vector resources = new Vector();

    if (connections == null)
      connections = new Hashtable();

    try {
      for (int i = 0; i < specs.length; i++) {
        if (! (specs[i] instanceof ActivationSpecImpl))
          throw new ResourceException("Provided ActivationSpec instance is "
                                      + "not a JORAM activation spec.");

        specImpl = (ActivationSpecImpl) specs[i];

        if (! specImpl.getResourceAdapter().equals(this))
          throw new ResourceException("Supplied ActivationSpec instance "
                                      + "associated to an other "
                                      + "ResourceAdapter.");

        userName = specImpl.getUserName();

        // The connection does not already exist: creating it.
        if (! connections.containsKey(userName)) {
          password = specImpl.getPassword();

          if (isHa) {
              if (collocated)
                  connectionFactory = XAHALocalConnectionFactory.create();
              else {
                  String urlHa = "hajoram://" + hostName + ":" + serverPort;
                  connectionFactory = XAHATcpConnectionFactory.create(urlHa);
              }
          }  else {
          if (collocated)
            connectionFactory = XALocalConnectionFactory.create();
          else
                  connectionFactory = XATcpConnectionFactory.create(hostName, serverPort);
          }

          ((org.objectweb.joram.client.jms.XAConnectionFactory) connectionFactory).getParameters().connectingTimer = connectingTimer;
          ((org.objectweb.joram.client.jms.XAConnectionFactory) connectionFactory).getParameters().cnxPendingTimer = cnxPendingTimer;
          ((org.objectweb.joram.client.jms.XAConnectionFactory) connectionFactory).getParameters().txPendingTimer = txPendingTimer;

          connection =
            connectionFactory.createXAConnection(userName, password);

          connections.put(userName, connection);

          resources.add(connection.createXASession().getXAResource());
        }
        if (AdapterTracing.dbgAdapter.isLoggable(BasicLevel.DEBUG))
          AdapterTracing.dbgAdapter.log(BasicLevel.DEBUG,
                                        this + " getXAResources resources = " + resources);
      }
    }
    catch (javax.jms.JMSSecurityException exc) {
      throw new SecurityException("Invalid user identification: " + exc);
    }
    catch (javax.jms.JMSException exc) {
      throw new CommException("Could not connect to the JORAM server: "
                              + exc);
    }

    return (XAResource[]) resources.toArray(new XAResource[resources.size()]);
  }


  /** Returns a code depending on the adapter properties. */
  public int hashCode()
  {
    return (collocated + " " + hostName + " " + serverPort).hashCode();
  }

  /** Compares adapters according to their properties. */
  public boolean equals(Object o)
  {
    if (! (o instanceof JoramAdapter))
      return false;

    JoramAdapter other = (JoramAdapter) o;

    boolean res =
      collocated == other.collocated
      && hostName.equals(other.hostName)
      && serverPort == other.serverPort;

    if (AdapterTracing.dbgAdapter.isLoggable(BasicLevel.DEBUG))
      AdapterTracing.dbgAdapter.log(BasicLevel.DEBUG,
                                    this + " equals = " + res);
    return res;
  }

  public List getDestinations() {
    return joramAdmin.getDestinations();
  }

  public List getDestinations(int serverId) {
    return joramAdmin.getDestinations(serverId);
  }

  public List getUsers() {
    return joramAdmin.getUsers();
  }

  public List getUsers(int serverId) {
    return joramAdmin.getUsers(serverId);
  }

  public List getPlatformServersIds() {
    return joramAdmin.getPlatformAdmin().getServersIds();
  }

  public List getLocalUsers() {
    return joramAdmin.getUsers(serverId);
  }

  public void setDefaultDMQ(int serverId, DeadMQueue dmq)
    throws ConnectException, AdminException {
    joramAdmin.setDefaultDMQ(serverId,dmq);
  }

  public DeadMQueue getDefaultDMQ(int serverId)
    throws ConnectException, AdminException {
    return joramAdmin.getDefaultDMQ(serverId);
  }

  public DeadMQueue getDefaultDMQ()
    throws ConnectException, AdminException {
    return joramAdmin.getDefaultDMQ();
  }

  public void exit() {
    joramAdmin.exit();
  }

  /**
   * wait before abort a request.
   */
  public void setTimeOutToAbortRequest(long timeOut) {
    joramAdmin.setTimeOutToAbortRequest(timeOut);
  }

  /**
   * wait before abort a request.
   */
  public long getTimeOutToAbortRequest() {
    return joramAdmin.getTimeOutToAbortRequest();
  }

  /**
   * Creates or retrieves a user on the underlying JORAM server.
   *
   * @exception AdminException   If the creation fails.
   */
  public void createUser(String name, String password)
    throws AdminException {
    try {
      User.create(name, password);
      if (AdapterTracing.dbgAdapter.isLoggable(BasicLevel.INFO))
        AdapterTracing.dbgAdapter.log(BasicLevel.INFO,
                                      "  - User [" + name + "] has been created.");
    }
    catch (ConnectException exc) {
      throw new AdminException("createUser() failed: admin connection "
                               + "has been lost.");
    }
  }

  /**
   * Creates or retrieves a user on the underlying JORAM server.
   *
   * @exception AdminException   If the creation fails.
   */
  public void createUser(String name, String password, int serverId)
    throws AdminException {
    try {
      User.create(name, password,serverId);
      if (AdapterTracing.dbgAdapter.isLoggable(BasicLevel.INFO))
        AdapterTracing.dbgAdapter.log(BasicLevel.INFO,
                                      "  - User [" + name + "] has been created.");
    }
    catch (ConnectException exc) {
      throw new AdminException("createUser() failed: admin connection "
                               + "has been lost.");
    }
  }

  /**
   * Creates a non managed connection factory and binds it to JNDI.
   */
  public void createCF(String name) {
    ManagedConnectionFactoryImpl mcf =
      new ManagedConnectionFactoryImpl();

    try {
      mcf.setResourceAdapter(this);
      mcf.setCollocated(new Boolean(false));

      Object factory = mcf.createConnectionFactory();
      bind(name, factory);
      if (AdapterTracing.dbgAdapter.isLoggable(BasicLevel.INFO))
        AdapterTracing.dbgAdapter.log(BasicLevel.INFO,
                                      "  - ConnectionFactory [" + name
                                      + "] has been created and bound.");
    } catch (Exception exc) {}
  }

  /**
   * Creates a non managed PTP connection factory and binds it to JNDI.
   */
  public void createQCF(String name) {
    ManagedConnectionFactoryImpl mcf =
      new ManagedQueueConnectionFactoryImpl();

    try {
      mcf.setResourceAdapter(this);
      mcf.setCollocated(new Boolean(false));

      Object factory = mcf.createConnectionFactory();
      bind(name, factory);
      if (AdapterTracing.dbgAdapter.isLoggable(BasicLevel.INFO))
        AdapterTracing.dbgAdapter.log(BasicLevel.INFO,
                                      "  - QueueConnectionFactory [" + name
                                      + "] has been created and bound.");
    } catch (Exception exc) {}
  }

  /**
   * Creates a non managed PubSub connection factory and binds it to JNDI.
   */
  public void createTCF(String name) {
    ManagedConnectionFactoryImpl mcf =
      new ManagedTopicConnectionFactoryImpl();

    try {
      mcf.setResourceAdapter(this);
      mcf.setCollocated(new Boolean(false));

      Object factory = mcf.createConnectionFactory();
      bind(name, factory);
      if (AdapterTracing.dbgAdapter.isLoggable(BasicLevel.INFO))
        AdapterTracing.dbgAdapter.log(BasicLevel.INFO,
                                      "  - TopicConnectionFactory [" + name
                                      + "] has been created and bound.");
    } catch (Exception exc) {}
  }

  /**
   * Initiates an admin session.
   *
   * @exception AdminException  If the admin session could not be started.
   */
  void adminConnect() throws AdminException
  {
    try {
      TopicConnectionFactory factory;

      if (isHa) {
          if (collocated)
              factory = TopicHALocalConnectionFactory.create();
          else {
              String urlHa = "hajoram://" + hostName + ":" + serverPort;
              factory = TopicHATcpConnectionFactory.create(urlHa);
          }
      } else {
      if (collocated)
        factory = TopicLocalConnectionFactory.create();
      else
        factory = TopicTcpConnectionFactory.create(hostName, serverPort);
      }

      ((org.objectweb.joram.client.jms.ConnectionFactory) factory)
        .getParameters().connectingTimer = 60;

      joramAdmin = new JoramAdmin(factory, "root", "root");
    }
    catch (ConnectException exc) {
      throw new AdminException("Admin connection can't be established: "
                               + exc.getMessage());
    }
  }

  /** Adds a given managed connection to the list of producers. */
  void addProducer(ManagedConnectionImpl managedCx)
  {
    producers.add(managedCx);
  }

  /** Removes a given managed connection from the list of producers. */
  void removeProducer(ManagedConnectionImpl managedCx)
  {
    producers.remove(managedCx);
  }

  /** remove prefix name scn:comp/ */
  private static String removePrefix(String name) {
    String PREFIX_NAME = "scn:comp/";
    try {
      if (name.startsWith(PREFIX_NAME))
        return name.substring(PREFIX_NAME.length());
      else
        return name;
    } catch (Exception e) {
      return name;
    }
  }

  /**
   * Creates or retrieves a queue destination on the underlying JORAM server,
   * (re)binds the corresponding <code>Queue</code> instance.
   *
   * @param serverId   The identifier of the server where deploying the queue.
   * @param name       The name of the queue.
   * @param className  The queue class name.
   * @param prop       The queue properties.
   *
   * @exception AdminException   If the creation fails.
   */
  public Destination createQueue(int serverId,
                                 String name,
                                 String className,
                                 Properties prop)
    throws AdminException {
    try {
      Context ctx = new InitialContext();
      return (Queue) ctx.lookup(name);
    } catch (javax.naming.NamingException exc) {
      try {