package xnioredis;

import com.google.common.base.Throwables;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import org.xnio.ChannelListener;
import org.xnio.IoUtils;
import org.xnio.Pool;
import org.xnio.Pooled;
import org.xnio.StreamConnection;
import org.xnio.channels.StreamSinkChannel;
import org.xnio.channels.StreamSourceChannel;
import xnioredis.decoder.parser.ReplyParser;

import javax.annotation.Nullable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.CharsetEncoder;
import java.util.Arrays;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import static java.nio.charset.StandardCharsets.UTF_8;

class XnioRedisClient extends RedisClient {
    private final Pool<ByteBuffer> bufferPool;
    private final BlockingQueue<CommandWriter> writerQueue = new LinkedBlockingQueue<>();
    private final BlockingQueue<ReplyDecoder> decoderQueue = new LinkedBlockingQueue<>();
    private final CharsetEncoder charsetEncoder = UTF_8.newEncoder();
    private final StreamConnection connection;
    private final StreamSinkChannel outChannel;
    private final StreamSourceChannel inChannel;
    private ReplyDecoder currentDecoder;
    private CommandWriter currentWriter;

    public XnioRedisClient(StreamConnection connection, Pool<ByteBuffer> bufferPool) {
        this.connection = connection;
        this.bufferPool = bufferPool;
        this.outChannel = connection.getSinkChannel();
        this.inChannel = connection.getSourceChannel();
        this.inChannel.getReadSetter().set(inChannel -> {
            try (Pooled<ByteBuffer> pooledByteBuffer = bufferPool.allocate()) {
                ByteBuffer readBuffer = pooledByteBuffer.getResource();
                while (inChannel.read(readBuffer) > 0) {
                    readBuffer.flip();
                    try {
                        while (readBuffer.hasRemaining()) {
                            decoder().parse(readBuffer);
                        }
                    } finally {
                        readBuffer.clear();
                    }
                }
            } catch (Throwable e) {
                decoder().fail(e);
            }
        });
        this.inChannel.resumeReads();
        this.outChannel.getWriteSetter().set(new ChannelListener<StreamSinkChannel>() {
            @Override
            public void handleEvent(StreamSinkChannel outChannel) {
                CommandWriter commandWriter;
                while ((commandWriter = writer()) != null) {
                    try {
                        if (commandWriter.write(outChannel, charsetEncoder, XnioRedisClient.this.bufferPool)) {
                            currentWriter = null;
                        } else {
                            return;
                        }
                    } catch (IOException e) {
                        throw Throwables.propagate(e);
                    }
                }
                outChannel.suspendWrites();
            }
        });
    }

    private ReplyDecoder decoder() {
        if (currentDecoder == null) {
            currentDecoder = decoderQueue.poll();
            if (currentDecoder == null) {
                currentDecoder = new ReplyDecoder() {
                    @Override
                    public void parse(ByteBuffer buffer) throws IOException {
                        int len = buffer.remaining();
                        byte[] bytes = new byte[len];
                        buffer.get(bytes);
                        throw new IllegalStateException("Unexpected input: " + Arrays.toString(bytes));
                    }

                    @Override
                    public void fail(Throwable e) {
                        throw Throwables.propagate(e);
                    }
                };
            }
        }
        return currentDecoder;
    }

    private CommandWriter writer() {
        if (currentWriter == null) {
            currentWriter = writerQueue.poll();
        }
        return currentWriter;
    }

    @Override
    public <T> ListenableFuture<T> send(Command<T> command, boolean autoFlush) {
        SettableFuture<T> future = SettableFuture.create();
        decoderQueue.add(new ReplyDecoder() {
            private ReplyParser<? extends T> parser = command.parser();

            @Override
            public void parse(ByteBuffer buffer) throws IOException {
                parser.parseReply(buffer, new ReplyParser.ReplyVisitor<T, Void>() {
                    @Override
                    public Void success(@Nullable T value) {
                        setReply(value);
                        return null;
                    }

                    @Override
                    public Void failure(CharSequence message) {
                        fail(new RedisException(message.toString()));
                        return null;
                    }

                    @Override
                    public Void partialReply(ReplyParser<? extends T> partial) {
                        parser = partial;
                        return null;
                    }
                });
            }

            private void setReply(@Nullable T reply) {
                future.set(reply);
                currentDecoder = null;
            }

            @Override
            public void fail(Throwable e) {
                future.setException(e);
                currentDecoder = null;
            }
        });
        writerQueue.add(command.writer());
        outChannel.resumeWrites();
        return future;
    }

    @Override
    public ListenableFuture<Void> flush() {
        SettableFuture<Void> future = SettableFuture.create();
        outChannel.getIoThread().execute(() -> {
            try {
                outChannel.flush();
                future.set(null);
            } catch (IOException e) {
                future.setException(e);
            }
        });
        return future;
    }

    @Override
    public void close() {
        IoUtils.safeClose(inChannel);
        IoUtils.safeClose(outChannel);
        IoUtils.safeClose(connection);
    }
}

@@ -175,41 +175,8 @@
 re;%0A
-import org.xnio.ChannelListener;%0A
 impo
@@ -764,55 +764,8 @@
 t %7B%0A
-    private final Pool%3CByteBuffer%3E bufferPool;%0A
     
@@ -1366,46 +1366,8 @@
 on;%0A
-        this.bufferPool = bufferPool;%0A
     
@@ -2248,127 +2248,8 @@
 set(
-new ChannelListener%3CStreamSinkChannel%3E() %7B%0A            @Override%0A            public void handleEvent(StreamSinkChannel 
 outC
@@ -2254,24 +2254,22 @@
 tChannel
-) %7B%0A    
+ -%3E %7B%0A
         
@@ -2301,20 +2301,16 @@
 Writer;%0A
-    
         
@@ -2370,38 +2370,30 @@
             
- 
-   
 try %7B%0A
-    
             
@@ -2456,29 +2456,8 @@
 er, 
-XnioRedisClient.this.
 buff
@@ -2463,28 +2463,24 @@
 ferPool)) %7B%0A
-    
             
@@ -2533,20 +2533,16 @@
         
-    
 %7D else %7B
@@ -2562,28 +2562,24 @@
             
-   
- 
 return;%0A    
@@ -2590,32 +2590,26 @@
             
-    %7D%0A  
+%7D%0A
             
@@ -2604,34 +2604,32 @@
                 
-  
 %7D catch (IOExcep
@@ -2630,36 +2630,32 @@
 OException e) %7B%0A
-    
                 
@@ -2701,34 +2701,26 @@
             
-    %7D%0A    
+%7D%0A
             
@@ -2713,36 +2713,32 @@
 %7D%0A            %7D%0A
-    
             outC
@@ -2761,30 +2761,16 @@
 ites();%0A
-            %7D%0A
         

Replace anonymous class with lambda