content: /**
   * 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.store.kahadb.disk.journal;
  
  import java.io.IOException;
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  
  /**
   * Used to pool DataFileAccessors.
   *
   * @author chirino
   */
  public class DataFileAccessorPool {
  
      private final Journal journal;
      private final Map<Integer, Pool> pools = new HashMap<Integer, Pool>();
      private boolean closed;
      private int maxOpenReadersPerFile = 5;
  
      class Pool {
  
          private final DataFile file;
          private final List<DataFileAccessor> pool = new ArrayList<DataFileAccessor>();
          private int openCounter;
          private boolean disposed;
  
          public Pool(DataFile file) {
              this.file = file;
          }
  
          public DataFileAccessor openDataFileReader() throws IOException {
              DataFileAccessor rc = null;
              if (pool.isEmpty()) {
                  rc = new DataFileAccessor(journal, file);
              } else {
                  rc = pool.remove(pool.size() - 1);
              }
              openCounter++;
              return rc;
          }
  
          public synchronized void closeDataFileReader(DataFileAccessor reader) {
              openCounter--;
              if (pool.size() >= maxOpenReadersPerFile || disposed) {
                  reader.dispose();
              } else {
                  pool.add(reader);
              }
          }
  
          public synchronized boolean isUsed() {
              return openCounter > 0;
          }
  
          public synchronized void dispose() {
              for (DataFileAccessor reader : pool) {
                  reader.dispose();
              }
              pool.clear();
              disposed = true;
          }
  
          public synchronized int getOpenCounter() {
              return openCounter;
          }
  
      }
  
      public DataFileAccessorPool(Journal dataManager) {
          this.journal = dataManager;
      }
  
      public synchronized int size() {
          return pools.size();
      }
  
      public synchronized void disposeUnused() {
          for (Iterator<Pool> iter = pools.values().iterator(); iter.hasNext();) {
              Pool pool = iter.next();
              if (!pool.isUsed()) {
                  pool.dispose();
                  iter.remove();
              }
          }
      }
  
      synchronized void disposeDataFileAccessors(DataFile dataFile) {
          if (closed) {
              throw new IllegalStateException("Closed.");
          }
          Pool pool = pools.get(dataFile.getDataFileId());
          if (pool != null) {
              if (pool.getOpenCounter() == 0) {
                  pool.dispose();
                  pools.remove(dataFile.getDataFileId());
              } else {
                  throw new IllegalStateException("The data file is still in use: " + dataFile + ", use count: " + pool.getOpenCounter());
              }
          }
      }
  
      synchronized DataFileAccessor openDataFileAccessor(DataFile dataFile) throws IOException {
          if (closed) {
              throw new IOException("Closed.");
          }
  
          Pool pool = pools.get(dataFile.getDataFileId());
          if (pool == null) {
              pool = new Pool(dataFile);
              pools.put(dataFile.getDataFileId(), pool);
          }
          return pool.openDataFileReader();
      }
  
      synchronized void closeDataFileAccessor(DataFileAccessor reader) {
          Pool pool = pools.get(reader.getDataFile().getDataFileId());
          if (pool == null || closed) {
              reader.dispose();
          } else {
              pool.closeDataFileReader(reader);
          }
      }
  
      public synchronized void close() {
          if (closed) {
              return;
          }
          closed = true;
          for (Pool pool : pools.values()) {
              pool.dispose();
          }
          pools.clear();
      }
  
  }
  