package com.alternativeinfrastructures.noise.sync;

import android.util.Log;

import com.alternativeinfrastructures.noise.storage.BloomFilter;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.charset.Charset;
import java.util.BitSet;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import okio.BufferedSink;
import okio.BufferedSource;
import okio.Okio;

public class StreamSync {
    public static final String TAG = "StreamSync";

    public static void bidirectionalSync(InputStream inputStream, OutputStream outputStream) {
        Log.d(TAG, "Starting sync");

        final BufferedSource source = Okio.buffer(Okio.source(inputStream));
        final BufferedSink sink = Okio.buffer(Okio.sink(outputStream));
        ExecutorService ioExecutors = Executors.newFixedThreadPool(2); // Separate threads for send and receive

        IOFutures<String> handshakeFutures = handshakeAsync(source, sink, ioExecutors);

        try {
            handshakeFutures.get();
        } catch (Exception e) {
            Log.e(TAG, "Handshake failed", e);
            return;
        }

        Log.d(TAG, "Connected to a peer");

        final BitSet myMessageVector = BloomFilter.getMessageVectorAsync().blockingGet();
        IOFutures<BitSet> messageVectorFutures = exchangeMessageVectorsAsync(myMessageVector, source, sink, ioExecutors);

        BitSet theirMessageVector;
        try {
            theirMessageVector = messageVectorFutures.get();
        } catch (Exception e) {
            Log.e(TAG, "Failed to exchange message vectors", e);
            return;
        }

        Log.d(TAG, "Exchanged message vectors");

        // TODO: Include a subset of the message vector in the broadcast and verify that theirMessageVector matches
        // TODO: Send and receive individual messages (in separate threads so we can send and receive simultaneously)
        // TODO: Write tests for this class/function
        ioExecutors.shutdown();
    }

    private static final String PROTOCOL_NAME = "Noise0";
    private static final Charset DEFAULT_CHARSET = Charset.forName("US-ASCII");

    private enum Messages {
        MESSAGE_VECTOR((byte) 1),
        MESSAGE((byte) 2),
        END((byte) 3);

        private byte value;

        Messages(byte value) {
            this.value = value;
        }

        byte getValue() {
            return value;
        }
    }

    static class IOFutures<T> {
        public Future<Void> sender;
        public Future<T> receiver;

        public T get() throws InterruptedException, ExecutionException {
            if (sender != null)
                sender.get();

            if (receiver != null)
                return receiver.get();
            else
                return null;
        }
    }

    static IOFutures<String> handshakeAsync(final BufferedSource source, final BufferedSink sink, ExecutorService ioExecutors) {
        if (PROTOCOL_NAME.length() > Byte.MAX_VALUE)
            Log.wtf(TAG, "Protocol name is too long");

        IOFutures<String> futures = new IOFutures<String>();

        futures.sender = ioExecutors.submit(new Callable<Void>() {
            @Override
            public Void call() throws Exception {
                sink.writeByte(PROTOCOL_NAME.length());
                sink.writeString(PROTOCOL_NAME, DEFAULT_CHARSET);
                sink.flush();
                return null;
            }
        });

        futures.receiver = ioExecutors.submit(new Callable<String>() {
            @Override
            public String call() throws Exception {
                byte protocolNameLength = source.readByte();
                String protocolName = source.readString(protocolNameLength, DEFAULT_CHARSET);
                if (!protocolName.equals(PROTOCOL_NAME))
                    throw new IOException("Protocol \"" + protocolName + "\" not supported");
                return protocolName;
            }
        });

        return futures;
    }

    static IOFutures<BitSet> exchangeMessageVectorsAsync(
            final BitSet myMessageVector, final BufferedSource source, final BufferedSink sink, ExecutorService ioExecutors) {
        IOFutures<BitSet> futures = new IOFutures<BitSet>();

        futures.sender = ioExecutors.submit(new Callable<Void>() {
            @Override
            public Void call() throws Exception {
                sink.writeByte(Messages.MESSAGE_VECTOR.getValue());
                sink.write(myMessageVector.toByteArray());
                sink.flush();
                return null;
            }
        });

        futures.receiver = ioExecutors.submit(new Callable<BitSet>() {
            @Override
            public BitSet call() throws Exception {
                byte messageType = source.readByte();
                // TODO: Make an exception type for protocol errors
                if (messageType != Messages.MESSAGE_VECTOR.getValue())
                    throw new IOException("Expected a message vector but got " + messageType);

                byte[] theirMessageVectorByteArray = source.readByteArray(BloomFilter.SIZE_IN_BYTES);
                return BitSet.valueOf(theirMessageVectorByteArray);
            }
        });

        return futures;
    }
}

@@ -284,46 +284,8 @@
 et;%0A
-import java.util.concurrent.Callable;%0A
 impo
@@ -3263,107 +3263,16 @@
 mit(
-new Callable%3CVoid%3E() %7B%0A            @Override%0A            public Void call() throws Exception %7B%0A    
+() -%3E %7B%0A
     
@@ -3323,36 +3323,32 @@
 ));%0A            
-    
 sink.writeString
@@ -3385,36 +3385,32 @@
 T);%0A            
-    
 sink.flush();%0A  
@@ -3411,36 +3411,32 @@
 ();%0A            
-    
 return null;%0A   
@@ -3428,38 +3428,24 @@
 eturn null;%0A
-            %7D%0A
         %7D);%0A
@@ -3495,111 +3495,16 @@
 mit(
-new Callable%3CString%3E() %7B%0A            @Override%0A            public String call() throws Exception %7B%0A    
+() -%3E %7B%0A
     
@@ -3568,20 +3568,16 @@
         
- 
-   
 String p
@@ -3654,28 +3654,24 @@
             
-    
 if (!protoco
@@ -3707,36 +3707,32 @@
                 
- 
-   
 throw new IOExce
@@ -3789,20 +3789,16 @@
 rted%22);%0A
-    
         
@@ -3822,30 +3822,16 @@
 olName;%0A
-            %7D%0A
         
@@ -4161,107 +4161,16 @@
 mit(
-new Callable%3CVoid%3E() %7B%0A            @Override%0A            public Void call() throws Exception %7B%0A    
+() -%3E %7B%0A
     
@@ -4233,36 +4233,32 @@
 ));%0A            
-  
-  
 sink.write(myMes
@@ -4288,36 +4288,32 @@
 ));%0A            
-    
 sink.flush();%0A  
@@ -4314,36 +4314,32 @@
 ();%0A            
-  
-  
 return null;%0A   
@@ -4335,30 +4335,16 @@
 n null;%0A
-            %7D%0A
         
@@ -4398,111 +4398,16 @@
 mit(
-new Callable%3CBitSet%3E() %7B%0A            @Override%0A            public BitSet call() throws Exception %7B%0A    
+() -%3E %7B%0A
     
@@ -4460,28 +4460,24 @@
             
-   
- 
 // TODO: Mak
@@ -4516,20 +4516,16 @@
  errors%0A
-    
         
@@ -4591,36 +4591,32 @@
                 
-   
- 
 throw new IOExce
@@ -4671,28 +4671,24 @@
 sageType);%0A%0A
-    
             
@@ -4781,28 +4781,24 @@
             
-   
- 
 return BitSe
@@ -4841,22 +4841,8 @@
 y);%0A
-            %7D%0A
     

Use lambdas instead of Callables in StreamSync