/*

  *

  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.

  *

  * Copyright 2007-2008 Sun Microsystems, Inc. All rights reserved.

  *

  * The contents of this file are subject to the terms of either the GNU

  * General Public License Version 2 only ("GPL") or the Common Development

  * and Distribution License("CDDL") (collectively, the "License").  You

  * may not use this file except in compliance with the License. You can obtain

  * a copy of the License at https://glassfish.dev.java.net/public/CDDL+GPL.html

  * or glassfish/bootstrap/legal/LICENSE.txt.  See the License for the specific

  * language governing permissions and limitations under the License.

  *

  * When distributing the software, include this License Header Notice in each

  * file and include the License file at glassfish/bootstrap/legal/LICENSE.txt.

  * Sun designates this particular file as subject to the "Classpath" exception

  * as provided by Sun in the GPL Version 2 section of the License file that

  * accompanied this code.  If applicable, add the following below the License

  * Header, with the fields enclosed by brackets [] replaced by your own

  * identifying information: "Portions Copyrighted [year]

  * [name of copyright owner]"

  *

  * Contributor(s):

  *

  * If you wish your version of this file to be governed by only the CDDL or

  * only the GPL Version 2, indicate your decision by adding "[Contributor]

  * elects to include this software in this distribution under the [CDDL or GPL

  * Version 2] license."  If you don't indicate a single choice of license, a

  * recipient has the option to distribute your version of this file under

  * either the CDDL, the GPL Version 2 or to extend the choice of license to

  * its licensees as provided above.  However, if you add GPL Version 2 code

  * and therefore, elected the GPL Version 2 license, then the option applies

  * only if the new code is made subject to such option by the copyright

  * holder.

  *

  */

 package com.sun.grizzly;

 import com.sun.grizzly.util.ByteBufferFactory.ByteBufferType;

 import com.sun.grizzly.util.WorkerThreadImpl;

 import java.util.concurrent.Callable;

 import java.util.concurrent.LinkedBlockingQueue;

 import java.util.concurrent.ThreadFactory;

 import java.util.concurrent.ThreadPoolExecutor;

 import java.util.concurrent.TimeUnit;

 import java.util.concurrent.atomic.AtomicInteger;

 /**

  * {@link Pipeline} implementation, based on {@link ThreadPoolExecutor}

  *

  * @author Alexey Stashok

  */

 public class ThreadPoolExecutorServicePipeline extends ThreadPoolExecutor

         implements Pipeline<Callable> {

     // Min number of worker threads in a pool

     private static int DEFAULT_MIN_THREAD_COUNT = 5;

     // Max number of worker threads in a pool

     private static int DEFAULT_MAX_THREAD_COUNT = 20;

     // Max number of tasks thread pool can enqueue

     private static int DEFAULT_MAX_TASKS_QUEUED = Integer.MAX_VALUE;

     // Timeout, after which idle thread will be stopped and excluded from pool

     private static int DEFAULT_IDLE_THREAD_KEEPALIVE_TIMEOUT = 30000;

     private String name;

     /**

      * The port used.

      */

     protected int port;

     private int maxTasksCount;

     private AtomicInteger workerThreadCounter = new AtomicInteger();

     /**

      * The Thread Priority

      */

     protected int priority;

     /**

      * The initial ByteBuffer size for newly created WorkerThread instances

      */

     protected int initialByteBufferSize = 8192;

     /**

      * The {@link ByteBufferType}

      */

     private ByteBufferType byteBufferType = ByteBufferType.HEAP_VIEW;

     public ThreadPoolExecutorServicePipeline() {

         this(DEFAULT_MIN_THREAD_COUNT, DEFAULT_MAX_THREAD_COUNT, 

                 DEFAULT_MAX_TASKS_QUEUED, DEFAULT_IDLE_THREAD_KEEPALIVE_TIMEOUT,

                 TimeUnit.MILLISECONDS);

     }

     public ThreadPoolExecutorServicePipeline(int minThreads,

             int maxThreads, int maxTasksCount, long keepAliveTime,

             TimeUnit unit) {

         this(minThreads, maxThreads, maxTasksCount, keepAliveTime, unit, 

                 "Grizzly", 8080, Thread.NORM_PRIORITY);

     }

     public ThreadPoolExecutorServicePipeline(int minThreads,

             int maxThreads, int maxTasksCount, long keepAliveTime,

             TimeUnit unit, String name, int port, int priority) {

         super(minThreads, maxThreads, keepAliveTime, unit,

                 new LinkedBlockingQueue<Runnable>(maxTasksCount));

         setThreadFactory(new WorkerThreadFactory(this));

         this.maxTasksCount = maxTasksCount;

         this.name = name;

         this.port = port;

         this.priority = priority;

     }

     /**

      * Set the name of this {@link Pipeline}

      * @param name Pipeline name to use

      */

     public synchronized void setName(String name){

         this.name = name;

     }

     /**

      * Return the name of this {@link Pipeline}

      * @return the name of this {@link Pipeline}

      */

     public synchronized String getName(){

         return name+port;

     }

     /**

      * Set the port used by this {@link Pipeline}

      * @param port the port used by this {@link Pipeline}

      */

     public synchronized void setPort(int port){

         this.port = port;

     }

     /**

      * {@inheritDoc}

      */

     public void execute(Callable task) throws PipelineFullException {

         submit(task);

     }

     /**

      * {@inheritDoc}

      */

     public Callable waitForIoTask() {

         Callable callable = null;

         try {

             final Runnable r = this.getQueue().take();

             if (r != null) {

                 callable = new Callable() {

                     public Object call() throws Exception {

                         r.run();

                         return null;

                     }

                 };

             }

         } catch (InterruptedException e) {

         }

         return callable;

     }

     /**

      * {@inheritDoc}

      */

     public int getWaitingThread() {

         int waitingThreads = getPoolSize() - getActiveCount();

         if (waitingThreads < 0) waitingThreads = 0;

         return waitingThreads;

     }

     /**

      * {@inheritDoc}

      */

     public int getMaxThreads() {

         return getMaximumPoolSize();

     }

     /**

      * {@inheritDoc}

      */

     public void setMaxThreads(int maxThread) {

         setMaximumPoolSize(maxThread);

     }

     /**

      * {@inheritDoc}

      */

     public int getCurrentThreadCount() {

         return getPoolSize();

     }

     /**

      * {@inheritDoc}

      */

     public int getCurrentThreadsBusy() {

         return getActiveCount();

     }

     /**

      * {@inheritDoc}

      */

     public void initPipeline() {

     }

     /**

      * {@inheritDoc}

      */

     public void startPipeline() {

         this.prestartCoreThread();

     }

     /**

      * {@inheritDoc}

      */

     public void stopPipeline() {

         shutdownNow();

     }

     /**

      * Set the thread priority of the {@link Pipeline}

      * @param priority thread priority to use

      */

     public synchronized void setPriority(int priority){

         this.priority = priority;

     }

     /**

      * {@inheritDoc}

      */

     public void setMinThreads(int minThread) {

         setCorePoolSize(minThread);

     }

     /**

      * Get the maximum pending connections this {@link Pipeline}

      * can handle.

      * @return maximum queue size (in bytes) this Pipeline is using

      */

     public synchronized int getMaxQueueSize(){

         return maxTasksCount;

     }

     /**

      * Is not supported for the <tt>ThreadPoolExecutorServicePipeline<tt>.

      * The value, passed to the cosntructor could not be changed at the runtime.

      * 

      * @param maxQueue

      */

     public void setMaxQueueSize(int maxQueue) {

         throw new UnsupportedOperationException("Value could not be changed!");

     }

     /**

      * Is not supported for the <tt>ThreadPoolExecutorServicePipeline<tt>.

      * @param threadIncrements

      */

     public void setThreadsIncrement(int threadIncrements) {

         throw new UnsupportedOperationException("Not supported.");

     }

     /**

      * {@inheritDoc}

      */

     public int size() {

         return getQueue().size();

     }

     /**

      * The {@link ByteBufferTypel} used to create the {@link ByteBuffer}

      * associated with {@link WorkerThreadImpl}s created by this instance.

      * @return The {@link ByteBufferTypel} used to create the {@link ByteBuffer}

      * associated with {@link WorkerThreadImpl}s created by this instance.

      */

     public ByteBufferType getByteBufferType() {

         return byteBufferType;

     }

     /**

      * Set the {@link ByteBufferTypel} to use when creating the

      * {@link ByteBuffer} associated with {@link WorkerThreadImpl}s

      * created by this instance.

      * @param byteBufferType The ByteBuffer type.

      */

     public void setByteBufferType(ByteBufferType byteBufferType) {

         this.byteBufferType = byteBufferType;

     }

     /**

      * Get the initial WorkerThreadImpl {@link ByteBuffer} size

      * @return initial WorkerThreadImpl <code>ByteBuffwaitingThreadser</code> size

      */

     public synchronized int getInitialByteBufferSize(){

         return initialByteBufferSize;

     }

     /**

      * Set the initial WorkerThreadImpl {@link ByteBuffer} size

      * @param size initial WorkerThreadImpl {@link ByteBuffer} size

      */

     public synchronized void setInitialByteBufferSize(int size){

         initialByteBufferSize = size;

     }

     protected static class WorkerThreadFactory implements ThreadFactory {

         private ThreadPoolExecutorServicePipeline threadPool;

         public WorkerThreadFactory(

                 ThreadPoolExecutorServicePipeline threadPool) {

             this.threadPool = threadPool;

         }

         public Thread newThread(Runnable r) {

             WorkerThreadImpl workerThread = new WorkerThreadImpl(threadPool,

                     threadPool.name + "WorkerThread-"  + threadPool.port + "-" +

                     threadPool.workerThreadCounter.getAndIncrement(), r,

                     threadPool.initialByteBufferSize);

             workerThread.setByteBufferType(threadPool.byteBufferType);

             workerThread.setPriority(threadPool.priority);

             return workerThread;

         }

     }

 }