/*
    * 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.hadoop.hbase.codec.prefixtree.decode;
  
  import java.nio.ByteBuffer;
  
  import org.apache.hadoop.hbase.classification.InterfaceAudience;
  import org.apache.hadoop.hbase.codec.prefixtree.PrefixTreeBlockMeta;
  import org.apache.hadoop.hbase.codec.prefixtree.scanner.CellSearcher;
  
  /**
   * Static wrapper class for the ArraySearcherPool.
   */
  @InterfaceAudience.Private
  public class DecoderFactory {
  
    private static final ArraySearcherPool POOL = new ArraySearcherPool();
  
    //TODO will need a PrefixTreeSearcher on top of CellSearcher
    public static PrefixTreeArraySearcher checkOut(final ByteBuffer buffer, 
        boolean includeMvccVersion) {
      if (buffer.isDirect()) {
        throw new IllegalArgumentException("DirectByteBuffers not supported yet");
        // TODO implement PtByteBufferBlockScanner
      }
  
      PrefixTreeArraySearcher searcher = POOL.checkOut(buffer,
        includeMvccVersion);
      return searcher;
    }
  
    public static void checkIn(CellSearcher pSearcher) {
      if (pSearcher == null) {
        return;
      }
      if (! (pSearcher instanceof PrefixTreeArraySearcher)) {
        throw new IllegalArgumentException("Cannot return "+pSearcher.getClass()+" to "
            +DecoderFactory.class);
      }
      PrefixTreeArraySearcher searcher = (PrefixTreeArraySearcher) pSearcher;
      POOL.checkIn(searcher);
    }
  
  
    /**************************** helper ******************************/
    public static PrefixTreeArraySearcher ensureArraySearcherValid(ByteBuffer buffer,
        PrefixTreeArraySearcher searcher, boolean includeMvccVersion) {
      if (searcher == null) {
        PrefixTreeBlockMeta blockMeta = new PrefixTreeBlockMeta(buffer);
        searcher = new PrefixTreeArraySearcher(blockMeta, blockMeta.getRowTreeDepth(),
            blockMeta.getMaxRowLength(), blockMeta.getMaxQualifierLength(),
            blockMeta.getMaxTagsLength());
        searcher.initOnBlock(blockMeta, buffer.array(), includeMvccVersion);
        return searcher;
      }
  
      PrefixTreeBlockMeta blockMeta = searcher.getBlockMeta();
      blockMeta.initOnBlock(buffer);
      if (!searcher.areBuffersBigEnough()) {
        int maxRowTreeStackNodes = Math.max(blockMeta.getRowTreeDepth(),
          searcher.getMaxRowTreeStackNodes());
        int rowBufferLength = Math.max(blockMeta.getMaxRowLength(), searcher.getRowBufferLength());
        int qualifierBufferLength = Math.max(blockMeta.getMaxQualifierLength(),
          searcher.getQualifierBufferLength());
        int tagBufferLength = Math.max(blockMeta.getMaxTagsLength(), searcher.getTagBufferLength());
        searcher = new PrefixTreeArraySearcher(blockMeta, maxRowTreeStackNodes, rowBufferLength,
            qualifierBufferLength, tagBufferLength);
      }
      //this is where we parse the BlockMeta
      searcher.initOnBlock(blockMeta, buffer.array(), includeMvccVersion);
      return searcher;
    }
  
  }