package org.dbxml.core.filer;

/*
 * dbXML License, Version 1.0
 *
 *
 * Copyright (c) 1999-2001 The dbXML Group, L.L.C.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 *
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in
 *    the documentation and/or other materials provided with the
 *    distribution.
 *
 * 3. The end-user documentation included with the redistribution,
 *    if any, must include the following acknowledgment:
 *       "This product includes software developed by
 *        The dbXML Group (http://www.dbxml.com/)."
 *    Alternately, this acknowledgment may appear in the software
 *    itself, if and wherever such third-party acknowledgments normally
 *    appear.
 *
 * 4. The names "dbXML" and "The dbXML Group" must not be used to
 *    endorse or promote products derived from this software without
 *    prior written permission. For written permission, please contact
 *    info@dbxml.com.
 *
 * 5. Products derived from this software may not be called "dbXML",
 *    nor may "dbXML" appear in their name, without prior written
 *    permission of The dbXML Group.
 *
 * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED
 * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED.  IN NO EVENT SHALL THE DBXML GROUP OR ITS CONTRIBUTORS
 * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY,
 * OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT
 * OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
 * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 * $Id: HashFiler.java,v 1.30 2001/07/22 21:35:45 tom Exp $
 */

import org.dbxml.server.util.*;
import org.dbxml.server.*;
import org.dbxml.core.*;
import org.dbxml.core.Collection;
import org.dbxml.core.data.*;

import java.io.*;
import java.util.*;

/**
 * HashFiler is a Filer implementation based on the Paged class.
 * By extending Paged, HashFiler inherits the ability to maintain
 * Record metadata such as creation and modification time.  It also
 * provides quite a bit more flexibility in its ability to retreive
 * blocks of data and allocate Record space.
 * <br><br>
 * NOTE: This class has been temporarily (maybe indefinitely)
 * deprecated by BTreeFiler.
 */

public final class HashFiler extends Paged implements Configurable, Filer {
   protected static final byte RECORD = 1;

   private static final String PAGECOUNT = "pagecount";
   private static final String PAGESIZE = "pagesize";
   private static final String MAXKEYSIZE = "maxkeysize";
   private static final String MAXKEYLOCKS = "maxkeylocks";

   private Collection collection;
   private Configuration config;
   private LockManager keyLocks;
   private HashFileHeader fileHeader;
   private BTree btree;

   public HashFiler() {
      super();
      fileHeader = (HashFileHeader)getFileHeader();
   }

   public void setConfig(Configuration config) {
      this.config = config;
      keyLocks = new LockManager(config.getIntAttribute(MAXKEYLOCKS, 16));
   }

   public Configuration getConfig() {
      return config;
   }

   public void setLocation(String location) {
      setFile(new File(collection.getCollectionRoot(), location+".tbl"));
      btree = new BTree(new File(collection.getCollectionRoot(), location+".pkx"));
   }

   public String getName() {
      return "HashFiler";
   }

   public boolean open() throws DBException {
      if ( super.open() ) {
         // These are the only properties that can be changed after creation
         fileHeader.setMaxKeySize(config.getShortAttribute(MAXKEYSIZE, fileHeader.getMaxKeySize()));
         btree.open();
         return true;
      }
      else
         return false;
   }

   public boolean create() throws DBException {
      fileHeader.setPageCount(config.getLongAttribute(PAGECOUNT, fileHeader.getPageCount()));
      fileHeader.setPageSize(config.getIntAttribute(PAGESIZE, fileHeader.getPageSize()));
      fileHeader.setMaxKeySize(config.getShortAttribute(MAXKEYSIZE, fileHeader.getMaxKeySize()));
      btree.create();
      return super.create();
   }

   public void setCollection(Collection collection) {
      this.collection = collection;
      setLocation(collection.getName());
   }

