Merge branch 'master' of sunsh318/pinpoint-2

from pull-request 210

* refs/heads/master:
  #22 feature: Allow stream data transfer in pinpointsocket.

profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java

@@ -30,6 +30,8 @@ import com.nhn.pinpoint.profiler.interceptor.bci.JavaAssistByteCodeInstrumentor;
 import com.nhn.pinpoint.profiler.logging.Slf4jLoggerBinder;
 import com.nhn.pinpoint.profiler.monitor.AgentStatMonitor;
 import com.nhn.pinpoint.profiler.receiver.CommandDispatcher;
+import com.nhn.pinpoint.profiler.receiver.service.EchoService;
+import com.nhn.pinpoint.profiler.receiver.service.ThreadDumpService;
 import com.nhn.pinpoint.profiler.sampler.SamplerFactory;
 import com.nhn.pinpoint.profiler.sender.BufferedUdpDataSender;
 import com.nhn.pinpoint.profiler.sender.DataSender;
@@ -119,7 +121,11 @@ public class DefaultAgent implements Agent {
         this.agentInformation = agentInformationFactory.createAgentInformation(typeResolver.getServerType());
         logger.info("agentInformation:{}", agentInformation);
 
-        this.factory = createPinpointSocketFactory(this.profilerConfig.isTcpDataSenderCommandAcceptEnable());
+        CommandDispatcher commandDispatcher = new CommandDispatcher();
+        commandDispatcher.registerCommandService(new ThreadDumpService());
+        commandDispatcher.registerCommandService(new EchoService());
+        
+        this.factory = createPinpointSocketFactory(commandDispatcher);
         this.socket = createPinpointSocket(this.profilerConfig.getCollectorTcpServerIp(), this.profilerConfig.getCollectorTcpServerPort(), factory);
 
         this.serverMetaDataHolder = createServerMetaDataHolder();
@@ -247,15 +253,17 @@ public class DefaultAgent implements Agent {
         return serverMetaDataHolder;
     }
 
-    protected PinpointSocketFactory createPinpointSocketFactory(boolean isSupportServerMode) {
-
+    protected PinpointSocketFactory createPinpointSocketFactory(CommandDispatcher commandDispatcher) {
     	PinpointSocketFactory pinpointSocketFactory = new PinpointSocketFactory();
         pinpointSocketFactory.setTimeoutMillis(1000 * 5);
 
         Map<String, Object> properties = this.agentInformation.toMap();
+        
+        boolean isSupportServerMode = this.profilerConfig.isTcpDataSenderCommandAcceptEnable();
+        
         if (isSupportServerMode) {
-        	CommandDispatcher.Builder builder = new CommandDispatcher.Builder();
-        	pinpointSocketFactory.setMessageListener(builder.build());
+        	pinpointSocketFactory.setMessageListener(commandDispatcher);
+        	pinpointSocketFactory.setServerStreamChannelMessageListener(commandDispatcher);
 
         	properties.put(AgentHandshakePropertyType.SUPPORT_SERVER.getName(), true);
         } else {

profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/CommandDispatcher.java

@@ -1,8 +1,5 @@
 package com.nhn.pinpoint.profiler.receiver;
 
-import java.util.ArrayList;
-import java.util.List;
-
 import org.apache.thrift.TBase;
 import org.apache.thrift.protocol.TCompactProtocol;
 import org.apache.thrift.protocol.TProtocolFactory;
@@ -11,12 +8,15 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.nhn.pinpoint.common.Version;
-import com.nhn.pinpoint.profiler.receiver.service.EchoService;
-import com.nhn.pinpoint.profiler.receiver.service.ThreadDumpService;
 import com.nhn.pinpoint.rpc.client.MessageListener;
 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.packet.stream.StreamClosePacket;
+import com.nhn.pinpoint.rpc.packet.stream.StreamCreateFailPacket;
+import com.nhn.pinpoint.rpc.packet.stream.StreamCreatePacket;
+import com.nhn.pinpoint.rpc.stream.ServerStreamChannelContext;
+import com.nhn.pinpoint.rpc.stream.ServerStreamChannelMessageListener;
 import com.nhn.pinpoint.rpc.util.AssertUtils;
 import com.nhn.pinpoint.thrift.dto.TResult;
 import com.nhn.pinpoint.thrift.io.DeserializerFactory;
@@ -25,121 +25,108 @@ import com.nhn.pinpoint.thrift.io.HeaderTBaseDeserializerFactory;
 import com.nhn.pinpoint.thrift.io.HeaderTBaseSerializer;
 import com.nhn.pinpoint.thrift.io.HeaderTBaseSerializerFactory;
 import com.nhn.pinpoint.thrift.io.SerializerFactory;
-import com.nhn.pinpoint.thrift.io.TBaseLocator;
 import com.nhn.pinpoint.thrift.io.TCommandRegistry;
 import com.nhn.pinpoint.thrift.io.TCommandTypeVersion;
 import com.nhn.pinpoint.thrift.io.ThreadLocalHeaderTBaseDeserializerFactory;
 import com.nhn.pinpoint.thrift.io.ThreadLocalHeaderTBaseSerializerFactory;
 import com.nhn.pinpoint.thrift.util.SerializationUtils;
 
-public class CommandDispatcher implements MessageListener  {
+public class CommandDispatcher implements MessageListener, ServerStreamChannelMessageListener  {
 
-	private final Logger logger = LoggerFactory.getLogger(this.getClass());
+    private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
-	private final ProfilerCommandServiceLocator locator;
+    private final ProfilerCommandServiceRegistry commandServiceRegistry = new ProfilerCommandServiceRegistry();
+    
+    private final SerializerFactory<HeaderTBaseSerializer> serializerFactory;
+    private final DeserializerFactory<HeaderTBaseDeserializer> deserializerFactory;
 
-	private final SerializerFactory<HeaderTBaseSerializer> serializerFactory;
-	private final DeserializerFactory<HeaderTBaseDeserializer> deserializerFactory;
+    public CommandDispatcher() {
+        this(Version.VERSION);
+    }
 
-	public CommandDispatcher(Builder builder) {
-		ProfilerCommandServiceRegistry registry = new ProfilerCommandServiceRegistry();
-		for (ProfilerCommandService service : builder.serviceList) {
-			registry.addService(service);
-		}
-		this.locator = registry;
-		
-		SerializerFactory<HeaderTBaseSerializer> serializerFactory = new HeaderTBaseSerializerFactory(true, builder.serializationMaxSize, builder.protocolFactory, builder.commandTbaseLocator);
-		this.serializerFactory = wrappedThreadLocalSerializerFactory(serializerFactory);
-		AssertUtils.assertNotNull(this.serializerFactory);
-		
-		DeserializerFactory<HeaderTBaseDeserializer> deserializerFactory = new HeaderTBaseDeserializerFactory(builder.protocolFactory, builder.commandTbaseLocator);
-		this.deserializerFactory = wrappedThreadLocalDeserializerFactory(deserializerFactory);
-		AssertUtils.assertNotNull(this.deserializerFactory);
-	}
+    public CommandDispatcher(String pinpointVersion) {
+        this(pinpointVersion, HeaderTBaseSerializerFactory.DEFAULT_UDP_STREAM_MAX_SIZE);
+    }
 
-	private SerializerFactory<HeaderTBaseSerializer> wrappedThreadLocalSerializerFactory(SerializerFactory<HeaderTBaseSerializer> serializerFactory) {
-		return new ThreadLocalHeaderTBaseSerializerFactory<HeaderTBaseSerializer>(serializerFactory);
-	}
-	
-	private DeserializerFactory<HeaderTBaseDeserializer> wrappedThreadLocalDeserializerFactory(DeserializerFactory<HeaderTBaseDeserializer> deserializerFactory) {
-		return new ThreadLocalHeaderTBaseDeserializerFactory<HeaderTBaseDeserializer>(deserializerFactory);
-	}
-	
-	@Override
-	public void handleSend(SendPacket sendPacket, Channel channel) {
-		logger.info("MessageReceive {} {}", sendPacket, channel);
-	}
+    public CommandDispatcher(String pinpointVersion, int serializationMaxSize) {
+        TProtocolFactory protocolFactory = new TCompactProtocol.Factory();
+        TCommandRegistry commandTbaseRegistry = new TCommandRegistry(TCommandTypeVersion.getVersion(pinpointVersion));
+        
+        SerializerFactory<HeaderTBaseSerializer> serializerFactory = new HeaderTBaseSerializerFactory(true, serializationMaxSize, protocolFactory, commandTbaseRegistry);
+        this.serializerFactory = wrappedThreadLocalSerializerFactory(serializerFactory);
+        AssertUtils.assertNotNull(this.serializerFactory);
 
-	
-	@Override
-	public void handleRequest(RequestPacket requestPacket, Channel channel) {
-		logger.info("MessageReceive {} {}", requestPacket, channel);
+        DeserializerFactory<HeaderTBaseDeserializer> deserializerFactory = new HeaderTBaseDeserializerFactory(protocolFactory, commandTbaseRegistry);
+        this.deserializerFactory = wrappedThreadLocalDeserializerFactory(deserializerFactory);
+        AssertUtils.assertNotNull(this.deserializerFactory);
+    }
 
-		TBase<?, ?> request = SerializationUtils.deserialize(requestPacket.getPayload(), deserializerFactory, null);
-		
-		TBase response = null;
-		if (request == null) {
-			TResult tResult = new TResult(false);
-			tResult.setMessage("Unsupported Type.");
-			
-			response = tResult;
-		} else {
-			ProfilerRequestCommandService service = locator.getRequestService(request);
-			
-			if (service == null) {
-				TResult tResult = new TResult(false);
-				tResult.setMessage("Unsupported Listener.");
+    @Override
+    public void handleSend(SendPacket sendPacket, Channel channel) {
+        logger.info("MessageReceive {} {}", sendPacket, channel);
+    }
 
-				response = tResult;
-			} else {
-				response = service.requestCommandService(request);
-			}
-		}
-		
-		byte[] payload = SerializationUtils.serialize(response, serializerFactory, null);
-		
-		if (payload != null) {
-			channel.write(new ResponsePacket(requestPacket.getRequestId(), payload));
-		}		
-	}
+    @Override
+    public void handleRequest(RequestPacket requestPacket, Channel channel) {
+        logger.info("MessageReceive {} {}", requestPacket, channel);
 
-	public static class Builder {
-		private List<ProfilerCommandService> serviceList = new ArrayList<ProfilerCommandService>();
+        TBase<?, ?> request = SerializationUtils.deserialize(requestPacket.getPayload(), deserializerFactory, null);
+        
+        TBase response = null;
+        if (request == null) {
+            TResult tResult = new TResult(false);
+            tResult.setMessage("Unsupported Type.");
+            
+            response = tResult;
+        } else {
+            ProfilerRequestCommandService service = commandServiceRegistry.getRequestService(request);
+            
+            if (service == null) {
+                TResult tResult = new TResult(false);
+                tResult.setMessage("Unsupported Listener.");
 
-		private int serializationMaxSize = HeaderTBaseSerializerFactory.DEFAULT_UDP_STREAM_MAX_SIZE;
-		private TProtocolFactory protocolFactory = new TCompactProtocol.Factory();
-		private TBaseLocator commandTbaseLocator = new TCommandRegistry(TCommandTypeVersion.getVersion(Version.VERSION));
-		
-		public Builder() {
-			serviceList.add(new ThreadDumpService());
-			serviceList.add(new EchoService());
-		}
-		
-		public void addService(ProfilerCommandService service) {
-			serviceList.add(service);
-		}
+                response = tResult;
+            } else {
+                response = service.requestCommandService(request);
+            }
+        }
+        
+        byte[] payload = SerializationUtils.serialize(response, serializerFactory, null);
+        if (payload != null) {
+            channel.write(new ResponsePacket(requestPacket.getRequestId(), payload));
+        }       
+    }
 
-		public void setProtocolFactory(TProtocolFactory protocolFactory) {
-			this.protocolFactory = protocolFactory;
-		}
+    @Override
+    public short handleStreamCreate(ServerStreamChannelContext streamChannelContext, StreamCreatePacket packet) {
+        logger.info("MessageReceived handleStreamCreate {} {}", packet, streamChannelContext);
 
-		public void setCommandTbaseLocator(TBaseLocator commandTbaseLocator) {
-			this.commandTbaseLocator = commandTbaseLocator;
-		}
+        TBase<?, ?> request = SerializationUtils.deserialize(packet.getPayload(), deserializerFactory, null);
+        
+        ProfilerStreamCommandService service = commandServiceRegistry.getStreamService(request);
+        if (service == null) {
+            return StreamCreateFailPacket.PACKET_UNSUPPORT;
+        }
+        
+        service.streamCommandService(request, streamChannelContext);
+        
+        return StreamCreatePacket.SUCCESS;
+    }
 
-		public void setSerializationMaxSize(int serializationMaxSize) {
-			this.serializationMaxSize = serializationMaxSize;
-		}
+    @Override
+    public void handleStreamClose(ServerStreamChannelContext streamChannelContext, StreamClosePacket packet) {
+    }
 
-		public CommandDispatcher build() {
-			AssertUtils.assertNotNull(protocolFactory, "protocolFactory may note be null.");
-			AssertUtils.assertNotNull(commandTbaseLocator, "commandTbaseLocator may note be null.");
-			AssertUtils.assertTrue(serializationMaxSize > 0, "serializationMaxSize must grater than zero.");
-			AssertUtils.assertTrue(serviceList.size() > 0, "serializationMaxSize must grater than zero.");
-			
-			return new CommandDispatcher(this);
-		}
-		
-	}
+    public boolean registerCommandService(ProfilerCommandService commandService) {
+        return this.commandServiceRegistry.addService(commandService);
+    }
+
+    private SerializerFactory<HeaderTBaseSerializer> wrappedThreadLocalSerializerFactory(SerializerFactory<HeaderTBaseSerializer> serializerFactory) {
+        return new ThreadLocalHeaderTBaseSerializerFactory<HeaderTBaseSerializer>(serializerFactory);
+    }
+
+    private DeserializerFactory<HeaderTBaseDeserializer> wrappedThreadLocalDeserializerFactory(DeserializerFactory<HeaderTBaseDeserializer> deserializerFactory) {
+        return new ThreadLocalHeaderTBaseDeserializerFactory<HeaderTBaseDeserializer>(deserializerFactory);
+    }
 	
 }

profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/ProfilerCommandServiceLocator.java

@@ -12,5 +12,7 @@ public interface ProfilerCommandServiceLocator {
 	ProfilerSimpleCommandService getSimpleService(TBase tBase);
 
 	ProfilerRequestCommandService getRequestService(TBase tBase);
+	
+	ProfilerStreamCommandService getStreamService(TBase tBase);
 
 }

profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/ProfilerCommandServiceRegistry.java

@@ -1,7 +1,6 @@
 package com.nhn.pinpoint.profiler.receiver;
 
-import java.util.HashMap;
-import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 import org.apache.thrift.TBase;
 import org.slf4j.Logger;
@@ -14,27 +13,24 @@ public class ProfilerCommandServiceRegistry implements ProfilerCommandServiceLoc
 	
 	private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
-	private final Map<Class<? extends TBase>, ProfilerCommandService> profilerCommandServiceRepository;
+	private final ConcurrentHashMap<Class<? extends TBase>, ProfilerCommandService> profilerCommandServiceRepository;
 
 	public ProfilerCommandServiceRegistry() {
-		profilerCommandServiceRepository = new HashMap<Class<? extends TBase>, ProfilerCommandService>();
+		profilerCommandServiceRepository = new ConcurrentHashMap<Class<? extends TBase>, ProfilerCommandService>();
 	}
-
-	/**
-	 * not guarantee thread safe.
-	 */
 	
 	public boolean addService(ProfilerCommandService service) {
 		return addService(service.getCommandClazz(), service);
 	}
 	
 	public boolean addService(Class<? extends TBase> clazz, ProfilerCommandService service) {
-		if (profilerCommandServiceRepository.containsKey(clazz)) {
-			logger.warn("Already Register Type({}).", clazz.getName());
-			return false;
-		}
-
-		profilerCommandServiceRepository.put(clazz, service);
+	    ProfilerCommandService inValue = profilerCommandServiceRepository.putIfAbsent(clazz, service);
+	    
+	    if (inValue != null) {
+            logger.warn("Already Register Type({}).", clazz.getName());
+            return false;
+	    }
+	    
 		return true;
 	}
 
@@ -64,5 +60,16 @@ public class ProfilerCommandServiceRegistry implements ProfilerCommandServiceLoc
 		
 		return null;
 	}
+	
+	@Override
+	public ProfilerStreamCommandService getStreamService(TBase tBase) {
+        ProfilerCommandService service = profilerCommandServiceRepository.get(tBase.getClass());
+        
+        if (service instanceof ProfilerStreamCommandService) {
+            return (ProfilerStreamCommandService) service;
+        }
+        
+        return null;
+	}
 
 }

profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/ProfilerStreamCommandService.java

@@ -0,0 +1,11 @@
+package com.nhn.pinpoint.profiler.receiver;
+
+import org.apache.thrift.TBase;
+
+import com.nhn.pinpoint.rpc.stream.ServerStreamChannelContext;
+
+public interface ProfilerStreamCommandService extends ProfilerCommandService {
+
+    short streamCommandService(TBase tBase, ServerStreamChannelContext streamChannelContext);
+    
+}

thrift/src/main/java/com/navercorp/pinpoint/thrift/io/TCommandRegistry.java

@@ -1,9 +1,8 @@
 package com.nhn.pinpoint.thrift.io;
 
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 import org.apache.thrift.TBase;
 import org.apache.thrift.TException;
@@ -13,15 +12,13 @@ import org.apache.thrift.TException;
  */
 public class TCommandRegistry implements TBaseLocator {
 
-	private final Map<Short, TCommandType> commandTBaseRepository;
+	private final ConcurrentHashMap<Short, TCommandType> commandTBaseRepository = new ConcurrentHashMap<Short, TCommandType>();
 
 	public TCommandRegistry(TCommandTypeVersion version) {
 		this(version.getSupportCommandList());
 	}
 	
 	public TCommandRegistry(List<TCommandType> supportCommandList) {
-		commandTBaseRepository = new HashMap<Short, TCommandType>(supportCommandList.size());
-
 		for (TCommandType type : supportCommandList) {
 			commandTBaseRepository.put(type.getType(), type);
 		}
@@ -30,7 +27,6 @@ public class TCommandRegistry implements TBaseLocator {
 	@Override
 	public TBase<?, ?> tBaseLookup(short type) throws TException {
 		TCommandType commandTBaseType = commandTBaseRepository.get(type);
-
 		if (commandTBaseType == null) {
 	        throw new TException("Unsupported type:" + type);
 		}