package com.feeyo.kafka.net.backend.callback;

import java.nio.ByteBuffer;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.feeyo.kafka.codec.Errors;
import com.feeyo.kafka.codec.FetchResponse;
import com.feeyo.kafka.codec.Record;
import com.feeyo.kafka.net.backend.broker.BrokerApiVersion;
import com.feeyo.kafka.net.backend.broker.offset.BrokerOffsetService;
import com.feeyo.kafka.protocol.ApiKeys;
import com.feeyo.kafka.protocol.types.Struct;
import com.feeyo.net.nio.NetSystem;
import com.feeyo.net.nio.util.ProtoUtils;
import com.feeyo.redis.net.backend.BackendConnection;
import com.feeyo.redis.net.front.RedisFrontConnection;

public class KafkaConsumerCmdCallback extends KafkaCmdCallback {
	
	private static Logger LOGGER = LoggerFactory.getLogger( KafkaConsumerCmdCallback.class );
	
	private String topic;
	private long consumeOffset;
	private int partition;
	
	// 消费失败是否把消费点位归还（指定点位消费时，不需要归还）
	private boolean isErrorOffsetRecovery = true;
	
	public KafkaConsumerCmdCallback(String topic, int partition, long offset, boolean isErrorOffsetRecovery) {
		this.topic = topic;
		this.partition = partition;
		this.consumeOffset = offset;
		this.isErrorOffsetRecovery = isErrorOffsetRecovery;
	}
	
	@Override
	public void parseResponseBody(BackendConnection conn, ByteBuffer respBuffer) {
		
		//
		RedisFrontConnection frontCon = getFrontCon( conn );
		
		short version = BrokerApiVersion.getConsumerVersion();
		
		Struct response = ApiKeys.FETCH.parseResponse(version, respBuffer);
		FetchResponse fr = new FetchResponse(response);
		if (fr.isCorrect()) {
			List<Record> records = fr.getRecords();
			if (records == null || records.isEmpty()) {
				if ( isErrorOffsetRecovery )
					returnConsumerOffset(frontCon.getPassword(), topic, partition, consumeOffset);
				
				frontCon.write(NULL);
				return;
			}
			
			byte[] size = ProtoUtils.convertIntToByteArray(CONSUMER_RESPONSE_SIZE * records.size());
			
			for (int i = 0;i<records.size();i++) {
				Record record = records.get(i);
				byte[] value = record.getValue();
				
				if (value == null) {
					if ( isErrorOffsetRecovery )
						returnConsumerOffset(frontCon.getPassword(), topic, partition, consumeOffset);
					
					frontCon.write(NULL);
					return;
				}
				byte[] partitonArr = ProtoUtils.convertIntToByteArray(partition);
				byte[] partitonLength = ProtoUtils.convertIntToByteArray(partitonArr.length);
				byte[] offsetArr = String.valueOf(record.getOffset()).getBytes();
				byte[] offsetLength = ProtoUtils.convertIntToByteArray(offsetArr.length);
				byte[] valueLenght = ProtoUtils.convertIntToByteArray(value.length);
				
				// 计算 bufferSize $1\r\n1\r\n$4\r\n2563\r\n$4\r\ntest\r\n
				int bufferSize = 1 + size.length + 2 
						+ 1 + partitonLength.length + 2 + partitonArr.length + 2
						+ 1 + offsetLength.length + 2 + offsetArr.length + 2 
						+ 1 + valueLenght.length + 2 + value.length + 2;
				ByteBuffer bb = NetSystem.getInstance().getBufferPool().allocate(bufferSize);
				if (i == 0) {
					bb.put(ASTERISK).put(size).put(CRLF);
				}
				bb.put(DOLLAR).put(partitonLength).put(CRLF).put(partitonArr).put(CRLF)
				.put(DOLLAR).put(offsetLength).put(CRLF).put(offsetArr).put(CRLF)
				.put(DOLLAR).put(valueLenght).put(CRLF).put(value).put(CRLF);
				frontCon.write(bb);
			}
		
		// 消费offset超出范围
		} else if (fr.getFetchErr() != null && fr.getFetchErr().getCode() == Errors.OFFSET_OUT_OF_RANGE.code()) {
			
			LOGGER.warn("consume callback fr err: fr range={} {}, msg={}", new Object[]{ fr.getLogStartOffset() , fr.getLastStableOffset() , fr.getErrorMessage()} );
			if ( isErrorOffsetRecovery )
				returnConsumerOffset(frontCon.getPassword(), topic, partition, consumeOffset);
			
			frontCon.write(NULL);
			
		// 其他错误
		} else {
			
			if ( isErrorOffsetRecovery )
				returnConsumerOffset(frontCon.getPassword(), topic, partition, consumeOffset);
			
			StringBuffer sb = new StringBuffer();
			sb.append("-ERR ").append(fr.getErrorMessage()).append("\r\n");
			frontCon.write(sb.toString().getBytes());
		}
	}
	
	private void returnConsumerOffset(String password, String topic, int partition, long offset) {
		BrokerOffsetService.INSTANCE().returnOffset(password, topic, partition, offset);
	}
}

@@ -869,50 +869,50 @@
 ate 
-long consumeOffset;%0A%09private int partition
+int partition;%0A%09private long consumeOffset
 ;%0A%09%0A
@@ -3499,18 +3499,51 @@
 fr err: 
-fr
+topic=%7B%7D, partition=%7B%7D, offset=%7B%7D, 
  range=%7B
@@ -3543,16 +3543,17 @@
 range=%7B%7D
+,
  %7B%7D, msg
@@ -3558,16 +3558,22 @@
 sg=%7B%7D%22, 
+%0A%09%09%09%09%09
 new Obje
@@ -3577,16 +3577,50 @@
 bject%5B%5D%7B
+  topic, partition, consumeOffset,
  fr.getL
@@ -3689,16 +3689,20 @@
 e()%7D );%0A
+%09%09%09%0A
 %09%09%09if ( 

add log info