package org.ethereum.net.eth.handler;

import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import io.netty.channel.ChannelHandlerContext;
import org.apache.commons.lang3.tuple.Pair;
import org.ethereum.config.SystemProperties;
import org.ethereum.core.Block;
import org.ethereum.core.Blockchain;
import org.ethereum.core.Transaction;
import org.ethereum.core.TransactionInfo;
import org.ethereum.core.TransactionReceipt;
import org.ethereum.datasource.test.RepositoryImpl;
import org.ethereum.listener.CompositeEthereumListener;
import org.ethereum.net.eth.EthVersion;
import org.ethereum.net.eth.message.EthMessage;
import org.ethereum.net.eth.message.GetNodeDataMessage;
import org.ethereum.net.eth.message.GetReceiptsMessage;
import org.ethereum.net.eth.message.NodeDataMessage;
import org.ethereum.net.eth.message.ReceiptsMessage;

import org.ethereum.sync.SyncState;
import org.ethereum.util.ByteArraySet;
import org.ethereum.util.Value;
import org.spongycastle.util.encoders.Hex;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;

import static org.ethereum.crypto.HashUtil.sha3;
import static org.ethereum.net.eth.EthVersion.V63;

/**
 * Fast synchronization (PV63) Handler
 */
@Component
@Scope("prototype")
public class Eth63 extends Eth62 {

    private static final EthVersion version = V63;

    @Autowired
    private RepositoryImpl repository;

    private Set<byte[]> requestedNodes;
    private SettableFuture<List<Pair<byte[], byte[]>>> requestNodesFuture;

    public Eth63() {
        super(version);
    }

    @Autowired
    public Eth63(final SystemProperties config, final Blockchain blockchain,
                 final CompositeEthereumListener ethereumListener) {
        super(version, config, blockchain, ethereumListener);
    }

    @Override
    public void channelRead0(final ChannelHandlerContext ctx, EthMessage msg) throws InterruptedException {

        super.channelRead0(ctx, msg);

        // Only commands that were added in V63, V62 are handled in child
        switch (msg.getCommand()) {
            case GET_NODE_DATA:
                processGetNodeData((GetNodeDataMessage) msg);
                break;
            case NODE_DATA:
                processNodeData((NodeDataMessage) msg);
                break;
            case GET_RECEIPTS:
                processGetReceipts((GetReceiptsMessage) msg);
                break;
            case RECEIPTS:
                // TODO: Implement
                break;
            default:
                break;
        }
    }

    protected synchronized void processGetNodeData(GetNodeDataMessage msg) {

        if (logger.isTraceEnabled()) logger.trace(
                "Peer {}: processing GetNodeData, size [{}]",
                channel.getPeerIdShort(),
                msg.getStateRoots().size()
        );

        List<Value> states = new ArrayList<>();
        for (byte[] stateRoot : msg.getStateRoots()) {
            Value value = repository.getState(stateRoot);
            if (value != null) {
                states.add(value);
                logger.trace("Eth63: " + Hex.toHexString(stateRoot).substring(0, 8) + " -> " + value);
            } else {
            }
        }

        sendMessage(new NodeDataMessage(states));
    }

    protected synchronized void processGetReceipts(GetReceiptsMessage msg) {

        if (logger.isTraceEnabled()) logger.trace(
                "Peer {}: processing GetReceipts, size [{}]",
                channel.getPeerIdShort(),
                msg.getBlockHashes().size()
        );

        List<List<TransactionReceipt>> receipts = new ArrayList<>();
        for (byte[] blockHash : msg.getBlockHashes()) {
            Block block = blockchain.getBlockByHash(blockHash);
            if (block == null) continue;

            List<TransactionReceipt> blockReceipts = new ArrayList<>();
            for (Transaction transaction : block.getTransactionsList()) {
                TransactionInfo transactionInfo = blockchain.getTransactionInfo(transaction.getHash());
                blockReceipts.add(transactionInfo.getReceipt());
            }
            receipts.add(blockReceipts);
        }

        sendMessage(new ReceiptsMessage(receipts));
    }

    public synchronized ListenableFuture<List<Pair<byte[], byte[]>>> requestTrieNodes(List<byte[]> hashes) {
        GetNodeDataMessage msg = new GetNodeDataMessage(hashes);
        requestedNodes = new ByteArraySet();
        requestedNodes.addAll(hashes);
        requestNodesFuture = SettableFuture.create();
        syncState = SyncState.NODE_RETRIEVING;
        sendMessage(msg);
        return requestNodesFuture;
    }

    protected synchronized void processNodeData(NodeDataMessage msg) {
        if (requestedNodes == null) {
            logger.debug("Received NodeDataMessage when requestedNodes == null. Dropping peer");
            dropConnection();
        }

        List<Pair<byte[], byte[]>> ret = new ArrayList<>();
        for (Value nodeVal : msg.getDataList()) {
            byte[] hash = nodeVal.hash();
            if (!requestedNodes.contains(hash)) {
                String err = "Received NodeDataMessage contains non-requested node with hash :" + Hex.toHexString(hash) + " . Dropping peer";
                logger.debug(err);
                requestNodesFuture.setException(new RuntimeException(err));
                dropConnection();
                return;
            }
            ret.add(Pair.of(hash, nodeVal.encode()));
        }
        requestNodesFuture.set(ret);

        requestedNodes = null;
        requestNodesFuture = null;
        syncState = SyncState.IDLE;
    }
}

@@ -1717,24 +1717,157 @@
 desFuture;%0A%0A
+    private long connectedTime = System.currentTimeMillis();%0A    private long processingTime = 0;%0A    private long lastReqSentTime;%0A%0A
     public E
@@ -4958,24 +4958,78 @@
 ssage(msg);%0A
+        lastReqSentTime = System.currentTimeMillis();%0A
         retu
@@ -6012,34 +6012,755 @@
     
-syncState = SyncState.IDLE
+processingTime += (System.currentTimeMillis() - lastReqSentTime);%0A        syncState = SyncState.IDLE;%0A    %7D%0A%0A    @Override%0A    public String getSyncStats() %7B%0A        double nodesPerSec = 1000d * channel.getNodeStatistics().eth63NodesReceived.get() / channel.getNodeStatistics().eth63NodesRetrieveTime.get();%0A        double missNodesRatio = 1 - (double) channel.getNodeStatistics().eth63NodesReceived.get() / channel.getNodeStatistics().eth63NodesRequested.get();%0A        long lifeTime = System.currentTimeMillis() - connectedTime;%0A        return super.getSyncStats() + String.format(%22%5CtNodes/sec: %251$.2f, miss: %252$.2f%22, nodesPerSec, missNodesRatio) +%0A                %22%5CtLife: %22 + lifeTime / 1000 + %22s,%5CtIdle: %22 + (lifeTime - processingTime) + %22ms%22
 ;%0A  

Add more Eth63 peer stats