package org.jetio;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;

import org.jetlang.channels.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Manages the {@link ByteBuffer}s to write for a {@link Session}
 *
 * @author <a href="mailto:peter.royal@pobox.com">peter royal</a>
 */
class WriteQueue {
    private static final Logger logger = LoggerFactory.getLogger( WriteQueue.class );

    private final List<ByteBuffer> queue = Collections.synchronizedList( new ArrayList<ByteBuffer>() );
    private final Publisher<Event> addToWriteSelector;
    private final Publisher<DataEvent<IOException>> failed;
    private final Session session;

    WriteQueue( Session session, Publisher<Event> addToWriteSelector, Publisher<DataEvent<IOException>> failed ) {
        this.session = session;
        this.addToWriteSelector = addToWriteSelector;
        this.failed = failed;
    }

    // TODO return a WriteFuture-like thing
    void add( ByteBuffer[] buffers ) {
        SocketChannel channel = session.channel();

        synchronized( channel.blockingLock() ) {
            try {
                boolean empty = queue.isEmpty();

                if ( channel.isBlocking() ) {
                    if ( !empty ) {
                        process();
                    }

                    write( buffers );
                } else if ( empty ) {
                    // Shamelessly borrow this idea from Netty!
                    int cleared = write( buffers );

                    if ( cleared < buffers.length ) {
                        queue.addAll( Arrays.asList( buffers ).subList( cleared, buffers.length ) );
                        addToWriteSelector.publish( new Event( session ) );
                    }
                } else {
                    queue.addAll( Arrays.asList( buffers ) );
                }
            } catch( IOException e ) {
                failed.publish( new DataEvent<IOException>( session, e ) );
                // TODO return FAIL on the future, when that occurs
            }
        }
    }

    void process() throws IOException {
        written( write( writeQueue() ) );
    }

    private int write( ByteBuffer[] buffers ) throws IOException {
        SocketChannel channel = session.channel();
        long written;

        synchronized( channel.blockingLock() ) {
            // Do not change blocking mode while writing!
            written = channel.write( buffers );
        }

        int cleared = 0;

        for ( ByteBuffer buffer : buffers ) {
            if ( buffer.remaining() == 0 ) {
                cleared++;
            } else {
                break;
            }
        }

        logger.debug( "wrote {} bytes to {}", written, this );

        return cleared;
    }

    private ByteBuffer[] writeQueue() {
        synchronized( queue ) {
            return queue.toArray( new ByteBuffer[queue.size()] );
        }
    }

    private void written( int count ) {
        synchronized( queue ) {
            queue.subList( 0, count ).clear();

            // TODO thinking about ensuring that the queue's underlying list doesn't get too big. ArrayList.trimToSize

            if ( queue.isEmpty() ) {
                session.selectionKeys().cancel( SelectionOp.Write );
            }
        }
    }
}

@@ -1194,61 +1194,8 @@
 );%0A%0A
-        synchronized( channel.blockingLock() ) %7B%0A    
     
@@ -1204,20 +1204,16 @@
   try %7B%0A
-    
         
@@ -1258,28 +1258,24 @@
             
-  
-  
 if ( channel
@@ -1304,28 +1304,24 @@
             
-    
 if ( !empty 
@@ -1344,20 +1344,16 @@
         
-  
-  
 process(
@@ -1367,28 +1367,24 @@
             
-    
 %7D%0A%0A         
@@ -1386,28 +1386,24 @@
             
- 
-   
 write( buffe
@@ -1424,31 +1424,14 @@
     
-    
 %7D else
- if ( empty )
  %7B%0A 
@@ -1449,95 +1449,35 @@
     
-    // Shamelessly borrow this idea from Netty!%0A                    int cleared = write
+queue.addAll( Arrays.asList
 ( bu
@@ -1481,25 +1481,23 @@
  buffers
+ )
  );%0A%0A
-    
         
@@ -1513,142 +1513,32 @@
 f ( 
-cleared %3C buffers.length ) %7B%0A                        queue.addAll( Arrays.asList( buffers ).subList( cleared, buffers.length ) );%0A    
+empty && !process() ) %7B%0A
     
@@ -1617,34 +1617,26 @@
             
-    %7D%0A    
+%7D%0A
             
@@ -1640,100 +1640,9 @@
    %7D
- else %7B
 %0A
-                    queue.addAll( Arrays.asList( buffers ) );%0A                %7D%0A    
     
@@ -1668,28 +1668,24 @@
 ption e ) %7B%0A
-    
             
@@ -1748,36 +1748,32 @@
  );%0A            
-    
 // TODO return F
@@ -1820,39 +1820,41 @@
     
-    %7D%0A        %7D%0A    %7D%0A%0A    void
+%7D%0A    %7D%0A%0A    synchronized boolean
  pro
@@ -1885,24 +1885,31 @@
 n %7B%0A        
+return 
 written( wri
@@ -2699,32 +2699,210 @@
        %7D%0A    %7D%0A%0A
+    /**%0A     * Remove written buffers from the queue%0A     *%0A     * @param count Number of buffers that were consumed%0A     *%0A     * @return True if the queue is now empty%0A     */%0A
     private void
@@ -2897,20 +2897,23 @@
 private 
-void
+boolean
  written
@@ -3234,29 +3234,86 @@
 rite );%0A
-            %7D
+%0A                return true;%0A            %7D%0A%0A            return false;
 %0A       

Write optimizations - Do not get the blocking lock over the entire write operation - Always add to the queue when in non-blocking mode - Try and process the queue if we saw it empty first - Prevent concurrent processing of a given queue