/**
* @license
* Copyright 2020 The FOAM Authors. All Rights Reserved.
* http://www.apache.org/licenses/LICENSE-2.0
*/

package foam.nanos.mrac;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;

import foam.dao.AbstractSink;
import foam.lib.NetworkPropertyPredicate;
import foam.lib.json.Outputter;

/**
* Serialize object and write into a SocketChannel.
* The class is not thread-safe.
* Make sure that only one thread can write into SocketChannelSink at a moment.
*/
//TODO: model this class
public class TcpSocketChannelSink extends AbstractSink {
    
    SelectionKey key;
    SocketChannel socketChannel;
    // This field use to determine if this sink is available to use.
    // We currently do not support removing a sink from listen_.
    // Once socket closed, this sink will still be picked while doing put and remove operation.
    boolean isClose = false;
    
    protected int writeWithPrefix(String source, String prefix) throws IOException {
        return write(prefix + "(" + source + ")");
    }

    protected int write(String source) throws IOException {
        byte[] bytes = source.getBytes(Charset.forName("UTF-8"));
        ByteBuffer byteBuffer = ByteBuffer.allocate(4 + bytes.length);
        byteBuffer.putInt(bytes.length);
        byteBuffer.put(bytes);
        byteBuffer.flip();
        int len = write(byteBuffer);
        byteBuffer.clear();
        return len;
    }

    protected int write(ByteBuffer source) throws IOException {
        return this.getSocketChannel().write(source);
    }
    
    @Override
    public void put(Object obj, foam.core.Detachable sub) {
        if ( ! isClose ) return;
        try {
            String out = new Outputter(getX()).setPropertyPredicate(new NetworkPropertyPredicate()).stringify(obj);
            writeWithPrefix(out, "p");
        } catch ( IOException e ) {
            // Close socket when IOException occurs.
            // Client side should sense this. And re-send sink cmd.
            handleFailure();
        }
    }

    @Override
    public void remove(Object obj, foam.core.Detachable sub) {
        if ( ! isClose ) return;
        try {
            String out = new Outputter(getX()).setPropertyPredicate(new NetworkPropertyPredicate()).stringify(obj);
            writeWithPrefix(out, "r");
        } catch ( IOException e ) {
            // Close socket when IOException occurs.
            // Client side should sense this. And re-send sink cmd.
            handleFailure();
        } 
    }

    public synchronized void handleFailure() {
        if ( ! isClose ) return;
        TCPNioServer.removeSelectionKey(key);
        TCPNioServer.hardCloseSocketChannel(socketChannel);
        isClose = true;
    }
    
    public void setKey(SelectionKey key) {
        this.key = key;
    }
    
    public SelectionKey getSelectionKey() {
        if ( key == null ) {
            key = (SelectionKey) getX().get("selectionKey");
        }
        //TODO: handle key == null exception.
        return key;
    }
    
    public void setSocketChannel(SocketChannel socketChannel) {
        this.socketChannel = socketChannel;
    }
    
    public SocketChannel getSocketChannel() {  
        if ( socketChannel == null ) {
            socketChannel = (SocketChannel) getX().get("socketChannel");
        }
        return socketChannel;
    }
}
@@ -308,16 +308,42 @@
 arset;%0A%0A
+import foam.core.FObject;%0A
 import f
@@ -443,177 +443,8 @@
 r;%0A%0A
-/**%0A* Serialize object and write into a SocketChannel.%0A* The class is not thread-safe.%0A* Make sure that only one thread can write into SocketChannelSink at a moment.%0A*/%0A
 //TO
@@ -908,35 +908,16 @@
  prefix)
- throws IOException
  %7B%0A     
@@ -1008,37 +1008,39 @@
 source) 
-throws IOException %7B%0A
+%7B%0A        // try %7B%0A    
         
@@ -1105,16 +1105,20 @@
         
+  
+  
 ByteBuff
@@ -1172,32 +1172,36 @@
 ength);%0A        
+    
 byteBuffer.putIn
@@ -1217,32 +1217,36 @@
 ength);%0A        
+ 
+   
 byteBuffer.put(b
@@ -1252,16 +1252,20 @@
 bytes);%0A
+    
         
@@ -1291,16 +1291,20 @@
         
+    
 int len 
@@ -1324,32 +1324,36 @@
 uffer);%0A        
+   
+ 
 byteBuffer.clear
@@ -1360,24 +1360,28 @@
 ();%0A        
+    
 return len;%0A
@@ -1388,60 +1388,24 @@
     
-%7D%0A%0A
     
-protected int write(ByteBuffer source) throws
+// %7D catch (
  IOE
@@ -1412,16 +1412,20 @@
 xception
+ e )
  %7B%0A     
@@ -1431,170 +1431,125 @@
     
-return this.getSocketChannel().write(source);%0A    %7D
+//     // This IOException is because of byteBuffer, so can be ignore.
 %0A    
-%0A
     
-@Override%0A    public void put(Object obj, foam.core.Detachable sub) %7B%0A        if ( ! isClose )
+//     // Log%0A        //    
  return
+ -1
 ;%0A  
@@ -1558,166 +1558,148 @@
     
-try %7B%0A            String out = new Outputter(getX()).setPropertyPredicate(new NetworkPropertyPredicate()).stringify(obj);%0A            writeWithPrefix(out, %22p%22
+// %7D%0A    %7D%0A%0A%0A    protected synchronized int write(ByteBuffer source) %7B%0A        try %7B%0A            return this.getSocketChannel().write(source
 );%0A 
@@ -1846,32 +1846,67 @@
 -send sink cmd.%0A
+            System.out.println(e);%0A
             hand
@@ -1910,32 +1910,55 @@
 andleFailure();%0A
+            return -1;%0A
         %7D%0A    %7D%0A
@@ -1953,24 +1953,28 @@
     %7D%0A    %7D%0A
+    
 %0A    @Overri
@@ -1992,22 +1992,19 @@
 ic void 
-remove
+put
 (Object 
@@ -2040,34 +2040,32 @@
 ) %7B%0A        if (
- !
  isClose ) retur
@@ -2067,34 +2067,16 @@
 return;%0A
-        try %7B%0A    
         
@@ -2177,18 +2177,24 @@
 ify(
+(FObject) 
 obj);%0A
-    
     
@@ -2223,209 +2223,285 @@
 t, %22
-r
+p
 %22);%0A
+%0A
     
+%7D%0A%0A
     
-%7D catch ( IOException e ) %7B%0A            // Close socket when IOException occurs.%0A            // Client side should sense this. And re-send sink cmd.%0A            handleFailure();%0A        %7D 
+@Override%0A    public void remove(Object obj, foam.core.Detachable sub) %7B%0A        if ( isClose ) return;%0A        String out = new Outputter(getX()).setPropertyPredicate(new NetworkPropertyPredicate()).stringify((FObject) obj);%0A        writeWithPrefix(out, %22r%22);
 %0A   

fix bug in TcpSocketChannelSink.java