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.howl;
    19
    20  import org.apache.commons.logging.Log;
    21  import org.apache.commons.logging.LogFactory;
    22  import org.codehaus.activemq.message.DefaultWireFormat;
    23  import org.codehaus.activemq.service.impl.PersistenceAdapterSupport;
    24  import org.codehaus.activemq.store.MessageStore;
    25  import org.codehaus.activemq.store.PersistenceAdapter;
    26  import org.codehaus.activemq.store.PreparedTransactionStore;
    27  import org.codehaus.activemq.store.TopicMessageStore;
    28  import org.codehaus.activemq.store.jdbm.JdbmPersistenceAdapter;
    29  import org.codehaus.activemq.util.JMSExceptionHelper;
    30  import org.objectweb.howl.log.Configuration;
    31  import org.objectweb.howl.log.LogConfigurationException;
    32  import org.objectweb.howl.log.Logger;
    33
    34  import javax.jms.JMSException;
    35  import java.io.File;
    36  import java.io.IOException;
    37  import java.io.InputStream;
    38  import java.util.Properties;
    39
    40  /***
    41   * An implementation of {@link PersistenceAdapter} designed for
    42   * optimal use with <a href="http://howl.objectweb.org/">Howl</a>
    43   * as the transaction log and then checkpointing asynchronously
    44   * on a timeout with some other persistent storage.
    45   *
    46   * @version $Revision: 1.6 $
    47   */
    48  public class HowlPersistenceAdapter extends PersistenceAdapterSupport {
    49
    50      private static final Log log = LogFactory.getLog(HowlPersistenceAdapter.class);
    51
    52      private PersistenceAdapter longTermPersistence;
    53      private Configuration configuration;
    54      private int maximumTotalCachedMessages = 10000;
    55      private int maximumCachedMessagesPerStore = 100;
    56      private int cachedMessageCount;
    57      private File directory;
    58      private Logger transactionLog;
    59
    60      /***
    61       * Factory method to create an instance using the defaults
    62       *
    63       * @param directory the directory in which to store the persistent files
    64       * @return
    65       * @throws JMSException
    66       */
    67      public static HowlPersistenceAdapter newInstance(File directory) throws JMSException {
    68          return new HowlPersistenceAdapter(directory, JdbmPersistenceAdapter.newInstance(directory));
    69      }
    70
    71      public HowlPersistenceAdapter() {
    72      }
    73
    74      public HowlPersistenceAdapter(File directory, PersistenceAdapter longTermPersistence) {
    75          this.directory = directory;
    76          this.longTermPersistence = longTermPersistence;
    77      }
    78
    79      public MessageStore createQueueMessageStore(String destinationName) throws JMSException {
    80          MessageStore checkpointStore = longTermPersistence.createQueueMessageStore(destinationName);
    81          return new HowlMessageStore(this, checkpointStore, transactionLog, new DefaultWireFormat());
    82      }
    83
    84      public TopicMessageStore createTopicMessageStore(String destinationName) throws JMSException {
    85          /*** TODO not yet implemented for topics */
    86          return longTermPersistence.createTopicMessageStore(destinationName);
    87      }
    88
    89      public PreparedTransactionStore createPreparedTransactionStore() throws JMSException {
    90          // delegate to long term store
    91          return longTermPersistence.createPreparedTransactionStore();
    92      }
    93
    94      public void beginTransaction() throws JMSException {
    95      }
    96
    97      public void commitTransaction() throws JMSException {
    98      }
    99
    100     public void rollbackTransaction() {
    101     }
    102
    103     public void start() throws JMSException {
    104         if (transactionLog == null) {
    105             if (directory != null) {
    106                 directory.mkdirs();
    107             }
    108             try {
    109                 transactionLog = createTransactionLog();
    110             }
    111             catch (Exception e) {
    112                 throw JMSExceptionHelper.newJMSException("Failed to create Howl based message store due to: " + e, e);
    113             }
    114         }
    115
    116         try {
    117             log.info("Using Howl transaction log in directory: " + getLogFileDir());
    118
    119             transactionLog.open();
    120         }
    121         catch (Exception e) {
    122             throw JMSExceptionHelper.newJMSException("Failed to open Howl transaction log: " + e, e);
    123         }
    124         longTermPersistence.start();
    125     }
    126
    127     public void stop() throws JMSException {
    128         try {
    129             transactionLog.close();
    130         }
    131         catch (Exception e) {
    132             throw JMSExceptionHelper.newJMSException("Failed to close Howl transaction log due to: " + e, e);
    133         }
    134     }
    135
    136     /***
    137      * Return true if a store is allowed to cache a message.
    138      * Called by a store when its about to store a message in its cache.
    139      *
    140      * @param messageStore
    141      * @return true if the cache is allowed to cache the mesage
    142      */
    143     public synchronized boolean hasCacheCapacity(HowlMessageStore messageStore) {
    144         if (cachedMessageCount < maximumTotalCachedMessages) {
    145             cachedMessageCount++;
    146             return true;
    147         }
    148         return false;
    149     }
    150
    151     public synchronized void onMessageRemove(HowlMessageStore messageStore) {
    152         cachedMessageCount--;
    153     }
    154
    155     // Properties
    156     //-------------------------------------------------------------------------
    157     public PersistenceAdapter getLongTermPersistence() {
    158         return longTermPersistence;
    159     }
    160
    161     public void setLongTermPersistence(PersistenceAdapter longTermPersistence) {
    162         this.longTermPersistence = longTermPersistence;
    163     }
    164
    165     public int getMaximumCachedMessagesPerStore() {
    166         return maximumCachedMessagesPerStore;
    167     }
    168
    169     public void setMaximumCachedMessagesPerStore(int maximumCachedMessagesPerStore) {
    170         this.maximumCachedMessagesPerStore = maximumCachedMessagesPerStore;
    171     }
    172
    173     public int getMaximumTotalCachedMessages() {
    174         return maximumTotalCachedMessages;
    175     }
    176
    177     public void setMaximumTotalCachedMessages(int maximumTotalCachedMessages) {
    178         this.maximumTotalCachedMessages = maximumTotalCachedMessages;
    179     }
    180
    181     public File getDirectory() {
    182         return directory;
    183     }
    184
    185     public void setDirectory(File directory) {
    186         this.directory = directory;
    187     }
    188
    189     public Configuration getConfiguration() throws LogConfigurationException, IOException {
    190         if (configuration == null) {
    191             configuration = createConfiguration();
    192         }
    193         return configuration;
    194     }
    195
    196     public void setConfiguration(Configuration configuration) {
    197         this.configuration = configuration;
    198     }
    199
    200     public Logger getTransactionLog() {
    201         return transactionLog;
    202     }
    203
    204     public void setTransactionLog(Logger transactionLog) {
    205         this.transactionLog = transactionLog;
    206     }
    207
    208
    209     // Delegate Howl configuration properties
    210     //-------------------------------------------------------------------------
    211     public String getBufferClassName() throws LogConfigurationException, IOException {
    212         return getConfiguration().getBufferClassName();
    213     }
    214
    215     public int getBufferSize() throws LogConfigurationException, IOException {
    216         return getConfiguration().getBufferSize();
    217     }
    218
    219     public int getFlushSleepTime() throws LogConfigurationException, IOException {
    220         return getConfiguration().getFlushSleepTime();
    221     }
    222
    223     public String getLogFileDir() throws LogConfigurationException, IOException {
    224         return getConfiguration().getLogFileDir();
    225     }
    226
    227     public String getLogFileExt() throws LogConfigurationException, IOException {
    228         return getConfiguration().getLogFileExt();
    229     }
    230
    231     public String getLogFileName() throws LogConfigurationException, IOException {
    232         return getConfiguration().getLogFileName();
    233     }
    234
    235     public int getMaxBlocksPerFile() throws LogConfigurationException, IOException {
    236         return getConfiguration().getMaxBlocksPerFile();
    237     }
    238
    239     public int getMaxBuffers() throws LogConfigurationException, IOException {
    240         return getConfiguration().getMaxBuffers();
    241     }
    242
    243     public int getMaxLogFiles() throws LogConfigurationException, IOException {
    244         return getConfiguration().getMaxLogFiles();
    245     }
    246
    247     public int getMinBuffers() throws LogConfigurationException, IOException {
    248         return getConfiguration().getMinBuffers();
    249     }
    250
    251     public int getThreadsWaitingForceThreshold() throws LogConfigurationException, IOException {
    252         return getConfiguration().getThreadsWaitingForceThreshold();
    253     }
    254
    255     public boolean isChecksumEnabled() throws LogConfigurationException, IOException {
    256         return getConfiguration().isChecksumEnabled();
    257     }
    258
    259     public void setBufferClassName(String s) throws LogConfigurationException, IOException {
    260         getConfiguration().setBufferClassName(s);
    261     }
    262
    263     public void setBufferSize(int i) throws LogConfigurationException, IOException {
    264         getConfiguration().setBufferSize(i);
    265     }
    266
    267     public void setChecksumEnabled(boolean b) throws LogConfigurationException, IOException {
    268         getConfiguration().setChecksumEnabled(b);
    269     }
    270
    271     public void setFlushSleepTime(int i) throws LogConfigurationException, IOException {
    272         getConfiguration().setFlushSleepTime(i);
    273     }
    274
    275     public void setLogFileDir(String s) throws LogConfigurationException, IOException {
    276         getConfiguration().setLogFileDir(s);
    277     }
    278
    279     public void setLogFileExt(String s) throws LogConfigurationException, IOException {
    280         getConfiguration().setLogFileExt(s);
    281     }
    282
    283     public void setLogFileName(String s) throws LogConfigurationException, IOException {
    284         getConfiguration().setLogFileName(s);
    285     }
    286
    287     public void setMaxBlocksPerFile(int i) throws LogConfigurationException, IOException {
    288         getConfiguration().setMaxBlocksPerFile(i);
    289     }
    290
    291     public void setMaxBuffers(int i) throws LogConfigurationException, IOException {
    292         getConfiguration().setMaxBuffers(i);
    293     }
    294
    295     public void setMaxLogFiles(int i) throws LogConfigurationException, IOException {
    296         getConfiguration().setMaxLogFiles(i);
    297     }
    298
    299     public void setMinBuffers(int i) throws LogConfigurationException, IOException {
    300         getConfiguration().setMinBuffers(i);
    301     }
    302
    303     public void setThreadsWaitingForceThreshold(int i) throws LogConfigurationException, IOException {
    304         getConfiguration().setThreadsWaitingForceThreshold(i);
    305     }
    306
    307
    308     // Implementation methods
    309     //-------------------------------------------------------------------------
    310
    311     protected Logger createTransactionLog() throws IOException, LogConfigurationException {
    312         return new Logger(getConfiguration());
    313     }
    314
    315     protected Configuration createConfiguration() throws IOException, LogConfigurationException {
    316         String[] names = {"org/codehaus/activemq/howl.properties", "org/codehaus/activemq/defaultHowl.properties"};
    317
    318         Configuration answer = null;
    319         for (int i = 0; i < names.length; i++) {
    320             InputStream in = Thread.currentThread().getContextClassLoader().getResourceAsStream(names[i]);
    321             if (in == null) {
    322                 in = getClass().getClassLoader().getResourceAsStream(names[i]);
    323             }
    324             if (in != null) {
    325                 Properties properties = new Properties();
    326                 properties.load(in);
    327                 answer = new Configuration(properties);
    328             }
    329         }
    330         if (answer == null) {
    331             log.warn("Could not find file: " + names[0] + " or: " + names[1] + " on the classpath to initialise Howl");
    332             answer = new Configuration();
    333         }
    334         if (directory != null) {
    335             answer.setLogFileDir(directory.getAbsolutePath());
    336         }
    337         return answer;
    338     }
    339 }