#28 change class

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/ClusterPointRouter.java

@@ -42,7 +42,7 @@ public class ClusterPointRouter {
 	private final WebClusterPoint webClusterPoint;
 
 	@Autowired
-	private SerializerFactory commandSerializerFactory;
+	private SerializerFactory<HeaderTBaseSerializer> commandSerializerFactory;
 
 	@Autowired
 	private DeserializerFactory<HeaderTBaseDeserializer> commandDeserializerFactory;

collector/src/main/java/com/navercorp/pinpoint/collector/receiver/tcp/TCPReceiver.java

@@ -65,7 +65,7 @@ public class TCPReceiver {
 
     private final ThreadPoolExecutor worker = ExecutorFactory.newFixedThreadPool(threadSize, workerQueueSize, THREAD_FACTORY);
 
-    private final SerializerFactory serializerFactory = new ThreadLocalHeaderTBaseSerializerFactory(new HeaderTBaseSerializerFactory(true, HeaderTBaseSerializerFactory.DEFAULT_UDP_STREAM_MAX_SIZE));
+    private final SerializerFactory<HeaderTBaseSerializer> serializerFactory = new ThreadLocalHeaderTBaseSerializerFactory<HeaderTBaseSerializer>(new HeaderTBaseSerializerFactory(true, HeaderTBaseSerializerFactory.DEFAULT_UDP_STREAM_MAX_SIZE));
 
     private final DeserializerFactory<HeaderTBaseDeserializer> deserializerFactory = new ThreadLocalHeaderTBaseDeserializerFactory<HeaderTBaseDeserializer>(new HeaderTBaseDeserializerFactory());
 

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

@@ -38,7 +38,7 @@ public class CommandDispatcher implements MessageListener  {
 
 	private final ProfilerCommandServiceLocator locator;
 
-	private final SerializerFactory serializerFactory;
+	private final SerializerFactory<HeaderTBaseSerializer> serializerFactory;
 	private final DeserializerFactory<HeaderTBaseDeserializer> deserializerFactory;
 
 	public CommandDispatcher(Builder builder) {
@@ -48,7 +48,7 @@ public class CommandDispatcher implements MessageListener  {
 		}
 		this.locator = registry;
 		
-		SerializerFactory serializerFactory = new HeaderTBaseSerializerFactory(true, builder.serializationMaxSize, builder.protocolFactory, builder.commandTbaseLocator);
+		SerializerFactory<HeaderTBaseSerializer> serializerFactory = new HeaderTBaseSerializerFactory(true, builder.serializationMaxSize, builder.protocolFactory, builder.commandTbaseLocator);
 		this.serializerFactory = wrappedThreadLocalSerializerFactory(serializerFactory);
 		AssertUtils.assertNotNull(this.serializerFactory);
 		
@@ -57,8 +57,8 @@ public class CommandDispatcher implements MessageListener  {
 		AssertUtils.assertNotNull(this.deserializerFactory);
 	}
 
-	private SerializerFactory wrappedThreadLocalSerializerFactory(SerializerFactory serializerFactory) {
-		return new ThreadLocalHeaderTBaseSerializerFactory(serializerFactory);
+	private SerializerFactory<HeaderTBaseSerializer> wrappedThreadLocalSerializerFactory(SerializerFactory<HeaderTBaseSerializer> serializerFactory) {
+		return new ThreadLocalHeaderTBaseSerializerFactory<HeaderTBaseSerializer>(serializerFactory);
 	}
 	
 	private DeserializerFactory<HeaderTBaseDeserializer> wrappedThreadLocalDeserializerFactory(DeserializerFactory<HeaderTBaseDeserializer> deserializerFactory) {

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

@@ -8,7 +8,7 @@ import org.apache.thrift.transport.TTransportException;
 
 /**
  * ByteArrayOutputStreamTransport
- * - unsupported read operation
+ * - write only
  * 
  * @author jaehong.kim
  */

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

@@ -12,6 +12,12 @@ import com.nhn.pinpoint.thrift.dto.TSpan;
 import com.nhn.pinpoint.thrift.dto.TSpanChunk;
 import com.nhn.pinpoint.thrift.dto.TSpanEvent;
 
+/**
+ * ChunkHeaderBufferedTBaseSerializer
+ * - need flush handler
+ * 
+ * @author jaehong.kim
+ */
 public class ChunkHeaderBufferedTBaseSerializer {
     private static final String FIELD_NAME_SPAN_EVENT_LIST = "spanEventList";
 
@@ -47,6 +53,7 @@ public class ChunkHeaderBufferedTBaseSerializer {
         }
     }
 
+    // TSpanChunk = TSpanChunk + TSpanChunk
     private void addTSpanChunk(TBase<?, ?> base) throws TException {
         final TSpanChunk chunk = (TSpanChunk) base;
         if (chunk.getSpanEventList() == null) {
@@ -67,6 +74,7 @@ public class ChunkHeaderBufferedTBaseSerializer {
         }
     }
 
+    // TSpan = TSpan + TSpanChunk
     private void addTSpan(TBase<?, ?> base) throws TException {
         final TSpan span = (TSpan) base;
         if (span.getSpanEventList() == null) {
@@ -88,12 +96,13 @@ public class ChunkHeaderBufferedTBaseSerializer {
         }
     }
 
+    // write chunk header + header + body
     private void write(final TBase<?, ?> base, final String fieldName, final List<TBaseStreamNode> list) throws TException {
         final ReplaceListCompactProtocol protocol = new ReplaceListCompactProtocol(new ByteArrayOutputStreamTransport(out));
 
         // write chunk header
         writeChunkHeader(protocol);
-        
+
         // write header
         writeHeader(protocol, locator.headerLookup(base));
         if (list != null && list.size() > 0) {
@@ -101,12 +110,13 @@ public class ChunkHeaderBufferedTBaseSerializer {
         }
 
         base.write(protocol);
-        
+
         if (isOverflow()) {
             flush();
         }
     }
 
+    // write chunk header + header + body
     private void write(final TBase<?, ?> base) throws TException {
         final TCompactProtocol protocol = new TCompactProtocol(new ByteArrayOutputStreamTransport(out));
 
@@ -117,7 +127,7 @@ public class ChunkHeaderBufferedTBaseSerializer {
         writeHeader(protocol, locator.headerLookup(base));
 
         base.write(protocol);
-        
+
         if (isOverflow()) {
             flush();
         }
@@ -140,7 +150,6 @@ public class ChunkHeaderBufferedTBaseSerializer {
     private void writeHeader(final TProtocol protocol, final Header header) throws TException {
         protocol.writeByte(header.getSignature());
         protocol.writeByte(header.getVersion());
-        // 프로토콜 변경에 관계 없이 고정 사이즈의 데이터로 인코딩 하도록 변경.
         short type = header.getType();
         protocol.writeByte(BytesUtils.writeShort1(type));
         protocol.writeByte(BytesUtils.writeShort2(type));
@@ -166,15 +175,13 @@ public class ChunkHeaderBufferedTBaseSerializer {
     }
 
     public String toString() {
-        
-        return toStringBinary(out.toByteArray(), 0, out.size());
-//        StringBuilder sb = new StringBuilder();
-//        sb.append("{");
-//        sb.append("bufferSize=").append(out.size()).append(", ");
-//        sb.append("flushSize=").append(flushSize);
-//        sb.append("}");
-//
-//        return sb.toString();
+        StringBuilder sb = new StringBuilder();
+        sb.append("{");
+        sb.append("bufferSize=").append(out.size()).append(", ");
+        sb.append("flushSize=").append(flushSize);
+        sb.append("}");
+
+        return sb.toString();
     }
 
     TSpanChunk toSpanChunk(TSpan span) {
@@ -199,20 +206,4 @@ public class ChunkHeaderBufferedTBaseSerializer {
 
         return spanChunk;
     }
-    
-    
-    
-    static String toStringBinary(final byte[] b, int off, int len) {
-        StringBuilder result = new StringBuilder();
-        for (int i = off; i < off + len; ++i) {
-            int ch = b[i] & 0xFF;
-            if ((ch >= '0' && ch <= '9') || (ch >= 'A' && ch <= 'Z') || (ch >= 'a' && ch <= 'z') || " `~!@#$%^&*()-_=+[]{}|;:'\",.<>/?".indexOf(ch) >= 0) {
-                result.append((char) ch);
-            } else {
-                result.append(String.format("\\x%02X", ch));
-            }
-        }
-        return result.toString();
-    }
-
 }

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

@@ -1,6 +1,10 @@
 package com.nhn.pinpoint.thrift.io;
 
+/**
+ * 
+ * @author jaehong.kim
+ */
 public interface ChunkHeaderBufferedTBaseSerializerFlushHandler {
 
     void handle(byte[] buffer, int offset, int length);
-}
+}

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

@@ -14,30 +14,18 @@ public class ChunkHeaderTBaseDeserializer {
     private final TMemoryInputTransport trans;
     private final TBaseLocator locator;
 
-    /**
-     * Create a new TDeserializer. It will use the TProtocol specified by the factory that is passed in.
-     *
-     * @param protocolFactory
-     *            Factory to create a protocol
-     */
     ChunkHeaderTBaseDeserializer(TProtocolFactory protocolFactory, TBaseLocator locator) {
         this.trans = new TMemoryInputTransport();
         this.protocol = protocolFactory.getProtocol(trans);
         this.locator = locator;
     }
 
-    /**
-     * Deserialize the Thrift object from a byte array.
-     *
-     * @param bytes
-     *            The array to read from
-     */
     public List<TBase<?, ?>> deserialize(byte[] bytes, int offset, int length) throws TException {
         List<TBase<?, ?>> list = new ArrayList<TBase<?, ?>>();
         try {
             trans.reset(bytes, offset, length);
 
-            Header header = readHeader();
+            final Header header = readHeader();
             if (locator.isChunkHeader(header.getType())) {
 
                 TBase<?, ?> base = null;
@@ -92,7 +80,6 @@ public class ChunkHeaderTBaseDeserializer {
 
         final byte signature = protocol.readByte();
         final byte version = protocol.readByte();
-        // 프로토콜 변경에 관계 없이 고정 사이즈의 데이터로 인코딩 하도록 변경.
         final byte type1 = protocol.readByte();
         final byte type2 = protocol.readByte();
         final short type = bytesToShort(type1, type2);

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

@@ -18,6 +18,8 @@ import com.nhn.pinpoint.thrift.dto.TStringMetaData;
  * @author koo.taejin
  * @author netspider
  * @author hyungil.jeong
+ * @author jaehong.kim
+ *   - add CHUNK_HEADER
  */
 class DefaultTBaseLocator implements TBaseLocator {
 

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

@@ -8,7 +8,7 @@ import java.io.ByteArrayOutputStream;
 /**
  * @author koo.taejin
  */
-public final class HeaderTBaseSerializerFactory implements SerializerFactory {
+public final class HeaderTBaseSerializerFactory implements SerializerFactory<HeaderTBaseSerializer> {
 
     private static final TBaseLocator DEFAULT_TBASE_LOCATOR = new DefaultTBaseLocator();
 

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

@@ -3,6 +3,6 @@ package com.nhn.pinpoint.thrift.io;
 /**
  * @author emeroad
  */
-public interface SerializerFactory {
-    HeaderTBaseSerializer createSerializer();
+public interface SerializerFactory<E> {
+    E createSerializer();
 }

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

@@ -2,28 +2,29 @@ package com.nhn.pinpoint.thrift.io;
 
 /**
  * @author emeroad
+ * @author jaehong.kim
+ *   - change to generic type
  */
-public class ThreadLocalHeaderTBaseSerializerFactory implements SerializerFactory {
+public class ThreadLocalHeaderTBaseSerializerFactory<E> implements SerializerFactory<E> {
 
-    private final ThreadLocal<HeaderTBaseSerializer> cache = new ThreadLocal<HeaderTBaseSerializer>() {
+    private final ThreadLocal<E> cache = new ThreadLocal<E>() {
         @Override
-        protected HeaderTBaseSerializer initialValue() {
+        protected E initialValue() {
             return factory.createSerializer();
         }
     };
 
-    private final SerializerFactory factory;
+    private final SerializerFactory<E> factory;
 
-    public ThreadLocalHeaderTBaseSerializerFactory(SerializerFactory factory) {
+    public ThreadLocalHeaderTBaseSerializerFactory(SerializerFactory<E> factory) {
         if (factory == null) {
             throw new NullPointerException("factory must not be null");
         }
         this.factory = factory;
     }
 
-
     @Override
-    public HeaderTBaseSerializer createSerializer() {
+    public E createSerializer() {
         return cache.get();
     }
 }

thrift/src/test/java/com/navercorp/pinpoint/thrift/io/ChunkHeaderBufferedTBaseSerializerTest.java

@@ -3,7 +3,6 @@ package com.nhn.pinpoint.thrift.io;
 import static org.junit.Assert.*;
 
 import java.util.Arrays;
-import java.util.concurrent.TimeUnit;
 
 import org.apache.thrift.TException;
 import org.junit.Test;
@@ -16,7 +15,6 @@ public class ChunkHeaderBufferedTBaseSerializerTest {
 
     @Test
     public void add() throws TException {
-        System.out.println("add");
         ChunkHeaderBufferedTBaseSerializer serializer = new ChunkHeaderBufferedTBaseSerializer(1024);
         serializer.setFlushHandler(new ChunkHeaderBufferedTBaseSerializerFlushHandler() {
 
@@ -27,24 +25,28 @@ public class ChunkHeaderBufferedTBaseSerializerTest {
             }
         });
 
+        // add and flush
         flush = false;
         TSpanChunk chunk = new TSpanMockBuilder().buildChunk(1, 1024);
         serializer.add(chunk);
         System.out.println(serializer);
         assertTrue(flush);
 
+        // add and flush * 3
         flush = false;
         chunk = new TSpanMockBuilder().buildChunk(3, 1024);
         serializer.add(chunk);
         System.out.println(serializer);
         assertTrue(flush);
 
+        // add
         flush = false;
         chunk = new TSpanMockBuilder().buildChunk(3, 10);
         serializer.add(chunk);
         System.out.println(serializer);
         assertFalse(flush);
 
+        // flush
         serializer.flush();
         assertTrue(flush);
     }