package com.nhn.pinpoint.collector.receiver.tcp;

import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;

import org.apache.thrift.TBase;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import com.nhn.pinpoint.collector.cluster.zookeeper.ZookeeperClusterService;
import com.nhn.pinpoint.collector.receiver.TcpDispatchHandler;
import com.nhn.pinpoint.rpc.packet.Packet;
import com.nhn.pinpoint.rpc.packet.RequestPacket;
import com.nhn.pinpoint.rpc.packet.ResponsePacket;
import com.nhn.pinpoint.rpc.packet.SendPacket;
import com.nhn.pinpoint.rpc.server.DoNothingChannelStateEventListener;
import com.nhn.pinpoint.thrift.dto.TAgentInfo;
import com.nhn.pinpoint.thrift.dto.TResult;
import com.nhn.pinpoint.thrift.io.HeaderTBaseDeserializer;
import com.nhn.pinpoint.thrift.io.HeaderTBaseDeserializerFactory;
import com.nhn.pinpoint.thrift.io.HeaderTBaseSerializer;
import com.nhn.pinpoint.thrift.io.HeaderTBaseSerializerFactory;


/**
 * @author koo.taejin
 */
@ContextConfiguration("classpath:applicationContext-collector.xml")
@RunWith(SpringJUnit4ClassRunner.class)
public class TCPReceiverBOTest {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());

	@Autowired
	TcpDispatchHandler handler;

	private TCPReceiver tcpReceiver;
	
    @Before
    public void setUp() {
    	ZookeeperClusterService service = mock(ZookeeperClusterService.class);
    	when(service.getChannelStateChangeEventListener()).thenReturn(DoNothingChannelStateEventListener.INSTANCE);
    	
    	tcpReceiver = new TCPReceiver(handler, "127.0.0.1", 9994, service);
    	tcpReceiver.start();
    }
    
    @After
    public void tearDown() {
    	tcpReceiver.stop();
    }
    
	@Test
	public void agentInfoTest1() throws Exception {
		Socket socket = connectTcpReceiver();
		OutputStream os = socket.getOutputStream();
		InputStream is = socket.getInputStream();

		TAgentInfo agentInfo = getAgentInfo();
		encodeAndWrite(os, agentInfo, false);
		ResponsePacket responsePacket = readAndDecode(is, 1000);
		Assert.assertNull(responsePacket);

	}

	@Test
	public void agentInfoTest2() throws Exception {
		Socket socket = connectTcpReceiver();
		OutputStream os = socket.getOutputStream();
		InputStream is = socket.getInputStream();

		TAgentInfo agentInfo = getAgentInfo();
		encodeAndWrite(os, agentInfo, true);
		ResponsePacket responsePacket = readAndDecode(is, 1000);

        HeaderTBaseDeserializer deserializer = new HeaderTBaseDeserializerFactory().createDeserializer();
		TResult result = (TResult) deserializer.deserialize(responsePacket.getPayload());
		
		Assert.assertTrue(result.isSuccess());
	}

	private Socket connectTcpReceiver() throws IOException {
		Socket socket = new Socket();
		socket.connect(new InetSocketAddress("127.0.0.1", 9994));

		return socket;
	}

	private void encodeAndWrite(OutputStream os, TBase tbase, boolean isReqRes) throws Exception {
		HeaderTBaseSerializer serializer = HeaderTBaseSerializerFactory.DEFAULT_FACTORY.createSerializer();
		byte[] payload = serializer.serialize(tbase);

		Packet packet = null;
		if (isReqRes) {
			packet = new RequestPacket(payload);
		} else {
			packet = new SendPacket(payload);
		}

		os.write(packet.toBuffer().toByteBuffer().array());
		os.flush();
	}

	private ResponsePacket readAndDecode(InputStream is, long waitTimeMillis) throws Exception {
		long startTimeMillis = System.currentTimeMillis();

		while (true) {
			int avaiableRead = is.available();

			if (avaiableRead > 0) {
				byte[] payload = new byte[avaiableRead];
				is.read(payload);

				for (byte b : payload) {
					logger.warn("!!!{}", b);
				}
				
				ChannelBuffer cb = ChannelBuffers.wrappedBuffer(payload);
				cb.readByte();
				cb.readByte();
				
				
				ResponsePacket responsePacket = ResponsePacket.readBuffer((short) 6, cb);
				return responsePacket;
			}

			Thread.sleep(20);
			if (waitTimeMillis < System.currentTimeMillis() - startTimeMillis) {
				return null;
			}
		}
	}

	private TAgentInfo getAgentInfo() {
		TAgentInfo agentInfo = new TAgentInfo("hostname", "127.0.0.1", "8081", "agentId", "appName", (short) 2, 1111, "1", System.currentTimeMillis());
		return agentInfo;
	}

}

@@ -47,89 +47,8 @@
 p;%0A%0A
-import static org.mockito.Mockito.mock;%0Aimport static org.mockito.Mockito.when;%0A%0A
 impo
@@ -67,24 +67,24 @@
 OException;%0A
+
 import java.
@@ -315,32 +315,8 @@
 rs;%0A
-import org.junit.After;%0A
 impo
@@ -340,63 +340,8 @@
 rt;%0A
-import org.junit.Before;%0Aimport org.junit.BeforeClass;%0A
 impo
@@ -651,85 +651,8 @@
 r;%0A%0A
-import com.nhn.pinpoint.collector.cluster.zookeeper.ZookeeperClusterService;%0A
 impo
@@ -905,79 +905,8 @@
 et;%0A
-import com.nhn.pinpoint.rpc.server.DoNothingChannelStateEventListener;%0A
 impo
@@ -1530,16 +1530,28 @@
 ndler;%0A%0A
+%09@Autowired%0A
 %09private
@@ -1582,426 +1582,9 @@
 ;%0A%09%0A
-    @Before%0A    public void setUp() %7B%0A    %09ZookeeperClusterService service = mock(ZookeeperClusterService.class);%0A    %09when(service.getChannelStateChangeEventListener()).thenReturn(DoNothingChannelStateEventListener.INSTANCE);%0A    %09%0A    %09tcpReceiver = new TCPReceiver(handler, %22127.0.0.1%22, 9994, service);%0A    %09tcpReceiver.start();%0A    %7D%0A    %0A    @After%0A    public void tearDown() %7B%0A    %09tcpReceiver.stop();%0A    %7D%0A    
+%09
 %0A%09@T

#34 CI 테스트 FAIL 나는 것 수정