View Javadoc

    1   /***
    2    *
    3    * Copyright 2004 Protique Ltd
    4    *
    5    * Licensed under the Apache License, Version 2.0 (the "License");
    6    * you may not use this file except in compliance with the License.
    7    * You may obtain a copy of the License at
    8    *
    9    * http://www.apache.org/licenses/LICENSE-2.0
    10   *
    11   * Unless required by applicable law or agreed to in writing, software
    12   * distributed under the License is distributed on an "AS IS" BASIS,
    13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    14   * See the License for the specific language governing permissions and
    15   * limitations under the License.
    16   *
    17   **/
    18  package org.codehaus.activemq.service.impl;
    19
    20  import org.apache.commons.logging.Log;
    21  import org.apache.commons.logging.LogFactory;
    22  import org.codehaus.activemq.message.ActiveMQMessage;
    23  import org.codehaus.activemq.message.MessageAck;
    24  import org.codehaus.activemq.service.MessageIdentity;
    25  import org.codehaus.activemq.service.Subscription;
    26  import org.codehaus.activemq.service.TopicMessageContainer;
    27  import org.codehaus.activemq.store.TopicMessageStore;
    28
    29  import javax.jms.JMSException;
    30
    31  /***
    32   * A default implemenation of a Durable Topic based
    33   * {@link org.codehaus.activemq.service.MessageContainer}
    34   * which acts as an adapter between the {@link org.codehaus.activemq.service.MessageContainerManager}
    35   * requirements and those of the persistent {@link TopicMessageStore} implementations.
    36   *
    37   * @version $Revision: 1.13 $
    38   */
    39  public class DurableTopicMessageContainer implements TopicMessageContainer {
    40      private static final Log log = LogFactory.getLog(DurableTopicMessageContainer.class);
    41
    42      private TopicMessageStore messageStore;
    43      private String destinationName;
    44      private MessageIdentity lastMessageIdentity;
    45
    46      public DurableTopicMessageContainer(TopicMessageStore messageStore, String destinationName) {
    47          this.messageStore = messageStore;
    48          this.destinationName = destinationName;
    49      }
    50
    51      public String getDestinationName() {
    52          return destinationName;
    53      }
    54
    55      public MessageIdentity addMessage(ActiveMQMessage message) throws JMSException {
    56          MessageIdentity answer = messageStore.addMessage(message);
    57          lastMessageIdentity = answer;
    58          return answer;
    59      }
    60
    61      public void delete(MessageIdentity messageID, MessageAck ack) throws JMSException {
    62          // only called in MessagePointer and so shouldn't really delete
    63          //messageStore.removeMessage(new MessageIdentity(messageID));
    64      }
    65
    66      public boolean containsMessage(MessageIdentity messageIdentity) throws JMSException {
    67          /*** TODO: make more optimal implementation */
    68          return getMessage(messageIdentity) != null;
    69      }
    70
    71      public ActiveMQMessage getMessage(MessageIdentity messageID) throws JMSException {
    72          return messageStore.getMessage(messageID);
    73      }
    74
    75      public void registerMessageInterest(MessageIdentity messageIdentity) throws JMSException {
    76          messageStore.incrementMessageCount(messageIdentity);
    77      }
    78
    79      public void unregisterMessageInterest(MessageIdentity messageIdentity, MessageAck ack) throws JMSException {
    80          messageStore.decrementMessageCountAndMaybeDelete(messageIdentity, ack);
    81      }
    82
    83
    84      public void setLastAcknowledgedMessageID(Subscription subscription, MessageIdentity messageIdentity) throws JMSException {
    85          messageStore.setLastAcknowledgedMessageIdentity(subscription, messageIdentity);
    86      }
    87
    88      public void recoverSubscription(Subscription subscription) throws JMSException {
    89          messageStore.recoverSubscription(subscription, lastMessageIdentity);
    90      }
    91
    92
    93      public void start() throws JMSException {
    94          messageStore.setMessageContainer(this);
    95          lastMessageIdentity = messageStore.getLastestMessageIdentity();
    96          messageStore.start();
    97      }
    98
    99      public void stop() throws JMSException {
    100         messageStore.stop();
    101     }
    102 }