/* $Id$ */

package ibis.ipl.impl.nio;

import java.io.IOException;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;

/**
 * Dissipator which reads from a single channel. Reads whenever the
 * SendReceiveThread askes it to.
 */
final class ThreadNioDissipator extends NioDissipator {

    final SendReceiveThread sendReceiveThread;

    SelectionKey key;

    boolean reading = false;

    int minimum = 0;

    IOException error = null; // used to notify user of exceptions

    ThreadNioDissipator(SendReceiveThread sendReceiveThread,
            ReadableByteChannel channel) throws IOException {
        super(channel);

        this.sendReceiveThread = sendReceiveThread;

        if (!(channel instanceof SelectableChannel)) {
            throw new IOException("wrong type of channel given on creation of"
                    + " ThreadNioDissipator");
        }
        key = sendReceiveThread.register((SelectableChannel) channel, this);
        sendReceiveThread.enableReading(key);
        reading = true;
    }

    synchronized void receive() throws IOException {
        super.receive();

        if (!reading) {
            sendReceiveThread.enableReading(key);
            reading = true;
        }
    }

    synchronized boolean messageWaiting() throws IOException {
        if (key == null) {
            // this dissipator is already closed
            return false;
        }
        return super.messageWaiting();
    }

    /**
     * Called by the send/receive thread to indicate we may read from the
     * channel now.
     */
    synchronized void doRead() {
        boolean bufferWasEmpty;

        synchronized (this) {
            try {
                if (!reading) {
                    return;
                }

                bufferWasEmpty = (unUsedLength() == 0);

                readFromChannel();

                // no use reading anymore, it's already full
                if (!buffer.hasRemaining()) {
                    key.interestOps(0);
                    reading = false;
                }

                if (minimum != 0 && unUsedLength() >= minimum) {
                    notifyAll();
                }
            } catch (IOException e) {
                error = e;
                key.interestOps(0);
                reading = false;
                try {
                    channel.close();
                } catch (IOException e2) {
                    // IGNORE
                }
                return;
            }
        }
        // signal the port data is available in this dissipator now
        if (bufferWasEmpty) {
            ((ThreadNioReceivePort) info.port).addToReadyList(this);
        }
    }

    /*
     * fills the buffer upto at least "minimum" bytes.
     * 
     */
    synchronized protected void fillBuffer(int minimum) throws IOException {
        // since the send/receive thread actually receives,
        // we can only wait for it to put the data in the buffer

        if (!reading) {
            sendReceiveThread.enableReading(key);
            reading = true;
        }

        while (unUsedLength() < minimum) {
            if (!reading) {
                sendReceiveThread.enableReading(key);
                reading = true;
            }
            if (error != null) {
                // an exception occured while receiving, throw exception now
                throw error;
            }
            try {
                this.minimum = minimum;
                wait();
            } catch (InterruptedException e) {
                // IGNORE
            }
            this.minimum = 0;
        }
    }

    public void close() throws IOException {
        reading = false;
        super.close();
    }
}

@@ -1694,42 +1694,8 @@
 y;%0A%0A
-        synchronized (this) %7B%0A    
     
@@ -1708,36 +1708,32 @@
 y %7B%0A            
- 
-   
 if (!reading) %7B%0A
@@ -1740,36 +1740,32 @@
                 
-    
 return;%0A        
@@ -1764,35 +1764,27 @@
             
-  
-  
 %7D%0A%0A
-    
             
@@ -1832,28 +1832,24 @@
             
-   
- 
 readFromChan
@@ -1852,28 +1852,24 @@
 Channel();%0A%0A
-    
             
@@ -1913,20 +1913,16 @@
 dy full%0A
-    
         
@@ -1963,36 +1963,32 @@
                 
-    
 key.interestOps(
@@ -1983,36 +1983,32 @@
 interestOps(0);%0A
-    
                 
@@ -2032,35 +2032,27 @@
             
-    %7D%0A%0A    
+%7D%0A%0A
             
@@ -2116,20 +2116,16 @@
         
-    
 notifyAl
@@ -2141,32 +2141,26 @@
         
-    %7D%0A  
+%7D%0A
         
   %7D catc
@@ -2143,34 +2143,32 @@
       %7D%0A        
-  
 %7D catch (IOExcep
@@ -2189,20 +2189,16 @@
         
-    
 error = 
@@ -2204,36 +2204,32 @@
  e;%0A            
-   
- 
 key.interestOps(
@@ -2236,36 +2236,32 @@
 0);%0A            
-    
 reading = false;
@@ -2269,38 +2269,30 @@
             
-   
- 
 try %7B%0A
-    
             
@@ -2308,28 +2308,24 @@
 el.close();%0A
-    
             
@@ -2359,36 +2359,32 @@
                 
-    
 // IGNORE%0A      
@@ -2381,36 +2381,32 @@
 ORE%0A            
-    
 %7D%0A              
@@ -2399,36 +2399,32 @@
             
- 
-   
 return;%0A
             
@@ -2411,30 +2411,16 @@
 return;%0A
-            %7D%0A
         

Remove unnecessary double synchronization.