package nanomsg.async;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import java.util.Queue;
import java.util.LinkedList;
import java.lang.Thread;

import nanomsg.ISocket;
import nanomsg.Nanomsg;
import nanomsg.async.IAsyncRunnable;
import nanomsg.async.IAsyncScheduler;
import nanomsg.exceptions.IOException;


public class SimpleAsyncScheduler implements Runnable, IAsyncScheduler {
    private final LinkedBlockingQueue<IAsyncRunnable> queue = new LinkedBlockingQueue<IAsyncRunnable>();
    private final int concurrency = Runtime.getRuntime().availableProcessors();
    private final AtomicBoolean started = new AtomicBoolean(false);

    public static final IAsyncScheduler instance = new SimpleAsyncScheduler();

    private void startThreadGroup() {
        final ThreadGroup group = new ThreadGroup("nanomsg-scheduler");
        group.setDaemon(false);

        for (int i=0; i<concurrency; ++i) {
            final Thread t = new Thread(group, this);
            t.setDaemon(false);
            t.start();
        }
    }

    public void schedule(final IAsyncRunnable handler) throws InterruptedException {
        /* Start scheduler if it not started */
        if (started.compareAndSet(false, true)) {
            startThreadGroup();
        }

        queue.put(handler);
    }

    @Override
    public String toString() {
        return "pending=" + queue.size() + ", pool started:" + started.get();
    }

    public void run() {
        while (true) {
            try {
                final IAsyncRunnable handler = queue.take();
                try {
                    handler.run();
                } catch (IOException e) {
                    final int errno = e.getErrno();
                    if (errno == Nanomsg.constants.EAGAIN) {
                        queue.put(handler);
                    }
                }
            } catch (InterruptedException e) {
                break;
            }
        }
    }
}

@@ -484,18 +484,16 @@
 duler %7B%0A
-  
   privat
@@ -583,26 +583,24 @@
 unnable%3E();%0A
-  
   private fi
@@ -665,18 +665,16 @@
 sors();%0A
-  
   privat
@@ -730,26 +730,24 @@
 (false);%0A%0A  
- 
- 
 public stati
@@ -810,18 +810,16 @@
 ler();%0A%0A
-  
   privat
@@ -846,28 +846,24 @@
 oup() %7B%0A    
-   
- 
 final Thread
@@ -914,20 +914,16 @@
 uler%22);%0A
-    
     grou
@@ -947,20 +947,16 @@
 );%0A%0A    
-  
-  
 for (int
@@ -985,30 +985,24 @@
 +i) %7B%0A      
-    
-  
 final Thread
@@ -1031,22 +1031,16 @@
  this);%0A
-      
       t.
@@ -1063,22 +1063,16 @@
 ;%0A      
-     
- 
 t.start(
@@ -1078,27 +1078,19 @@
 ();%0A
-    
     %7D%0A
-  
   %7D%0A%0A
-  
   pu
@@ -1172,20 +1172,16 @@
 n %7B%0A    
-   
- 
 /* Start
@@ -1212,28 +1212,24 @@
 rted */%0A    
-   
- 
 if (started.
@@ -1264,22 +1264,16 @@
 %7B%0A      
-  
-    
 startThr
@@ -1288,27 +1288,19 @@
 ();%0A    
-  
-  
 %7D%0A%0A
-    
     queu
@@ -1317,23 +1317,19 @@
 ler);%0A  
- 
- 
 %7D%0A%0A
-  
   @Overr
@@ -1332,18 +1332,16 @@
 verride%0A
-  
   public
@@ -1365,20 +1365,16 @@
 ) %7B%0A    
- 
-   
 return %22
@@ -1435,25 +1435,21 @@
 .get();%0A
-  
   %7D%0A%0A
-  
   public
@@ -1466,20 +1466,16 @@
 ) %7B%0A    
- 
-   
 while (t
@@ -1487,36 +1487,22 @@
 %7B%0A      
-  
-    
 try %7B%0A
-        
         
@@ -1542,32 +1542,24 @@
 eue.take();%0A
-        
         try 
@@ -1570,26 +1570,16 @@
         
-     
-     
 handler.
@@ -1581,32 +1581,24 @@
 dler.run();%0A
-        
         %7D ca
@@ -1615,34 +1615,24 @@
 eption e) %7B%0A
-          
           fi
@@ -1671,26 +1671,16 @@
         
-   
-       
 if (errn
@@ -1724,28 +1724,16 @@
         
-     
-       
 queue.pu
@@ -1754,39 +1754,26 @@
         
-      
-    %7D%0A   
+%7D%0A
         
      %7D%0A 
@@ -1760,37 +1760,32 @@
   %7D%0A        
-     
 %7D%0A      
       %7D catc
@@ -1772,22 +1772,16 @@
 %7D%0A      
- 
-     
 %7D catch 
@@ -1803,32 +1803,24 @@
 eption e) %7B%0A
-        
         brea
@@ -1828,34 +1828,22 @@
 ;%0A      
-      %7D%0A        
 %7D%0A  
+  %7D%0A
   %7D%0A%7D%0A

Fix code style on SimpleAsyncScheduler.java.