queue.java

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

      AdminModule.doRequest(new UnsetQueueThreshold(agentId));
    else
      AdminModule.doRequest(new SetQueueThreshold(agentId, threshold));
  } 

  /** 
   * Monitoring method returning the threshold of this queue, -1 if not set.
   * <p>
   * The request fails if the queue is deleted server side.
   *
   * @exception ConnectException  If the admin connection is closed or broken.
   * @exception AdminException  If the request fails.
   */
  public int getThreshold() throws ConnectException, AdminException
  {
    Monitor_GetDMQSettings request = new Monitor_GetDMQSettings(agentId);
    Monitor_GetDMQSettingsRep reply;
    reply = (Monitor_GetDMQSettingsRep) AdminModule.doRequest(request);
    
    if (reply.getThreshold() == null)
      return -1;
    else
      return reply.getThreshold().intValue();
  }

  /**
   * Admin method setting nbMaxMsg for this queue.
   * <p>
   * The request fails if the queue is deleted server side.
   *
   * @param nbMaxMsg  nb Max of Message (-1 no limit).
   *
   * @exception ConnectException  If the admin connection is closed or broken.
   * @exception AdminException  If the request fails.
   */
  public void setNbMaxMsg(int nbMaxMsg)
    throws ConnectException, AdminException {
    AdminModule.doRequest(new SetNbMaxMsg(agentId, nbMaxMsg));
  } 

  /** 
   * Monitoring method returning the nbMaxMsg of this queue, -1 if no limit.
   * <p>
   * The request fails if the queue is deleted server side.
   *
   * @exception ConnectException  If the admin connection is closed or broken.
   * @exception AdminException  If the request fails.
   */
  public int getNbMaxMsg() 
    throws ConnectException, AdminException {
    Monitor_GetNbMaxMsg request = new Monitor_GetNbMaxMsg(agentId);
    Monitor_GetNbMaxMsgRep reply;
    reply = (Monitor_GetNbMaxMsgRep) AdminModule.doRequest(request);
    return reply.getNbMaxMsg();
  }
   
  /**
   * Monitoring method returning the number of pending messages on this queue.
   * <p>
   * The request fails if the queue is deleted server side.
   *
   * @exception ConnectException  If the admin connection is closed or broken.
   * @exception AdminException  If the request fails.
   */
  public int getPendingMessages() throws ConnectException, AdminException
  {
    Monitor_GetPendingMessages request =
      new Monitor_GetPendingMessages(agentId);
    Monitor_GetNumberRep reply;
    reply = (Monitor_GetNumberRep) AdminModule.doRequest(request);

    return reply.getNumber();
  }

  /**
   * Monitoring method returning the number of pending requests on this queue.
   * <p>
   * The request fails if the queue is deleted server side.
   *
   * @exception ConnectException  If the admin connection is closed or broken.
   * @exception AdminException  If the request fails.
   */
  public int getPendingRequests() throws ConnectException, AdminException
  {
    Monitor_GetPendingRequests request =
      new Monitor_GetPendingRequests(agentId);
    Monitor_GetNumberRep reply =
      (Monitor_GetNumberRep) AdminModule.doRequest(request);

    return reply.getNumber();
  }

  public String[] getMessageIds() 
    throws AdminException, ConnectException {
    GetQueueMessageIdsRep reply = 
      (GetQueueMessageIdsRep)AdminModule.doRequest(
        new GetQueueMessageIds(agentId));
    return reply.getMessageIds();
  }
  
  public javax.jms.Message readMessage(String msgId) throws AdminException, ConnectException, JMSException {
    GetQueueMessageRep reply = 
      (GetQueueMessageRep)AdminModule.doRequest(
        new GetQueueMessage(agentId, msgId));
    return Message.wrapMomMessage(null, reply.getMessage());
  }

  public String getMessageDigest(String msgId)
    throws AdminException, ConnectException, JMSException {
    GetQueueMessageRep reply = 
      (GetQueueMessageRep)AdminModule.doRequest(
        new GetQueueMessage(agentId, msgId));
    javax.jms.Message msg = Message.wrapMomMessage(null, reply.getMessage());
    StringBuffer strbuf = new StringBuffer();
    strbuf.append("Message: ").append(msg.getJMSMessageID());
    strbuf.append("\n\tTo: ").append(msg.getJMSDestination());
    strbuf.append("\n\tCorrelationId: ").append(msg.getJMSCorrelationID());
    strbuf.append("\n\tDeliveryMode: ").append(msg.getJMSDeliveryMode());
    strbuf.append("\n\tExpiration: ").append(msg.getJMSExpiration());
    strbuf.append("\n\tPriority: ").append(msg.getJMSPriority());
    strbuf.append("\n\tRedelivered: ").append(msg.getJMSRedelivered());
    strbuf.append("\n\tReplyTo: ").append(msg.getJMSReplyTo());
    strbuf.append("\n\tTimestamp: ").append(msg.getJMSTimestamp());
    strbuf.append("\n\tType: ").append(msg.getJMSType());
    return strbuf.toString();
  }

  public Properties getMessageHeader(String msgId)
    throws AdminException, ConnectException, JMSException {
    GetQueueMessageRep reply = 
      (GetQueueMessageRep)AdminModule.doRequest(
        new GetQueueMessage(agentId, msgId));
    Message msg = (Message) Message.wrapMomMessage(null, reply.getMessage());

    Properties prop = new Properties();
    prop.setProperty("JMSMessageID", msg.getJMSMessageID());
    prop.setProperty("JMSDestination", msg.getJMSDestination().toString());
    if (msg.getJMSCorrelationID() != null)
      prop.setProperty("JMSCorrelationID", msg.getJMSCorrelationID());
    prop.setProperty("JMSDeliveryMode",
                     new Integer(msg.getJMSDeliveryMode()).toString());
    prop.setProperty("JMSExpiration",
                     new Long(msg.getJMSExpiration()).toString());
    prop.setProperty("JMSPriority",
                     new Integer(msg.getJMSPriority()).toString());
    prop.setProperty("JMSRedelivered",
                     new Boolean(msg.getJMSRedelivered()).toString());
    if (msg.getJMSReplyTo() != null)
      prop.setProperty("JMSReplyTo", msg.getJMSReplyTo().toString());
    prop.setProperty("JMSTimestamp",
                     new Long(msg.getJMSTimestamp()).toString());
    if (msg.getJMSType() != null)
      prop.setProperty("JMSType", msg.getJMSType());

    if (msg.momMsg != null) {
      msg.momMsg.getOptionalHeader(prop);
    }

    return prop;
  }

  public Properties getMessageProperties(String msgId)
    throws AdminException, ConnectException, JMSException {
    GetQueueMessageRep reply = 
      (GetQueueMessageRep)AdminModule.doRequest(
        new GetQueueMessage(agentId, msgId));
    Message msg = (Message) Message.wrapMomMessage(null, reply.getMessage());

    Properties prop = new Properties();
    if (msg.momMsg != null) {
      msg.momMsg.getProperties(prop);
    }
    return prop;
  }

  public void deleteMessage(
    String msgId)
    throws AdminException, ConnectException {
    AdminModule.doRequest(new DeleteQueueMessage(agentId, msgId));
  }

  public void clear() 
    throws AdminException, ConnectException {
    AdminModule.doRequest(new ClearQueue(agentId));
  }

  /**
   * Adds a queue into the cluster this queue belongs to.
   * If this queue doesn't belong to a cluster then a cluster is
   * created by clustering this queue with the added queue.
   * <p>
   * The request fails if one or both of the queues are deleted, or
   * can't belong to a cluster.
   *
   * @param addedQueue queue added to the cluster
   *
   * @exception ConnectException  If the admin connection is closed or broken.
   * @exception AdminException  If the request fails.
   */
  public void addClusteredQueue(Queue addedQueue)
    throws ConnectException, AdminException {
    AdminModule.doRequest(
      new AddQueueCluster(agentId, addedQueue.getName()));
  }

  /**
   * Removes a queue from the cluster this queue belongs to.
   * <p>
   * The request fails if the queue does not exist or is not part of any 
   * cluster.
   *
   * @param removedQueue queue removed from the cluster
   * 
   * @exception ConnectException  If the admin connection is closed or broken.
   * @exception AdminException  If the request fails.
   */
  public void removeClusteredQueue(Queue removedQueue)
    throws ConnectException, AdminException {
    AdminModule.doRequest(
      new RemoveQueueCluster(agentId, removedQueue.getName()));
  }

  /**
   * Returns the reference of the queues that belong to the cluster.
   *
   * @exception ConnectException  If the admin connection is closed or broken.
   * @exception AdminException  If the request fails.
   */
  public String[] getQueueClusterElements()
    throws ConnectException, AdminException {
    AdminReply reply = AdminModule.doRequest(
      new ListClusterQueue(agentId));
    Vector list = (Vector)reply.getReplyObject();
    String[] res = new String[list.size()];
    list.copyInto(res);
    return res;
  }
}