/**
 * 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.transport.nio;

import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class SelectorWorker implements Runnable {

    private static final AtomicInteger NEXT_ID = new AtomicInteger();

    final SelectorManager manager;
    final Selector selector;
    final int id = NEXT_ID.getAndIncrement();
    private final int maxChannelsPerWorker;

    final AtomicInteger retainCounter = new AtomicInteger(1);
    private final ConcurrentLinkedQueue<Runnable> ioTasks = new ConcurrentLinkedQueue<Runnable>();
       
    public SelectorWorker(SelectorManager manager) throws IOException {
        this.manager = manager;
        selector = Selector.open();
        maxChannelsPerWorker = manager.getMaxChannelsPerWorker();
        manager.getSelectorExecutor().execute(this);
    }

    void retain() {
        if (retainCounter.incrementAndGet() == maxChannelsPerWorker) {
            manager.onWorkerFullEvent(this);
        }
    }

    void release() {
        int use = retainCounter.decrementAndGet();
        if (use == 0) {
            manager.onWorkerEmptyEvent(this);
        } else if (use < maxChannelsPerWorker) {
            manager.onWorkerNotFullEvent(this);
        }
    }
    
    boolean isReleased() {
        return retainCounter.get()==0;
    }


    public void addIoTask(Runnable work) {
        ioTasks.add(work);
        selector.wakeup();
    }
    
    private void processIoTasks() {
        Runnable task; 
        while( (task= ioTasks.poll()) !=null ) {
            try {
                task.run();
            } catch (Throwable e) {
                e.printStackTrace();
            }
        }
    }

    

    public void run() {

        String origName = Thread.currentThread().getName();
        try {
            Thread.currentThread().setName("Selector Worker: " + id);
            while (!isReleased()) {
                
                processIoTasks();
                int count = selector.select(10);
                if (count == 0) {
                    continue;
                }

                // Get a java.util.Set containing the SelectionKey objects
                // for all channels that are ready for I/O.
                Set keys = selector.selectedKeys();

                for (Iterator i = keys.iterator(); i.hasNext();) {
                    final SelectionKey key = (SelectionKey)i.next();
                    i.remove();

                    final SelectorSelection s = (SelectorSelection)key.attachment();
                    try {
                        if( key.isValid() ) {
                            key.interestOps(0);
                        }

                        // Kick off another thread to find newly selected keys
                        // while we process the
                        // currently selected keys
                        manager.getChannelExecutor().execute(new Runnable() {
                            public void run() {
                                try {
                                    s.onSelect();
                                    s.enable();
                                } catch (Throwable e) {
                                    s.onError(e);
                                }
                            }
                        });

                    } catch (Throwable e) {
                        s.onError(e);
                    }

                }

            }
            
        } catch (Throwable e) {
            // Notify all the selections that the error occurred.
            Set keys = selector.keys();
            for (Iterator i = keys.iterator(); i.hasNext();) {
                SelectionKey key = (SelectionKey)i.next();
                SelectorSelection s = (SelectorSelection)key.attachment();
                s.onError(e);
            }

        } finally {
            try {
                manager.onWorkerEmptyEvent(this);
                selector.close();
            } catch (IOException ignore) {
            }
            Thread.currentThread().setName(origName);
        }
    }

}

@@ -2413,20 +2413,23 @@
 private 
-void
+boolean
  process
@@ -2436,24 +2436,52 @@
 IoTasks() %7B%0A
+        boolean rc = false;%0A
         Runn
@@ -2551,32 +2551,59 @@
           try %7B%0A
+                rc = true;%0A
                 
@@ -2703,32 +2703,51 @@
     %7D%0A        %7D%0A
+        return rc;%0A
     %7D%0A%0A    %0A%0A   
@@ -2980,16 +2980,20 @@
         
+ if(
  process
@@ -3001,17 +3001,68 @@
 oTasks()
-;
+ ) %7B%0A                    continue;%0A                %7D
 %0A       

check for release after processig io tasks