Merge branch 'master' of sunsh318/pinpoint-tomcat-profiler

from pull request 11

src/main/java/com/nhn/pinpoint/profiler/tools/NetworkAvailabilityChecker.java

@@ -1,9 +1,19 @@
 package com.nhn.pinpoint.profiler.tools;
 
+import java.util.Collections;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import com.nhn.pinpoint.bootstrap.config.ProfilerConfig;
+import com.nhn.pinpoint.profiler.receiver.CommandDispatcher;
 import com.nhn.pinpoint.profiler.sender.DataSender;
 import com.nhn.pinpoint.profiler.sender.TcpDataSender;
 import com.nhn.pinpoint.profiler.sender.UdpDataSender;
+import com.nhn.pinpoint.rpc.PinpointSocketException;
+import com.nhn.pinpoint.rpc.client.MessageListener;
+import com.nhn.pinpoint.rpc.client.PinpointSocket;
+import com.nhn.pinpoint.rpc.client.PinpointSocketFactory;
 
 /**
  * 
@@ -11,6 +21,9 @@ import com.nhn.pinpoint.profiler.sender.UdpDataSender;
  * 
  */
 public class NetworkAvailabilityChecker implements PinpointTools {
+	
+	private static final Logger LOGGER = LoggerFactory.getLogger(NetworkAvailabilityChecker.class);
+	
 	public static void main(String[] args) {
 		if (args.length != 1) {
 			System.out.println("usage : " + NetworkAvailabilityChecker.class.getSimpleName() + " AGENT_CONFIG_FILE");
@@ -22,6 +35,9 @@ public class NetworkAvailabilityChecker implements PinpointTools {
         DataSender udpSender = null;
         DataSender udpSpanSender = null;
         DataSender tcpSender = null;
+        
+        PinpointSocketFactory socketFactory = null;
+        PinpointSocket socket = null;
 		try {
 			ProfilerConfig profilerConfig = new ProfilerConfig();
 			profilerConfig.readConfigFile(configPath);
@@ -33,7 +49,11 @@ public class NetworkAvailabilityChecker implements PinpointTools {
 
 			udpSender = new UdpDataSender(collector, uPort, "UDP", 10);
 			udpSpanSender = new UdpDataSender(collector, usPort, "UDP-SPAN", 10);
-			tcpSender = new TcpDataSender(collector, tPort);
+			
+			socketFactory = createPinpointSocketFactory();
+			socket = createPinpointSocket(collector, tPort, socketFactory);
+			
+			tcpSender = new TcpDataSender(socket);
 
 			boolean udpSenderResult = udpSender.isNetworkAvailable();
 			boolean udpSpanSenderResult = udpSpanSender.isNetworkAvailable();
@@ -53,6 +73,13 @@ public class NetworkAvailabilityChecker implements PinpointTools {
             closeDataSender(udpSpanSender);
             closeDataSender(tcpSender);
 			System.out.println("END.");
+			
+			if (socket != null) {
+				socket.close();
+			}
+			if (socketFactory != null) {
+				socketFactory.release();
+			}
 		}
     }
 
@@ -61,4 +88,33 @@ public class NetworkAvailabilityChecker implements PinpointTools {
             dataSender.stop();
         }
     }
+    
+    private static PinpointSocketFactory createPinpointSocketFactory() {
+    	PinpointSocketFactory pinpointSocketFactory = new PinpointSocketFactory();
+        pinpointSocketFactory.setTimeoutMillis(1000 * 5);
+        pinpointSocketFactory.setAgentProperties(Collections.EMPTY_MAP);
+
+        return pinpointSocketFactory;
+	}
+
+    
+    private static PinpointSocket createPinpointSocket(String host, int port, PinpointSocketFactory factory) {
+    	MessageListener messageListener = new CommandDispatcher();
+    	
+    	PinpointSocket socket = null;
+    	for (int i = 0; i < 3; i++) {
+            try {
+                socket = factory.connect(host, port, messageListener);
+                LOGGER.info("tcp connect success:{}/{}", host, port);
+                return socket;
+            } catch (PinpointSocketException e) {
+            	LOGGER.warn("tcp connect fail:{}/{} try reconnect, retryCount:{}", host, port, i);
+            }
+        }
+    	LOGGER.warn("change background tcp connect mode  {}/{} ", host, port);
+        socket = factory.scheduledConnect(host, port, messageListener);
+    	
+        return socket;
+    }
+    
 }

src/test/java/com/nhn/pinpoint/profiler/HeartBitCheckerStressTest.java

@@ -1,14 +1,19 @@
 package com.nhn.pinpoint.profiler;
 
+import java.util.Collections;
 import java.util.Random;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import com.nhn.pinpoint.thrift.io.HeaderTBaseSerializerFactory;
 import org.apache.thrift.TException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.nhn.pinpoint.profiler.receiver.CommandDispatcher;
 import com.nhn.pinpoint.profiler.sender.TcpDataSender;
+import com.nhn.pinpoint.rpc.PinpointSocketException;
+import com.nhn.pinpoint.rpc.client.MessageListener;
+import com.nhn.pinpoint.rpc.client.PinpointSocket;
+import com.nhn.pinpoint.rpc.client.PinpointSocketFactory;
 import com.nhn.pinpoint.rpc.packet.RequestPacket;
 import com.nhn.pinpoint.rpc.packet.SendPacket;
 import com.nhn.pinpoint.rpc.packet.StreamPacket;
@@ -19,6 +24,7 @@ import com.nhn.pinpoint.rpc.server.SocketChannel;
 import com.nhn.pinpoint.thrift.dto.TAgentInfo;
 import com.nhn.pinpoint.thrift.dto.TResult;
 import com.nhn.pinpoint.thrift.io.HeaderTBaseSerializer;
+import com.nhn.pinpoint.thrift.io.HeaderTBaseSerializerFactory;
 
 public class HeartBitCheckerStressTest {
 
@@ -36,7 +42,10 @@ public class HeartBitCheckerStressTest {
 
 		ResponseServerMessageListener serverListener = new ResponseServerMessageListener(requestCount, successCount);
 
-		TcpDataSender sender = new TcpDataSender(HOST, PORT);
+        PinpointSocketFactory socketFactory = createPinpointSocketFactory();
+        PinpointSocket socket = createPinpointSocket(HOST, PORT, socketFactory);
+
+		TcpDataSender sender = new TcpDataSender(socket);
 		HeartBitChecker checker = new HeartBitChecker(sender, 1000L, getAgentInfo());
 
 		long strarTime = System.currentTimeMillis();
@@ -56,6 +65,14 @@ public class HeartBitCheckerStressTest {
 			if (checker != null) {
 				checker.stop();
 			}
+			
+			if (socket != null) {
+				socket.close();
+			}
+			
+			if (socketFactory != null) {
+				socketFactory.release();
+			}
 		}
 	}
 
@@ -149,4 +166,32 @@ public class HeartBitCheckerStressTest {
 		}
 	}
 
+	 private PinpointSocketFactory createPinpointSocketFactory() {
+	    	PinpointSocketFactory pinpointSocketFactory = new PinpointSocketFactory();
+	        pinpointSocketFactory.setTimeoutMillis(1000 * 5);
+	        pinpointSocketFactory.setAgentProperties(Collections.EMPTY_MAP);
+
+	        return pinpointSocketFactory;
+		}
+
+	    
+	    private PinpointSocket createPinpointSocket(String host, int port, PinpointSocketFactory factory) {
+	    	MessageListener messageListener = new CommandDispatcher();
+	    	
+	    	PinpointSocket socket = null;
+	    	for (int i = 0; i < 3; i++) {
+	            try {
+	                socket = factory.connect(host, port, messageListener);
+	                logger.info("tcp connect success:{}/{}", host, port);
+	                return socket;
+	            } catch (PinpointSocketException e) {
+	            	logger.warn("tcp connect fail:{}/{} try reconnect, retryCount:{}", host, port, i);
+	            }
+	        }
+	    	logger.warn("change background tcp connect mode  {}/{} ", host, port);
+	        socket = factory.scheduledConnect(host, port, messageListener);
+	    	
+	        return socket;
+	    }
+	    
 }

src/test/java/com/nhn/pinpoint/profiler/HeartBitCheckerTest.java

@@ -1,9 +1,11 @@
 package com.nhn.pinpoint.profiler;
 
 
+import java.util.Collections;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import com.nhn.pinpoint.thrift.io.HeaderTBaseSerializerFactory;
+
 import junit.framework.Assert;
 
 import org.apache.thrift.TException;
@@ -11,7 +13,12 @@ import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.nhn.pinpoint.profiler.receiver.CommandDispatcher;
 import com.nhn.pinpoint.profiler.sender.TcpDataSender;
+import com.nhn.pinpoint.rpc.PinpointSocketException;
+import com.nhn.pinpoint.rpc.client.MessageListener;
+import com.nhn.pinpoint.rpc.client.PinpointSocket;
+import com.nhn.pinpoint.rpc.client.PinpointSocketFactory;
 import com.nhn.pinpoint.rpc.packet.RequestPacket;
 import com.nhn.pinpoint.rpc.packet.SendPacket;
 import com.nhn.pinpoint.rpc.packet.StreamPacket;
@@ -38,7 +45,11 @@ public class HeartBitCheckerTest {
 		ResponseServerMessageListener serverListener = new ResponseServerMessageListener(requestCount, successCount);
 
 		PinpointServerSocket server = createServer(serverListener);
-		TcpDataSender sender = new TcpDataSender(HOST, PORT);
+		
+        PinpointSocketFactory socketFactory = createPinpointSocketFactory();
+        PinpointSocket socket = createPinpointSocket(HOST, PORT, socketFactory);
+
+		TcpDataSender sender = new TcpDataSender(socket);
 		HeartBitChecker checker = new HeartBitChecker(sender, 1000L, getAgentInfo());
 
 		try {
@@ -48,7 +59,7 @@ public class HeartBitCheckerTest {
 
 			Assert.assertEquals(1, successCount.get());
 		} finally {
-			closeAll(server, checker);
+			closeAll(server, checker, socket, socketFactory);
 		}
 	}
 
@@ -61,7 +72,10 @@ public class HeartBitCheckerTest {
 
 		PinpointServerSocket server = createServer(serverListener);
 
-		TcpDataSender sender = new TcpDataSender(HOST, PORT);
+        PinpointSocketFactory socketFactory = createPinpointSocketFactory();
+        PinpointSocket socket = createPinpointSocket(HOST, PORT, socketFactory);
+
+        TcpDataSender sender = new TcpDataSender(socket);
 		HeartBitChecker checker = new HeartBitChecker(sender, 1000L, getAgentInfo());
 
 		try {
@@ -72,7 +86,7 @@ public class HeartBitCheckerTest {
 			Assert.assertEquals(2, requestCount.get());
 			Assert.assertEquals(0, successCount.get());
 		} finally {
-			closeAll(server, checker);
+			closeAll(server, checker, socket, socketFactory);
 		}
 	}
 
@@ -83,7 +97,10 @@ public class HeartBitCheckerTest {
 
 		ResponseServerMessageListener serverListener = new ResponseServerMessageListener(requestCount, successCount);
 
-		TcpDataSender sender = new TcpDataSender(HOST, PORT);
+        PinpointSocketFactory socketFactory = createPinpointSocketFactory();
+        PinpointSocket socket = createPinpointSocket(HOST, PORT, socketFactory);
+
+		TcpDataSender sender = new TcpDataSender(socket);
 		HeartBitChecker checker = new HeartBitChecker(sender, 1000L, getAgentInfo());
 
 		try {
@@ -97,9 +114,7 @@ public class HeartBitCheckerTest {
 			
 			Assert.assertEquals(3, successCount.get());
 		} finally {
-			if (checker != null) {
-				checker.stop();
-			}
+			closeAll(null, checker, socket, socketFactory);
 		}
 	}
 
@@ -125,7 +140,7 @@ public class HeartBitCheckerTest {
 		}
 	}
 
-	private void closeAll(PinpointServerSocket server, HeartBitChecker checker) {
+	private void closeAll(PinpointServerSocket server, HeartBitChecker checker, PinpointSocket socket, PinpointSocketFactory factory) {
 		if (server != null) {
 			server.close();
 		}
@@ -133,6 +148,14 @@ public class HeartBitCheckerTest {
 		if (checker != null) {
 			checker.stop();
 		}
+		
+		if (socket != null) {
+			socket.close();
+		}
+		
+		if (factory != null) {
+			factory.release();
+		}
 	}
 
 	private TAgentInfo getAgentInfo() {
@@ -192,5 +215,33 @@ public class HeartBitCheckerTest {
 			logger.info("handleStreamPacket:{}", streamPacket);
 		}
 	}
+	
+	private PinpointSocketFactory createPinpointSocketFactory() {
+    	PinpointSocketFactory pinpointSocketFactory = new PinpointSocketFactory();
+        pinpointSocketFactory.setTimeoutMillis(1000 * 5);
+        pinpointSocketFactory.setAgentProperties(Collections.EMPTY_MAP);
+
+        return pinpointSocketFactory;
+	}
+
+    
+    private PinpointSocket createPinpointSocket(String host, int port, PinpointSocketFactory factory) {
+    	MessageListener messageListener = new CommandDispatcher();
+    	
+    	PinpointSocket socket = null;
+    	for (int i = 0; i < 3; i++) {
+            try {
+                socket = factory.connect(host, port, messageListener);
+                logger.info("tcp connect success:{}/{}", host, port);
+                return socket;
+            } catch (PinpointSocketException e) {
+            	logger.warn("tcp connect fail:{}/{} try reconnect, retryCount:{}", host, port, i);
+            }
+        }
+    	logger.warn("change background tcp connect mode  {}/{} ", host, port);
+        socket = factory.scheduledConnect(host, port, messageListener);
+    	
+        return socket;
+    }
 
 }

src/test/java/com/nhn/pinpoint/profiler/sender/TcpDataSenderReconnectTest.java

@@ -1,6 +1,13 @@
 package com.nhn.pinpoint.profiler.sender;
 
+import java.util.Collections;
+
 import com.nhn.pinpoint.thrift.dto.TApiMetaData;
+import com.nhn.pinpoint.profiler.receiver.CommandDispatcher;
+import com.nhn.pinpoint.rpc.PinpointSocketException;
+import com.nhn.pinpoint.rpc.client.MessageListener;
+import com.nhn.pinpoint.rpc.client.PinpointSocket;
+import com.nhn.pinpoint.rpc.client.PinpointSocketFactory;
 import com.nhn.pinpoint.rpc.packet.RequestPacket;
 import com.nhn.pinpoint.rpc.packet.SendPacket;
 import com.nhn.pinpoint.rpc.packet.StreamPacket;
@@ -8,6 +15,9 @@ import com.nhn.pinpoint.rpc.server.PinpointServerSocket;
 import com.nhn.pinpoint.rpc.server.ServerMessageListener;
 import com.nhn.pinpoint.rpc.server.ServerStreamChannel;
 import com.nhn.pinpoint.rpc.server.SocketChannel;
+
+import net.sf.cglib.proxy.Factory;
+
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -52,7 +62,11 @@ public class TcpDataSenderReconnectTest {
     @Test
     public void connectAndSend() throws InterruptedException {
         PinpointServerSocket old = serverStart();
-        TcpDataSender sender = new TcpDataSender(HOST, PORT);
+
+        PinpointSocketFactory socketFactory = createPinpointSocketFactory();
+        PinpointSocket socket = createPinpointSocket(HOST, PORT, socketFactory);
+
+        TcpDataSender sender = new TcpDataSender(socket);
         Thread.sleep(500);
         old.close();
 
@@ -69,5 +83,35 @@ public class TcpDataSenderReconnectTest {
         sender.stop();
 
         pinpointServerSocket.close();
+        socket.close();
+        socketFactory.release();
+    }
+    
+    private PinpointSocketFactory createPinpointSocketFactory() {
+    	PinpointSocketFactory pinpointSocketFactory = new PinpointSocketFactory();
+        pinpointSocketFactory.setTimeoutMillis(1000 * 5);
+        pinpointSocketFactory.setAgentProperties(Collections.EMPTY_MAP);
+
+        return pinpointSocketFactory;
+	}
+
+    
+    private PinpointSocket createPinpointSocket(String host, int port, PinpointSocketFactory factory) {
+    	MessageListener messageListener = new CommandDispatcher();
+    	
+    	PinpointSocket socket = null;
+    	for (int i = 0; i < 3; i++) {
+            try {
+                socket = factory.connect(host, port, messageListener);
+                logger.info("tcp connect success:{}/{}", host, port);
+                return socket;
+            } catch (PinpointSocketException e) {
+            	logger.warn("tcp connect fail:{}/{} try reconnect, retryCount:{}", host, port, i);
+            }
+        }
+    	logger.warn("change background tcp connect mode  {}/{} ", host, port);
+        socket = factory.scheduledConnect(host, port, messageListener);
+    	
+        return socket;
     }
 }

src/test/java/com/nhn/pinpoint/profiler/sender/TcpDataSenderTest.java

@@ -1,6 +1,11 @@
 package com.nhn.pinpoint.profiler.sender;
 
 import com.nhn.pinpoint.thrift.dto.TApiMetaData;
+import com.nhn.pinpoint.profiler.receiver.CommandDispatcher;
+import com.nhn.pinpoint.rpc.PinpointSocketException;
+import com.nhn.pinpoint.rpc.client.MessageListener;
+import com.nhn.pinpoint.rpc.client.PinpointSocket;
+import com.nhn.pinpoint.rpc.client.PinpointSocketFactory;
 import com.nhn.pinpoint.rpc.packet.RequestPacket;
 import com.nhn.pinpoint.rpc.packet.SendPacket;
 import com.nhn.pinpoint.rpc.packet.StreamPacket;
@@ -8,13 +13,16 @@ import com.nhn.pinpoint.rpc.server.PinpointServerSocket;
 import com.nhn.pinpoint.rpc.server.ServerMessageListener;
 import com.nhn.pinpoint.rpc.server.ServerStreamChannel;
 import com.nhn.pinpoint.rpc.server.SocketChannel;
+
 import junit.framework.Assert;
+
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Collections;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -68,7 +76,10 @@ public class TcpDataSenderTest {
     public void connectAndSend() throws InterruptedException {
         this.sendLatch = new CountDownLatch(2);
 
-        TcpDataSender sender = new TcpDataSender(HOST, PORT);
+        PinpointSocketFactory socketFactory = createPinpointSocketFactory();
+        PinpointSocket socket = createPinpointSocket(HOST, PORT, socketFactory);
+        
+        TcpDataSender sender = new TcpDataSender(socket);
         try {
             sender.send(new TApiMetaData("test", System.currentTimeMillis(), 1, "TestApi"));
             sender.send(new TApiMetaData("test", System.currentTimeMillis(), 1, "TestApi"));
@@ -78,8 +89,42 @@ public class TcpDataSenderTest {
             Assert.assertTrue(received);
         } finally {
             sender.stop();
+            
+            if (socket != null) {
+            	socket.close();
+            }
+            
+            if (socketFactory != null) {
+            	socketFactory.release();
+            }
         }
+    }
+    
+    private PinpointSocketFactory createPinpointSocketFactory() {
+    	PinpointSocketFactory pinpointSocketFactory = new PinpointSocketFactory();
+        pinpointSocketFactory.setTimeoutMillis(1000 * 5);
+        pinpointSocketFactory.setAgentProperties(Collections.EMPTY_MAP);
 
+        return pinpointSocketFactory;
+	}
 
+    
+    private PinpointSocket createPinpointSocket(String host, int port, PinpointSocketFactory factory) {
+    	MessageListener messageListener = new CommandDispatcher();
+    	
+    	PinpointSocket socket = null;
+    	for (int i = 0; i < 3; i++) {
+            try {
+                socket = factory.connect(host, port, messageListener);
+                logger.info("tcp connect success:{}/{}", host, port);
+                return socket;
+            } catch (PinpointSocketException e) {
+            	logger.warn("tcp connect fail:{}/{} try reconnect, retryCount:{}", host, port, i);
+            }
+        }
+    	logger.warn("change background tcp connect mode  {}/{} ", host, port);
+        socket = factory.scheduledConnect(host, port, messageListener);
+    	
+        return socket;
     }
 }

src/test/java/com/nhn/pinpoint/profiler/util/MockAgent.java

@@ -29,11 +29,6 @@ public class MockAgent extends DefaultAgent {
         return super.createUdpDataSender(port, threadName, writeQueueSize, timeout, sendBufferSize);
     }
 
-    @Override
-    protected EnhancedDataSender createTcpDataSender() {
-        return new LoggingDataSender();
-    }
-    
     @Override 
     protected StorageFactory createStorageFactory() {
 		return new ReadableSpanStorageFactory();