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.RecordManager;
    21  import jdbm.RecordManagerFactory;
    22  import jdbm.btree.BTree;
    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.service.impl.PersistenceAdapterSupport;
    27  import org.codehaus.activemq.store.MessageStore;
    28  import org.codehaus.activemq.store.PreparedTransactionStore;
    29  import org.codehaus.activemq.store.TopicMessageStore;
    30  import org.codehaus.activemq.util.DefaultComparator;
    31  import org.codehaus.activemq.util.JMSExceptionHelper;
    32
    33  import javax.jms.JMSException;
    34  import java.io.File;
    35  import java.io.IOException;
    36  import java.util.Comparator;
    37  import java.util.Map;
    38  import java.util.Properties;
    39
    40  /***
    41   * A {@link org.codehaus.activemq.store.PersistenceAdapter} implementation for
    42   * <a href="http://jdbm.sf.net/">JDBM</a>
    43   *
    44   * @version $Revision: 1.6 $
    45   */
    46  public class JdbmPersistenceAdapter extends PersistenceAdapterSupport {
    47
    48      private static final Log log = LogFactory.getLog(JdbmPersistenceAdapter.class);
    49
    50      private RecordManager manager;
    51      private File directory = new File("ActiveMQ");
    52      private Properties properties;
    53
    54      /***
    55       * Factory method to create an instance using the defaults
    56       *
    57       * @param directory the directory in which to store the persistent files
    58       * @return
    59       * @throws JMSException
    60       */
    61      public static JdbmPersistenceAdapter newInstance(File directory) throws JMSException {
    62          return new JdbmPersistenceAdapter(directory);
    63      }
    64
    65
    66      public JdbmPersistenceAdapter() {
    67      }
    68
    69      public JdbmPersistenceAdapter(File directory) {
    70          this.directory = directory;
    71      }
    72
    73      public JdbmPersistenceAdapter(RecordManager manager) {
    74          this.manager = manager;
    75      }
    76
    77      public Map getInitialDestinations() {
    78          return null;  /*** TODO */
    79      }
    80
    81      public MessageStore createQueueMessageStore(String destinationName) throws JMSException {
    82          try {
    83              BTree messageDb = createDatabase("Queue_" + destinationName);
    84              BTree sequenceDb = createDatabase("Sequence_Queue_" + destinationName);
    85              JdbmMessageStore messageStore = new JdbmMessageStore(messageDb, sequenceDb);
    86              return messageStore;
    87          }
    88          catch (IOException e) {
    89              throw JMSExceptionHelper.newJMSException("Failed to create a QueueMessageContainer for destination: " + destinationName + ". Reason: " + e, e);
    90          }
    91      }
    92
    93      public TopicMessageStore createTopicMessageStore(String destinationName) throws JMSException {
    94          try {
    95              BTree messageDb = createDatabase("Topic_" + destinationName);
    96              BTree sequenceDb = createDatabase("Sequence_Topic_" + destinationName);
    97              BTree consumerAckDb = createDatabase("Consumer_Acks_Topic_" + destinationName);
    98              BTree subscriberDb = createDatabase("Subscriber_" + destinationName);
    99              BTree messageCountDb = createDatabase("MessageCount_Topic_" + destinationName);
    100             JdbmTopicMessageStore messageStore = new JdbmTopicMessageStore(messageDb, sequenceDb, consumerAckDb, subscriberDb, messageCountDb);
    101             return messageStore;
    102         }
    103         catch (IOException e) {
    104             throw JMSExceptionHelper.newJMSException("Failed to create a TopicMessageContainer for destination: " + destinationName + ". Reason: " + e, e);
    105         }
    106     }
    107
    108     public PreparedTransactionStore createPreparedTransactionStore() throws JMSException {
    109         try {
    110             return new JdbmPreparedTransactionStore(createDatabase("XaPrepareTxnDb"));
    111         }
    112         catch (IOException e) {
    113             throw JMSExceptionHelper.newJMSException("Could not create XA Prepare Transaction Database. Reason: " + e, e);
    114         }
    115     }
    116
    117     public void beginTransaction() {
    118     }
    119
    120     public void commitTransaction() throws JMSException {
    121         try {
    122             manager.commit();
    123         }
    124         catch (IOException e) {
    125             throw JMSExceptionHelper.newJMSException("Could not commit transaction. Reason: " + e, e);
    126         }
    127     }
    128
    129     public void rollbackTransaction() {
    130         try {
    131             manager.rollback();
    132         }
    133         catch (IOException e) {
    134             log.error("Could not rollback transaction. Reason: " + e, e);
    135         }
    136     }
    137
    138     public void start() throws JMSException {
    139         if (manager == null) {
    140             directory.mkdirs();
    141
    142             log.info("Creating JDBM based message store in directory: " + directory.getAbsolutePath());
    143
    144             try {
    145                 String name = directory.getAbsolutePath() + "/Store";
    146                 if (properties != null) {
    147                     manager = RecordManagerFactory.createRecordManager(name, properties);
    148                 }
    149                 else {
    150                     manager = RecordManagerFactory.createRecordManager(name);
    151                 }
    152             }
    153             catch (IOException e) {
    154                 throw JMSExceptionHelper.newJMSException("Failed to create JDBM persistent store at directory: "
    155                         + directory + ". Reason: " + e, e);
    156             }
    157         }
    158     }
    159
    160     public synchronized void stop() throws JMSException {
    161         if (manager != null) {
    162             try {
    163                 manager.close();
    164             }
    165             catch (IOException e) {
    166                 throw JMSExceptionHelper.newJMSException("Failed to close PersistenceAdapter. Reason: " + e, e);
    167             }
    168             finally {
    169                 manager = null;
    170             }
    171         }
    172     }
    173
    174     // Properties
    175     //-------------------------------------------------------------------------
    176     public RecordManager getManager() {
    177         return manager;
    178     }
    179
    180     public void setManager(RecordManager manager) {
    181         this.manager = manager;
    182     }
    183
    184     public File getDirectory() {
    185         return directory;
    186     }
    187
    188     public void setDirectory(File directory) {
    189         this.directory = directory;
    190     }
    191
    192     // Implementation methods
    193     //-------------------------------------------------------------------------
    194     public synchronized BTree createDatabase(String name) throws IOException, AlreadyClosedException {
    195         if (manager == null) {
    196             throw new AlreadyClosedException("JDBM PersistenceAdapter");
    197         }
    198
    199         // try to reload an existing B+Tree
    200         long recid = manager.getNamedObject(name);
    201         BTree tree = null;
    202         if (recid != 0) {
    203             tree = BTree.load(manager, recid);
    204         }
    205         else {
    206             Comparator comparator = new DefaultComparator();
    207             //Comparator comparator = new ObjectBAComparator(new DefaultComparator());
    208             tree = BTree.createInstance(manager, comparator);
    209             manager.setNamedObject(name, tree.getRecid());
    210         }
    211         return tree;
    212     }
    213
    214 }