1   /**
    2    * Licensed to the Apache Software Foundation (ASF) under one or more
    3    * contributor license agreements.  See the NOTICE file distributed with
    4    * this work for additional information regarding copyright ownership.
    5    * The ASF licenses this file to You under the Apache License, Version 2.0
    6    * (the "License"); you may not use this file except in compliance with
    7    * the License.  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   package org.apache.activemq.kaha.impl.data;
   18   
   19   import java.io.IOException;
   20   import java.io.RandomAccessFile;
   21   import org.apache.activemq.kaha.Marshaller;
   22   import org.apache.activemq.kaha.StoreLocation;
   23   import org.apache.activemq.util.DataByteArrayInputStream;
   24   
   25   /**
   26    * Optimized Store reader
   27    * 
   28    * @version $Revision: 1.1.1.1 $
   29    */
   30   public final class SyncDataFileReader {
   31   
   32       private DataManagerImpl dataManager;
   33       private DataByteArrayInputStream dataIn;
   34   
   35       /**
   36        * Construct a Store reader
   37        * 
   38        * @param fileId
   39        */
   40       SyncDataFileReader(DataManagerImpl fileManager) {
   41           this.dataManager = fileManager;
   42           this.dataIn = new DataByteArrayInputStream();
   43       }
   44   
   45       /*
   46        * (non-Javadoc)
   47        * 
   48        * @see org.apache.activemq.kaha.impl.data.DataFileReader#readDataItemSize(org.apache.activemq.kaha.impl.data.DataItem)
   49        */
   50       public synchronized byte readDataItemSize(DataItem item) throws IOException {
   51           RandomAccessFile file = dataManager.getDataFile(item).getRandomAccessFile();
   52           file.seek(item.getOffset()); // jump to the size field
   53           byte rc = file.readByte();
   54           item.setSize(file.readInt());
   55           return rc;
   56       }
   57   
   58       /*
   59        * (non-Javadoc)
   60        * 
   61        * @see org.apache.activemq.kaha.impl.data.DataFileReader#readItem(org.apache.activemq.kaha.Marshaller,
   62        *      org.apache.activemq.kaha.StoreLocation)
   63        */
   64       public synchronized Object readItem(Marshaller marshaller, StoreLocation item) throws IOException {
   65           RandomAccessFile file = dataManager.getDataFile(item).getRandomAccessFile();
   66   
   67           // TODO: we could reuse the buffer in dataIn if it's big enough to avoid
   68           // allocating byte[] arrays on every readItem.
   69           byte[] data = new byte[item.getSize()];
   70           file.seek(item.getOffset() + DataManagerImpl.ITEM_HEAD_SIZE);
   71           file.readFully(data);
   72           dataIn.restart(data);
   73           return marshaller.readPayload(dataIn);
   74       }
   75   }