case class Book( isbn: String )

val frankenstein = Book( "978-0486282114" )

/*
 * Copyright (C) 2014 Stratio (http://stratio.com)
 *
 * Licensed 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 com.stratio.cassandra.lucene
.
.
.
class IndexQueryHandler extends QueryHandler with Logging
{
  .
  .
  .
}

/** Companion object for [[IndexQueryHandler]]. */
object IndexQueryHandler
{
  val processResults: Method = classOf[SelectStatement].getDeclaredMethod( "processResults",
                                   classOf[PartitionIterator],
                                   classOf[QueryOptions],
                                   classOf[Int],
                                   classOf[Int] )
  processResults.setAccessible( true )

  /** Sets this query handler as the Cassandra CQL query handler, replacing the previous one. */
  def activate(): Unit =
  {
    this.synchronized
    {
      if( !ClientState.getCQLQueryHandler.isInstanceOf[IndexQueryHandler] )
      {
        try
        {
          val field = classOf[ClientState].getDeclaredField( "cqlQueryHandler" )
          field.setAccessible( true )
          val modifiersField = classOf[Field].getDeclaredField( "modifiers" )
          modifiersField.setAccessible( true )
          modifiersField.setInt( field, field.getModifiers & ~Modifier.FINAL )
          field.set( null, new IndexQueryHandler )
        }
        catch
        {
          case e:
            Exception => throw new IndexException( "Unable to set Lucene CQL query handler", e )
        }
      }
    }
  }
}

.
.
.
public class OurQueryHandler implements Logging, QueryHandler
{
  .
  .
  .

  public static Method processResults()
  {
    try
    {
      Class< ? > clazz  = SelectStatement.class;
      Method     method = clazz.getDeclaredMethod( "processResults", PartitionIterator.class,
                                                     QueryOptions.class,
                                                     int.class, // (strange, but true!)
                                                     int.class );
      method.setAccessible( true );
      return method;
    }
    catch( NoSuchMethodException e )
    {
      throw new IndexException( "Unable to return reflection of method SelectStatement.processResults()", e );
    }
  }

  public static void activate()
  {
    synchronized( OurQueryHandler.class )
    {
      if( !( ClientState.getCQLQueryHandler() instanceof OurQueryHandler ) )
      {
        try
        {
          Class< ? > clazz          = ClientState.class;
          Field      methodField    = clazz.getDeclaredField( "cqlQueryHandler" );
          Field      modifiersField = Field.class.getDeclaredField( "modifiers" );
          methodField.setAccessible( true );
          modifiersField.setAccessible( true );
          modifiersField.setInt( methodField, methodField.getModifiers() & ~Modifier.FINAL );
          methodField.set( null, new OurQueryHandler() );
        }
        catch( Exception e )
        {
          throw new IndexException( "Unable to set our CQL handler", e );
        }
      }
    }
  }
}

private val pools =
  // Start with a range from 1 to numThreads, which you can essentially
  // consider a Seq(1, 2, 3, ..., numThreads).
  ( 1 to numThreads )
    // Now, for each value in that range, pass it to this function and store
    // the results in another sequence. The function ignores the input value
    // (which is what the _ means before the arrow) and returns a
    // new ArrayBlockingQueue[Runnable]. After this step, we will have
    // a Seq[ArrayBlockingQueue[Runnable]], with a number of elements
    // equal to numThreads.
    .map( _ => new ArrayBlockingQueue[Runnable]( queuesSize, true ) )
    // Again, for each of these ArrayBlockingQueues, pass it to this
    // new function (captured in the q argument) and create a new Seq
    // of the results. In this case, it is a Seq[ThreadPoolExecutor]
    // where each of the ThreadPoolExecutor objects has been through the
    // initialization code below
    .map( q => new ThreadPoolExecutor( 1, 1, 1, DAYS, q,
                     new BasicThreadFactory.Builder()
                          .namingPattern( "indexer-%d" )
                          .build(),
                     ( task, executor ) =>
                          if( !executor.isShutdown )
                            executor.getQueue.put( task ) ) )
  // The final resulting type of pools should be Seq[ThreadPoolExecutor],
  // and it will have a number of element equal to numThreads.

List< ThreadPoolExecutor > pools = new ArrayList<>();

BasicThreadFactory threadFactoryBuilder = new BasicThreadFactory.Builder()
                                      .namingPattern( "indexer-%d" )
                                      .build();
// create a list...
List< Queue< Runnable > > queues = new ArrayList<>( numThreads );

// ...and fill it with new queues of queueSize and our default access policy...
while( numThreads-- > 0 )
  queues.add( new ArrayBlockingQueue<>( queueSize, DEFAULT_ACCESS_POLICY ) );

// fill the list of executors with initialized executors that use, each, a queue from above...
for( Queue< Runnable > queue : queues )
  pools.add( new ThreadPoolExecutor( CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME, KEEP_ALIVE_FOR_DAY,
                                  ( BlockingQueue< Runnable > ) queue, threadFactoryBuilder ) );

private[this] val indexes: List[FsIndex] = partitions match
{
  case 1 =>
     List( new FsIndex(name, path, analyzer, refreshSeconds, ramBufferMB, maxMergeMB, maxCachedMB ) )
  case n if n > 1 =>
     val root = path.toFile.getAbsolutePath + File.separator
     ( 0 until n )
       .map(root + File.separator + _ )
       .map(Paths.get( _ ) )
       .map(new FsIndex( name, _, analyzer, refreshSeconds, ramBufferMB, maxMergeMB, maxCachedMB ) )
       .toList
  case _ =>
     throw new IndexException( s"The number of partitions should be strictly positive but found $partitions" )
}

if( numPartitions < 1 )
  throw new IndexException( "The number of partitions should be greater than 0" );

List< FsIndex > indices = new ArrayList<>();
final String    ROOT    = path.toFile().getAbsolutePath();

switch( numPartitions )
{
  case 1 :
   indices.add( new FsIndex( name, path, analyzer, refreshSeconds, ramBufferMb, maxMergeMb, maxCachedMb ) );
   break;
  default :
   // build a new index for each set of data we can muster here...
   for( int partition = 0; partition < numPartitions; partition++ )
   {
     indices.add( new FsIndex( name,
                         Paths.get( ROOT + File.separator + partition ),
                         analyzer, refreshSeconds, ramBufferMb, maxMergeMb, maxCachedMb ) );
   }
   break;
}

List< FsIndex > indices = [ ... ];
long count = 0;

for( FsIndex index : indices )
  count += index.getNumDocs();