   private Page seekRecordPage(Key key) throws IOException {
      int hash = key.getHash();
      long pageNum = hash % fileHeader.getPageCount();
      Page p = null;
      HashPageHeader ph = null;
      while ( true ) {
         p = getPage(pageNum);
         ph = (HashPageHeader)p.getPageHeader();
         if ( ph.getStatus() == RECORD
           && ph.getKeyHash() == key.getHash()
           && p.getKey().equals(key) )
            return p;
         pageNum = ph.getNextCollision();
         if ( pageNum == -1 )
            return null;
      }
   }

   private Page seekInsertionPage(Key key) throws IOException {
      int hash = key.getHash();
      long pageNum = hash % fileHeader.getPageCount();
      Page p = null;
      HashPageHeader ph = null;
      while ( true ) {
         p = getPage(pageNum);
         ph = (HashPageHeader)p.getPageHeader();
         if ( ( ph.getStatus() == UNUSED || ph.getStatus() == DELETED )
           || ( ph.getStatus() == RECORD
             && ph.getKeyHash() == key.getHash()
             && p.getKey().equals(key) ) )
            return p;
         pageNum = ph.getNextCollision();
         if ( pageNum == -1 ) {
            Page np = getFreePage();
            ph.setNextCollision(np.getPageNum());
            p.write();
            return np;
         }
      }
   }

   public Record readRecord(Key key) throws DBException {
      checkOpened();
      keyLocks.acquireSharedLock(key);
      try {
         Page startPage = seekRecordPage(key);
         if ( startPage != null ) {
            Value v = readValue(startPage);
            HashPageHeader sph = (HashPageHeader)startPage.getPageHeader();
            
            HashMap meta = new HashMap(2);
            meta.put(Record.CREATED, new Long(sph.getCreated()));
            meta.put(Record.MODIFIED, new Long(sph.getModified()));
            
            return new Record(key, v, meta);
         }
      }
      catch ( Exception e ) {
         org.dbxml.Debug.printStackTrace(e);
      }
      finally {
         keyLocks.releaseSharedLock(key);
      }
      return null;
   }

   public boolean writeRecord(Key key, Value value) throws DBException {
      checkOpened();
      keyLocks.acquireExclusiveLock(key);
      try {
         Page p = seekInsertionPage(key);
         HashPageHeader ph = (HashPageHeader)p.getPageHeader();
         
         long t = System.currentTimeMillis();
         if ( ph.getStatus() == UNUSED ) {
            // This is a new Record
            fileHeader.incRecordCount();
            btree.addValue(key, p.getPageNum());
            ph.setCreated(t);
         }
         
         ph.setModified(t);
         ph.setStatus(RECORD);

         p.setKey(key);
         writeValue(p, value);
         fileHeader.write();
      }
      catch ( Exception e ) {
         org.dbxml.Debug.printStackTrace(e);
      }
      finally {
         keyLocks.releaseExclusiveLock(key);
      }
      return true;
   }

   public boolean deleteRecord(Key key) throws DBException {
      checkOpened();
      keyLocks.acquireExclusiveLock(key);
      try {
         int hash = key.getHash();
         long pageNum = hash % fileHeader.getPageCount();
         Page prev = null;
         Page page = null;
         
         HashPageHeader prevHead = null;
         HashPageHeader pageHead = null;
         
         while ( true ) {
            page = getPage(pageNum);
            pageHead = (HashPageHeader)page.getPageHeader();
            if ( pageHead.getStatus() == RECORD
              && pageHead.getKeyHash() == key.getHash()
              && page.getKey().equals(key) )
               break;
            pageNum = pageHead.getNextCollision();
            if ( pageNum == -1 )
               return false;
            prev = page;
            prevHead = pageHead;
         }

         if ( page == null )
            return false;

         if ( prev != null ) {
            prevHead.setNextCollision(pageHead.nextCollision);
            prev.write();
         }
         
         btree.removeValue(key);
         unlinkPages(page);

         fileHeader.decRecordCount();
         fileHeader.write();
         
         return true;
      }
      catch ( Exception e ) {
         org.dbxml.Debug.printStackTrace(e);
      }
      finally {
         keyLocks.releaseExclusiveLock(key);
      }
      return false;
   }

   public long getRecordCount() throws DBException {
      checkOpened();
      return fileHeader.getRecordCount();
   }

