package com.basho.riak.client.core.operations.ts;

import com.basho.riak.client.core.codec.TermToBinaryCodec;
import com.basho.riak.client.core.operations.TTBFutureOperation;
import com.basho.riak.client.core.query.timeseries.Cell;
import com.basho.riak.client.core.query.timeseries.QueryResult;
import com.basho.riak.protobuf.RiakMessageCodes;
import com.ericsson.otp.erlang.OtpErlangDecodeException;
import com.ericsson.otp.erlang.OtpOutputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;

import java.util.LinkedList;
import java.util.logging.Level;
import java.util.logging.Logger;

class TTBConverters
{
    private static abstract class MemoizingEncoder<T> implements TTBFutureOperation.TTBEncoder
    {
        // private final byte reqMessageCode = RiakMessageCodes.MSG_TsTtbMsg;
        protected byte[] message = null;
        protected final T builder;

        MemoizingEncoder(T builder)
        {
            this.builder = builder;
        }

        abstract OtpOutputStream buildMessage();

        @Override
        public byte[] build()
        {
            if (message == null)
            {
                try {
                    OtpOutputStream os = buildMessage();
                    os.flush();
                    
                    // TODO GH-611 should the output stream or base type be returned?
                    // TODO GH-611 we still need to add msgCode 141 and message length
                    final int msgLen = os.length() + 5; // Add 4-byte msg length and 1-byte msgCode
                    ByteArrayOutputStream bs = new ByteArrayOutputStream(msgLen);
                    
                    DataOutputStream ds = new DataOutputStream(bs);
                    ds.writeInt(msgLen);
                    ds.writeByte(RiakMessageCodes.MSG_TsTtbMsg);
                    ds.flush();
                    
                    os.writeTo(bs);
                    bs.flush();
                    message = bs.toByteArray();
                } catch (IOException ex) {
                    // TODO GH-611
                    Logger.getLogger(TTBConverters.class.getName()).log(Level.SEVERE, null, ex);
                }
            }

            return message;
        }
    }

    static class StoreEncoder extends MemoizingEncoder<StoreOperation.Builder>
    {
        StoreEncoder(StoreOperation.Builder builder)
        {
            super(builder);
        }

        @Override
        OtpOutputStream buildMessage()
        {
            return TermToBinaryCodec.encodeTsPutRequest(builder.getTableName(), builder.getRows());
        }
    }

    static class FetchEncoder extends MemoizingEncoder<FetchOperation.Builder>
    {
        FetchEncoder(FetchOperation.Builder builder)
        {
            super(builder);
        }

        @Override
        OtpOutputStream buildMessage()
        {
            // TODO: Remove this later
            LinkedList<Cell> list = new LinkedList<>();
            for (Cell c : builder.getKeyValues())
            {
                list.add(c);
            }
            return TermToBinaryCodec.encodeTsGetRequest(builder.getTableName(),
                                                        list,
                                                        builder.getTimeout());
        }
    }

    static class QueryEncoder extends MemoizingEncoder<QueryOperation.Builder>
    {
        QueryEncoder(QueryOperation.Builder builder)
        {
            super(builder);
        }

        @Override
        OtpOutputStream buildMessage()
        {
            return TermToBinaryCodec.encodeTsQueryRequest(builder.getQueryText());
        }
    }

    static class VoidDecoder implements TTBFutureOperation.TTBParser<Void>
    {
        @Override
        public Void parseFrom(byte[] data)
        {
            return null;
        }
    }

    static class QueryResultDecoder implements TTBFutureOperation.TTBParser<QueryResult>
    {
        @Override
        public QueryResult parseFrom(byte[] data)
        {
            QueryResult rv = null;

            try {
                rv = TermToBinaryCodec.decodeTsGetResponse(data);
            } catch (OtpErlangDecodeException ex) {
                // TODO GH-611
                Logger.getLogger(TTBConverters.class.getName()).log(Level.SEVERE, null, ex);
            }

            return rv;
        }
    }
}

@@ -293,57 +293,8 @@
 lt;%0A
-import com.basho.riak.protobuf.RiakMessageCodes;%0A
 impo
@@ -398,79 +398,8 @@
 am;%0A
-import java.io.ByteArrayOutputStream;%0Aimport java.io.DataOutputStream;%0A
 impo
@@ -422,17 +422,16 @@
 eption;%0A
-%0A
 import j
@@ -644,86 +644,8 @@
   %7B%0A
-        // private final byte reqMessageCode = RiakMessageCodes.MSG_TsTtbMsg;%0A
     
@@ -1100,711 +1100,19 @@
     
-%0A                    // TODO GH-611 should the output stream or base type be returned?%0A                    // TODO GH-611 we still need to add msgCode 141 and message length%0A                    final int msgLen = os.length() + 5; // Add 4-byte msg length and 1-byte msgCode%0A                    ByteArrayOutputStream bs = new ByteArrayOutputStream(msgLen);%0A                    %0A                    DataOutputStream ds = new DataOutputStream(bs);%0A                    ds.writeInt(msgLen);%0A                    ds.writeByte(RiakMessageCodes.MSG_TsTtbMsg);%0A                    ds.flush();%0A                    %0A                    os.writeTo(bs);%0A                    bs.flush();%0A                    message = b
+message = o
 s.to

Remove the double-header from TTB data.