trunk/activemq-core/src/main/java/org/apache/activemq/kaha/impl/async/ReadOnlyAsyncDataManager.java

/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.activemq.kaha.impl.async;

import java.io.File;
import java.io.FilenameFilter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import org.apache.activemq.thread.Scheduler;
import org.apache.activemq.util.ByteSequence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * An AsyncDataManager that works in read only mode against multiple data directories.
 * Useful for reading back archived data files.
 */
public class ReadOnlyAsyncDataManager extends AsyncDataManager {
    
    private static final Logger LOG = LoggerFactory.getLogger(ReadOnlyAsyncDataManager.class);
    private final ArrayList<File> dirs;

    public ReadOnlyAsyncDataManager(final ArrayList<File> dirs) {
        this.dirs = dirs;
    }

    @SuppressWarnings("unchecked")
    public synchronized void start() throws IOException {
        if (started) {
            return;
        }

        started = true;

        accessorPool = new DataFileAccessorPool(this);
                
        ArrayList<File> files = new ArrayList<File>();
        for (File directory : dirs) {
            final File d = directory;
            File[] f = d.listFiles(new FilenameFilter() {
                public boolean accept(File dir, String n) {
                    return dir.equals(d) && n.startsWith(filePrefix);
                }
            });
            for (int i = 0; i < f.length; i++) {
                files.add(f[i]);
            }
        }
       
        for (File file : files) {
            try {
                String n = file.getName();
                String numStr = n.substring(filePrefix.length(), n.length());
                int num = Integer.parseInt(numStr);
                DataFile dataFile = new ReadOnlyDataFile(file, num, preferedFileLength);
                fileMap.put(dataFile.getDataFileId(), dataFile);
                storeSize.addAndGet(dataFile.getLength());
            } catch (NumberFormatException e) {
                // Ignore file that do not match the pattern.
            }
        }

        // Sort the list so that we can link the DataFiles together in the
        // right order.
        List<DataFile> dataFiles = new ArrayList<DataFile>(fileMap.values());
        Collections.sort(dataFiles);
        currentWriteFile = null;
        for (DataFile df : dataFiles) {
            if (currentWriteFile != null) {
                currentWriteFile.linkAfter(df);
            }
            currentWriteFile = df;
            fileByFileMap.put(df.getFile(), df);
        }
        
        // Need to check the current Write File to see if there was a partial
        // write to it.
        if (currentWriteFile != null) {

            // See if the lastSyncedLocation is valid..
            Location l = lastAppendLocation.get();
            if (l != null && l.getDataFileId() != currentWriteFile.getDataFileId().intValue()) {
                l = null;
            }
        }
    }
    
    public synchronized void close() throws IOException {
        if (!started) {
            return;
        }
        accessorPool.close();
        fileMap.clear();
        fileByFileMap.clear();
        started = false;
    }

    
    public Location getFirstLocation() throws IllegalStateException, IOException {
        if( currentWriteFile == null ) {
            return null;
        }
        
        DataFile first = (DataFile)currentWriteFile.getHeadNode();
        Location cur = new Location();
        cur.setDataFileId(first.getDataFileId());
        cur.setOffset(0);
        cur.setSize(0);
        return getNextLocation(cur);
    }
    
    @Override
    public synchronized boolean delete() throws IOException {
        throw new RuntimeException("Cannot delete a ReadOnlyAsyncDataManager");
    }    
}