   public RecordSet getRecordSet() throws DBException {
      checkOpened();
      return new HashFilerRecordSet();
   }

   private class HashFilerRecordSet implements RecordSet, BTreeCallback {
      private List keys = new ArrayList();
      private Iterator enum;

      public HashFilerRecordSet() {
         try {
            btree.query(null, this);
            enum = keys.iterator();
         }
         catch ( Exception e ) {
            org.dbxml.Debug.printStackTrace(e);
         }
      }

      public boolean indexInfo(Value value, long pointer) {
         keys.add(new Key(value));
         return true;
      }

      public Key getNextKey() {
         return (Key)enum.next();
      }

      public Record getNextRecord() throws DBException {
         return readRecord((Key)enum.next());
      }

      public Value getNextValue() throws DBException {
         return getNextRecord().getValue();
      }

      public boolean hasMoreRecords() {
         return enum.hasNext();
      }
   }

   ////////////////////////////////////////////////////////////////////

   public FileHeader createFileHeader() {
      return new HashFileHeader();
   }

   public FileHeader createFileHeader(boolean read) throws IOException {
      return new HashFileHeader(read);
   }

   public FileHeader createFileHeader(long pageCount) {
      return new HashFileHeader(pageCount);
   }

   public FileHeader createFileHeader(long pageCount, int pageSize) {
      return new HashFileHeader(pageCount, pageSize);
   }

   public PageHeader createPageHeader() {
      return new HashPageHeader();
   }


   /**
    * HashFileHeader
    */

   private final class HashFileHeader extends FileHeader {
      private long totalBytes = 0;

      public HashFileHeader() {
      }

      public HashFileHeader(long pageCount) {
         super(pageCount);
      }

      public HashFileHeader(long pageCount, int pageSize) {
         super(pageCount, pageSize);
      }

      public HashFileHeader(boolean read) throws IOException {
         super(read);
      }

      public void read(RandomAccessFile raf) throws IOException {
         super.read(raf);
         totalBytes = raf.readLong();
      }

      public void write(RandomAccessFile raf) throws IOException {
         super.write(raf);
         raf.writeLong(totalBytes);
      }

      /** The total number of bytes in use by the file */
      public void setTotalBytes(long totalBytes) {
         this.totalBytes = totalBytes;
         setDirty();
      }

      /** The total number of bytes in use by the file */
      public long getTotalBytes() {
         return totalBytes;
      }
   }

   /**
    * HashPageHeader
    */

   private final class HashPageHeader extends PageHeader {
      private long created = 0;
      private long modified = 0;
      private long nextCollision = -1;

      public HashPageHeader() {
      }

      public HashPageHeader(DataInputStream dis) throws IOException {
         super(dis);
      }

      public void read(DataInputStream dis) throws IOException {
         super.read(dis);

         if ( getStatus() == UNUSED )
            return;

         created = dis.readLong();
         modified = dis.readLong();
         nextCollision = dis.readLong();
      }

      public void write(DataOutputStream dos) throws IOException {
         super.write(dos);
         dos.writeLong(created);
         dos.writeLong(modified);
         dos.writeLong(nextCollision);
      }

      public void setRecordLen(int recordLen) {
         synchronized(fileHeader) {
            fileHeader.setTotalBytes((fileHeader.totalBytes - getRecordLen()) + recordLen);
         }
         super.setRecordLen(recordLen);
      }

      /** UNIX-time when this record was created */
      public void setCreated(long created) {
         this.created = created;
         setDirty();
      }

      /** UNIX-time when this record was created */
      public long getCreated() {
         return created;
      }

      /** UNIX-time when this record was last modified */
      public void setModified(long modified) {
         this.modified = modified;
         setDirty();
      }

      /** UNIX-time when this record was last modified */
      public long getModified() {
         return modified;
      }

      /** The next page for a Record collision (if any) */
      public void setNextCollision(long nextCollision) {
         this.nextCollision = nextCollision;
         setDirty();
      }

      /** The next page for a Record collision (if any) */
      public long getNextCollision() {
         return nextCollision;
      }
   }
}









