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.index;
  
  import java.io.IOException;
  import java.lang.ref.WeakReference;
  import java.util.Iterator;
  import java.util.Map;
  import java.util.Map.Entry;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.concurrent.atomic.AtomicLong;
  
  import org.apache.activemq.store.kahadb.disk.index.ListNode.ListIterator;
  import org.apache.activemq.store.kahadb.disk.page.Page;
  import org.apache.activemq.store.kahadb.disk.page.PageFile;
  import org.apache.activemq.store.kahadb.disk.page.Transaction;
  import org.apache.activemq.store.kahadb.disk.util.Marshaller;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  public class ListIndex<Key,Value> implements Index<Key,Value> {
  
      private static final Logger LOG = LoggerFactory.getLogger(ListIndex.class);
      public  final static long NOT_SET = -1;
      protected PageFile pageFile;
      protected long headPageId;
      protected long tailPageId;
      private AtomicLong size = new AtomicLong(0);
  
      protected AtomicBoolean loaded = new AtomicBoolean();
  
      private ListNode.NodeMarshaller<Key, Value> marshaller;
      private Marshaller<Key> keyMarshaller;
      private Marshaller<Value> valueMarshaller;
  
      public ListIndex() {
      }
  
      public ListIndex(PageFile pageFile, long headPageId) {
          this.pageFile = pageFile;
          setHeadPageId(headPageId);
      }
  
      @SuppressWarnings("rawtypes")
      public ListIndex(PageFile pageFile, Page page) {
          this(pageFile, page.getPageId());
      }
  
      @Override
      synchronized public void load(Transaction tx) throws IOException {
          if (loaded.compareAndSet(false, true)) {
              LOG.debug("loading");
              if( keyMarshaller == null ) {
                  throw new IllegalArgumentException("The key marshaller must be set before loading the ListIndex");
              }
              if( valueMarshaller == null ) {
                  throw new IllegalArgumentException("The value marshaller must be set before loading the ListIndex");
              }
  
              marshaller = new ListNode.NodeMarshaller<Key, Value>(keyMarshaller, valueMarshaller);
              final Page<ListNode<Key,Value>> p = tx.load(getHeadPageId(), null);
              if( p.getType() == Page.PAGE_FREE_TYPE ) {
                   // Need to initialize it..
                  ListNode<Key, Value> root = createNode(p);
                  storeNode(tx, root, true);
                  setHeadPageId(p.getPageId());
                  setTailPageId(getHeadPageId());
              } else {
                  ListNode<Key, Value> node = loadNode(tx, getHeadPageId());
                  setTailPageId(getHeadPageId());
                  size.addAndGet(node.size(tx));
                  onLoad(node, tx);
                  while (node.getNext() != NOT_SET ) {
                      node = loadNode(tx, node.getNext());
                      size.addAndGet(node.size(tx));
                      onLoad(node, tx);
                      setTailPageId(node.getPageId());
                  }
              }
          }
      }
  
      protected void onLoad(ListNode<Key, Value> node, Transaction tx) {
  
      }
  
      @Override
      synchronized public void unload(Transaction tx) {
          if (loaded.compareAndSet(true, false)) {
          }
      }
  
      protected ListNode<Key,Value> getHead(Transaction tx) throws IOException {
          return loadNode(tx, getHeadPageId());
      }
  
      protected ListNode<Key,Value> getTail(Transaction tx) throws IOException {
          return loadNode(tx, getTailPageId());
      }
  
      @Override
      synchronized public boolean containsKey(Transaction tx, Key key) throws IOException {
          assertLoaded();
  
          if (size.get() == 0) {
              return false;
          }
  
          for (Iterator<Map.Entry<Key,Value>> iterator = iterator(tx); iterator.hasNext(); ) {
              Map.Entry<Key,Value> candidate = iterator.next();
              if (key.equals(candidate.getKey())) {
                  return true;
              }
          }
          return false;
      }
  
      private ListNode<Key, Value> lastGetNodeCache = null;
      private Map.Entry<Key, Value> lastGetEntryCache = null;
      private WeakReference<Transaction> lastCacheTxSrc = new WeakReference<Transaction>(null);
  
      @Override
      @SuppressWarnings({ "rawtypes", "unchecked" })
      synchronized public Value get(Transaction tx, Key key) throws IOException {
          assertLoaded();
          for (Iterator<Map.Entry<Key,Value>> iterator = iterator(tx); iterator.hasNext(); ) {
              Map.Entry<Key,Value> candidate = iterator.next();
              if (key.equals(candidate.getKey())) {
                  this.lastGetNodeCache = ((ListIterator) iterator).getCurrent();
                  this.lastGetEntryCache = candidate;
                  this.lastCacheTxSrc = new WeakReference<Transaction>(tx);
                  return candidate.getValue();
              }
          }
          return null;
      }
  
      /**
       * Update the value of the item with the given key in the list if ot exists, otherwise
       * it appends the value to the end of the list.
       *
       * @return the old value contained in the list if one exists or null.
       */
      @Override
      @SuppressWarnings({ "rawtypes" })
      synchronized public Value put(Transaction tx, Key key, Value value) throws IOException {
  
          Value oldValue = null;
  
          if (lastGetNodeCache != null && tx.equals(lastCacheTxSrc.get())) {
  
              if(lastGetEntryCache.getKey().equals(key)) {
                  oldValue = lastGetEntryCache.setValue(value);
                  lastGetEntryCache.setValue(value);
                  lastGetNodeCache.storeUpdate(tx);
                  flushCache();
                  return oldValue;
              }
  
              // This searches from the last location of a call to get for the element to replace
              // all the way to the end of the ListIndex.
              Iterator<Map.Entry<Key, Value>> iterator = lastGetNodeCache.iterator(tx);
              while (iterator.hasNext()) {
                  Map.Entry<Key, Value> entry = iterator.next();
                  if (entry.getKey().equals(key)) {
                      oldValue = entry.setValue(value);
                      ((ListIterator) iterator).getCurrent().storeUpdate(tx);
                      flushCache();
                      return oldValue;
                  }
              }
          } else {
              flushCache();
          }
  
          // Not found because the cache wasn't set or its not at the end of the list so we
          // start from the beginning and go to the cached location or the end, then we do
          // an add if its not found.
          Iterator<Map.Entry<Key, Value>> iterator = iterator(tx);
          while (iterator.hasNext() && ((ListIterator) iterator).getCurrent() != lastGetNodeCache) {
              Map.Entry<Key, Value> entry = iterator.next();
              if (entry.getKey().equals(key)) {
                  oldValue = entry.setValue(value);
                  ((ListIterator) iterator).getCurrent().storeUpdate(tx);
                  flushCache();
                  return oldValue;
              }
          }
  
          // Not found so add it last.
          flushCache();
  
          return add(tx, key, value);
      }
  
      synchronized public Value add(Transaction tx, Key key, Value value) throws IOException {
          assertLoaded();
          getTail(tx).put(tx, key, value);
          size.incrementAndGet();
          flushCache();
          return null;
      }
  
      synchronized public Value addFirst(Transaction tx, Key key, Value value) throws IOException {
          assertLoaded();
          getHead(tx).addFirst(tx, key, value);
          size.incrementAndGet();
          flushCache();
          return null;
      }
  
      @Override
      @SuppressWarnings("rawtypes")
      synchronized public Value remove(Transaction tx, Key key) throws IOException {
          assertLoaded();
  
          if (size.get() == 0) {
              return null;
          }
  
          if (lastGetNodeCache != null && tx.equals(lastCacheTxSrc.get())) {
  
              // This searches from the last location of a call to get for the element to remove
              // all the way to the end of the ListIndex.
              Iterator<Map.Entry<Key, Value>> iterator = lastGetNodeCache.iterator(tx);
              while (iterator.hasNext()) {
                  Map.Entry<Key, Value> entry = iterator.next();
                  if (entry.getKey().equals(key)) {
                      iterator.remove();
                      flushCache();
                      return entry.getValue();
                  }
              }
          } else {
              flushCache();
          }
  
          // Not found because the cache wasn't set or its not at the end of the list so we
          // start from the beginning and go to the cached location or the end to find the
          // element to remove.
          Iterator<Map.Entry<Key, Value>> iterator = iterator(tx);
          while (iterator.hasNext() && ((ListIterator) iterator).getCurrent() != lastGetNodeCache) {
              Map.Entry<Key, Value> entry = iterator.next();
              if (entry.getKey().equals(key)) {
                  iterator.remove();
                  flushCache();
                  return entry.getValue();
              }
          }
  
          return null;
      }
  
      public void onRemove(Entry<Key, Value> removed) {
          size.decrementAndGet();
          flushCache();
      }
  
      @Override
      public boolean isTransient() {
          return false;
      }
  
      @Override
      synchronized public void clear(Transaction tx) throws IOException {
          for (Iterator<ListNode<Key,Value>> iterator = listNodeIterator(tx); iterator.hasNext(); ) {
              ListNode<Key,Value>candidate = iterator.next();
              candidate.clear(tx);
              // break up the transaction
              tx.commit();
          }
          flushCache();
          size.set(0);
      }
  
      synchronized public Iterator<ListNode<Key, Value>> listNodeIterator(Transaction tx) throws IOException {
          return getHead(tx).listNodeIterator(tx);
      }
  
      synchronized public boolean isEmpty(final Transaction tx) throws IOException {
          return getHead(tx).isEmpty(tx);
      }
  
      @Override
      synchronized public Iterator<Map.Entry<Key,Value>> iterator(final Transaction tx) throws IOException {
          return getHead(tx).iterator(tx);
      }
  
      synchronized public Iterator<Map.Entry<Key,Value>> iterator(final Transaction tx, long initialPosition) throws IOException {
          return getHead(tx).iterator(tx, initialPosition);
      }
  
      synchronized public Map.Entry<Key,Value> getFirst(Transaction tx) throws IOException {
          return getHead(tx).getFirst(tx);
      }
  
      synchronized public Map.Entry<Key,Value> getLast(Transaction tx) throws IOException {
          return getTail(tx).getLast(tx);
      }
  
      private void assertLoaded() throws IllegalStateException {
          if( !loaded.get() ) {
              throw new IllegalStateException("TheListIndex is not loaded");
          }
      }
  
      ListNode<Key,Value> loadNode(Transaction tx, long pageId) throws IOException {
          Page<ListNode<Key,Value>> page = tx.load(pageId, marshaller);
          ListNode<Key, Value> node = page.get();
          node.setPage(page);
          node.setContainingList(this);
          return node;
      }
  
      ListNode<Key,Value> createNode(Page<ListNode<Key,Value>> page) throws IOException {
          ListNode<Key,Value> node = new ListNode<Key,Value>();
          node.setPage(page);
          page.set(node);
          node.setContainingList(this);
          return node;
      }
  
      public ListNode<Key,Value> createNode(Transaction tx) throws IOException {
          return createNode(tx.<ListNode<Key,Value>>load(tx.<ListNode<Key,Value>>allocate().getPageId(), null));
      }
  
      public void storeNode(Transaction tx, ListNode<Key,Value> node, boolean overflow) throws IOException {
          tx.store(node.getPage(), marshaller, overflow);
          flushCache();
      }
  
      public PageFile getPageFile() {
          return pageFile;
      }
  
      public void setPageFile(PageFile pageFile) {
          this.pageFile = pageFile;
      }
  
      public long getHeadPageId() {
          return headPageId;
      }
  
      public void setHeadPageId(long headPageId) {
          this.headPageId = headPageId;
      }
  
      public Marshaller<Key> getKeyMarshaller() {
          return keyMarshaller;
      }
      @Override
      public void setKeyMarshaller(Marshaller<Key> keyMarshaller) {
          this.keyMarshaller = keyMarshaller;
      }
  
      public Marshaller<Value> getValueMarshaller() {
          return valueMarshaller;
      }
      @Override
      public void setValueMarshaller(Marshaller<Value> valueMarshaller) {
          this.valueMarshaller = valueMarshaller;
      }
  
      public void setTailPageId(long tailPageId) {
          this.tailPageId = tailPageId;
      }
  
      public long getTailPageId() {
         return tailPageId;
      }
  
      public long size() {
          return size.get();
      }
  
      private void flushCache() {
          this.lastGetEntryCache = null;
          this.lastGetNodeCache = null;
          this.lastCacheTxSrc.clear();
      }
  }
  