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.store.jdbm;
    19
    20  import jdbm.btree.BTree;
    21  import jdbm.helper.Tuple;
    22  import jdbm.helper.TupleBrowser;
    23  import org.apache.commons.logging.Log;
    24  import org.apache.commons.logging.LogFactory;
    25  import org.codehaus.activemq.AlreadyClosedException;
    26  import org.codehaus.activemq.message.ActiveMQMessage;
    27  import org.codehaus.activemq.message.MessageAck;
    28  import org.codehaus.activemq.service.MessageContainer;
    29  import org.codehaus.activemq.service.MessageIdentity;
    30  import org.codehaus.activemq.service.QueueMessageContainer;
    31  import org.codehaus.activemq.service.impl.MessageEntry;
    32  import org.codehaus.activemq.store.MessageStore;
    33  import org.codehaus.activemq.util.JMSExceptionHelper;
    34
    35  import javax.jms.JMSException;
    36  import java.io.IOException;
    37
    38  /***
    39   * @version $Revision: 1.3 $
    40   */
    41  public class JdbmMessageStore implements MessageStore {
    42      private static final Log log = LogFactory.getLog(JdbmMessageStore.class);
    43
    44      private MessageContainer container;
    45      private BTree messageTable;
    46      private BTree orderedIndex;
    47      private long lastSequenceNumber = 0;
    48
    49      public JdbmMessageStore(BTree messageTable, BTree orderedIndex) {
    50          this.messageTable = messageTable;
    51          this.orderedIndex = orderedIndex;
    52      }
    53
    54      public void setMessageContainer(MessageContainer container) {
    55          this.container = container;
    56      }
    57
    58      public synchronized MessageIdentity addMessage(ActiveMQMessage message) throws JMSException {
    59          if (log.isDebugEnabled()) {
    60              log.debug("Adding message to container: " + message);
    61          }
    62          MessageEntry entry = new MessageEntry(message);
    63          Object sequenceNumber = null;
    64          synchronized (this) {
    65              sequenceNumber = new Long(++lastSequenceNumber);
    66          }
    67          try {
    68              String messageID = message.getJMSMessageID();
    69              getMessageTable().insert(messageID, entry, true);
    70              getOrderedIndex().insert(sequenceNumber, messageID, true);
    71
    72              MessageIdentity answer = message.getJMSMessageIdentity();
    73              answer.setSequenceNumber(sequenceNumber);
    74              return answer;
    75          }
    76          catch (IOException e) {
    77              throw JMSExceptionHelper.newJMSException("Failed to add message: " + message + " in container: " + e, e);
    78          }
    79      }
    80
    81      public synchronized ActiveMQMessage getMessage(MessageIdentity identity) throws JMSException {
    82          String messageID = identity.getMessageID();
    83          ActiveMQMessage message = null;
    84          try {
    85              MessageEntry entry = (MessageEntry) getMessageTable().find(messageID);
    86              if (entry != null) {
    87                  message = entry.getMessage();
    88                  message.getJMSMessageIdentity().setSequenceNumber(identity.getSequenceNumber());
    89              }
    90          }
    91          catch (IOException e) {
    92              throw JMSExceptionHelper.newJMSException("Failed to get message for messageID: " + messageID + " " + e, e);
    93          }
    94          return message;
    95      }
    96
    97      public synchronized void removeMessage(MessageIdentity identity, MessageAck ack) throws JMSException {
    98          String messageID = identity.getMessageID();
    99          if (messageID == null) {
    100             throw new JMSException("Cannot remove message with null messageID for sequence number: " + identity.getSequenceNumber());
    101         }
    102         try {
    103             Object sequenceNumber = identity.getSequenceNumber();
    104             if (sequenceNumber == null) {
    105                 sequenceNumber = findSequenceNumber(messageID);
    106                 identity.setSequenceNumber(sequenceNumber);
    107             }
    108             getMessageTable().remove(messageID);
    109             getOrderedIndex().remove(sequenceNumber);
    110         }
    111         catch (IOException e) {
    112             throw JMSExceptionHelper.newJMSException("Failed to delete message for messageID: " + messageID + " " + e, e);
    113         }
    114     }
    115
    116     public synchronized void recover(QueueMessageContainer container) throws JMSException {
    117         try {
    118             Tuple tuple = new Tuple();
    119             TupleBrowser iter = getOrderedIndex().browse();
    120             while (iter.getNext(tuple)) {
    121                 Long key = (Long) tuple.getKey();
    122                 MessageIdentity messageIdentity = null;
    123                 if (key != null) {
    124                     String messageID = (String) tuple.getValue();
    125                     if (messageID != null) {
    126                         messageIdentity = new MessageIdentity(messageID, key);
    127                     }
    128                 }
    129                 if (messageIdentity != null) {
    130                     container.recoverMessageToBeDelivered(messageIdentity);
    131                 }
    132                 else {
    133                     log.warn("Could not find message for sequenceNumber: " + key);
    134                 }
    135             }
    136         }
    137         catch (IOException e) {
    138             throw JMSExceptionHelper.newJMSException("Failed to recover the durable queue store. Reason: " + e, e);
    139         }
    140     }
    141
    142     public synchronized void start() throws JMSException {
    143         try {
    144             // lets iterate through all IDs from the
    145             Tuple tuple = new Tuple();
    146             Long lastSequenceNumber = null;
    147             TupleBrowser iter = getOrderedIndex().browse();
    148             while (iter.getNext(tuple)) {
    149                 lastSequenceNumber = (Long) tuple.getKey();
    150             }
    151             if (lastSequenceNumber != null) {
    152                 this.lastSequenceNumber = lastSequenceNumber.longValue();
    153                 if (log.isDebugEnabled()) {
    154                     log.debug("Last sequence number is: " + lastSequenceNumber + " for: " + this);
    155                 }
    156             }
    157             else {
    158                 if (log.isDebugEnabled()) {
    159                     log.debug("Started empty database for: " + this);
    160                 }
    161             }
    162         }
    163         catch (IOException e) {
    164             throw JMSExceptionHelper.newJMSException("Failed to find the last sequence number. Reason: " + e, e);
    165         }
    166     }
    167
    168     public synchronized void stop() throws JMSException {
    169         JMSException firstException = closeTable(orderedIndex, null);
    170         firstException = closeTable(messageTable, firstException);
    171         orderedIndex = null;
    172         messageTable = null;
    173         if (firstException != null) {
    174             throw firstException;
    175         }
    176     }
    177
    178
    179     // Implementation methods
    180     //-------------------------------------------------------------------------
    181
    182     protected MessageContainer getContainer() {
    183         return container;
    184     }
    185
    186     protected long getLastSequenceNumber() {
    187         return lastSequenceNumber;
    188     }
    189
    190     protected BTree getMessageTable() throws AlreadyClosedException {
    191         if (messageTable == null) {
    192             throw new AlreadyClosedException("JDBM MessageStore");
    193         }
    194         return messageTable;
    195     }
    196
    197     protected BTree getOrderedIndex() throws AlreadyClosedException {
    198         if (orderedIndex == null) {
    199             throw new AlreadyClosedException("JDBM MessageStore");
    200         }
    201         return orderedIndex;
    202     }
    203
    204
    205     /***
    206      * Looks up the message using the given sequence number
    207      */
    208     protected ActiveMQMessage getMessageBySequenceNumber(Long sequenceNumber) throws IOException, JMSException {
    209         ActiveMQMessage message = null;
    210         String messageID = (String) getOrderedIndex().find(sequenceNumber);
    211         if (messageID != null) {
    212             message = getMessage(new MessageIdentity(messageID, sequenceNumber));
    213         }
    214         return message;
    215     }
    216
    217     /***
    218      * Finds the sequence number for the given messageID
    219      *
    220      * @param messageID
    221      * @return
    222      */
    223     protected Object findSequenceNumber(String messageID) throws IOException, AlreadyClosedException {
    224         log.warn("Having to table scan to find the sequence number for messageID: " + messageID);
    225
    226         Tuple tuple = new Tuple();
    227         TupleBrowser iter = getOrderedIndex().browse();
    228         while (iter.getNext(tuple)) {
    229             Object value = tuple.getValue();
    230             if (messageID.equals(value)) {
    231                 return tuple.getKey();
    232             }
    233         }
    234         return null;
    235     }
    236
    237     protected JMSException closeTable(BTree table, JMSException firstException) {
    238         table = null;
    239         return null;
    240     }
    241 }