Parse and listener

apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/data/AbstractHashMessage.java

@@ -36,4 +36,8 @@ public abstract class AbstractHashMessage {
     public int getHashCode() {
         return hashCode;
     }
+
+    public void setKey(String key) {
+        this.hashCode = key.hashCode();
+    }
 }

apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/data/Data.java

@@ -129,6 +129,11 @@ public abstract class Data extends EndOfBatchQueueMessage {
         return dataStrings[0];
     }
 
+    public void setId(String id) {
+        setKey(id);
+        this.dataStrings[0] = id;
+    }
+
     public void mergeData(Data newData) {
         for (int i = 0; i < stringColumns.length; i++) {
             String stringData = stringColumns[i].getOperation().operate(newData.getDataString(i), this.dataStrings[i]);

apm-collector/apm-collector-queue/collector-queue-define/src/main/java/org/skywalking/apm/collector/queue/base/QueueEventHandler.java

@@ -18,11 +18,9 @@
 
 package org.skywalking.apm.collector.queue.base;
 
-import org.skywalking.apm.collector.core.data.Data;
-
 /**
  * @author peng-yongsheng
  */
-public interface QueueEventHandler<MESSAGE extends Data> {
+public interface QueueEventHandler<MESSAGE> {
     void tell(MESSAGE message);
 }

apm-collector/apm-collector-queue/collector-queue-define/src/main/java/org/skywalking/apm/collector/queue/base/QueueExecutor.java

@@ -18,11 +18,10 @@
 
 package org.skywalking.apm.collector.queue.base;
 
-import org.skywalking.apm.collector.core.data.Data;
 import org.skywalking.apm.collector.core.framework.Executor;
 
 /**
  * @author peng-yongsheng
  */
-public interface QueueExecutor<MESSAGE extends Data> extends Executor<MESSAGE> {
+public interface QueueExecutor<MESSAGE> extends Executor<MESSAGE> {
 }

apm-collector/apm-collector-queue/collector-queue-define/src/main/java/org/skywalking/apm/collector/queue/service/QueueCreatorService.java

@@ -18,7 +18,6 @@
 
 package org.skywalking.apm.collector.queue.service;
 
-import org.skywalking.apm.collector.core.data.Data;
 import org.skywalking.apm.collector.core.module.Service;
 import org.skywalking.apm.collector.queue.base.QueueEventHandler;
 import org.skywalking.apm.collector.queue.base.QueueExecutor;
@@ -26,6 +25,6 @@ import org.skywalking.apm.collector.queue.base.QueueExecutor;
 /**
  * @author peng-yongsheng
  */
-public interface QueueCreatorService<MESSAGE extends Data> extends Service {
+public interface QueueCreatorService<MESSAGE> extends Service {
     QueueEventHandler<MESSAGE> create(int queueSize, QueueExecutor<MESSAGE> executor);
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/table/global/GlobalTrace.java

@@ -52,11 +52,23 @@ public class GlobalTrace extends Data {
         return getDataString(1);
     }
 
+    public void setSegmentId(String segmentId) {
+        setDataString(1, segmentId);
+    }
+
     public String getGlobalTraceId() {
         return getDataString(2);
     }
 
+    public void setGlobalTraceId(String globalTraceId) {
+        setDataString(2, globalTraceId);
+    }
+
     public Long getTimeBucket() {
         return getDataLong(0);
     }
+
+    public void setTimeBucket(long timeBucket) {
+        setDataLong(0, timeBucket);
+    }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/table/segment/Segment.java

@@ -50,4 +50,8 @@ public class Segment extends Data {
     public byte[] getDataBinary() {
         return getDataBytes(0);
     }
+
+    public void setDataBinary(byte[] dataBinary) {
+        setDataBytes(0, dataBinary);
+    }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/table/segment/SegmentCost.java

@@ -58,31 +58,63 @@ public class SegmentCost extends Data {
         return getDataString(1);
     }
 
+    public void setSegmentId(String segmentId) {
+        setDataString(1, segmentId);
+    }
+
     public String getServiceName() {
         return getDataString(2);
     }
 
+    public void setServiceName(String serviceName) {
+        setDataString(2, serviceName);
+    }
+
     public Long getCost() {
         return getDataLong(0);
     }
 
+    public void setCost(Long cost) {
+        setDataLong(0, cost);
+    }
+
     public Long getStartTime() {
         return getDataLong(1);
     }
 
+    public void setStartTime(Long startTime) {
+        setDataLong(1, startTime);
+    }
+
     public Long getEndTime() {
         return getDataLong(2);
     }
 
+    public void setEndTime(Long endTime) {
+        setDataLong(2, endTime);
+    }
+
     public Long getTimeBucket() {
         return getDataLong(3);
     }
 
+    public void setTimeBucket(Long timeBucket) {
+        setDataLong(3, timeBucket);
+    }
+
     public Integer getApplicationId() {
         return getDataInteger(0);
     }
 
+    public void setApplicationId(Integer applicationId) {
+        setDataInteger(0, applicationId);
+    }
+
     public Boolean getIsError() {
         return getDataBoolean(0);
     }
+
+    public void setIsError(Boolean isError) {
+        setDataBoolean(0, isError);
+    }
 }

apm-collector/apm-collector-stream/collector-stream-provider/src/main/java/org/skywalking/apm/collector/stream/worker/base/AbstractLocalAsyncWorker.java

@@ -19,7 +19,6 @@
 package org.skywalking.apm.collector.stream.worker.base;
 
 import org.skywalking.apm.collector.cache.CacheServiceManager;
-import org.skywalking.apm.collector.core.data.Data;
 import org.skywalking.apm.collector.queue.base.QueueExecutor;
 import org.skywalking.apm.collector.storage.service.DAOService;
 
@@ -30,7 +29,7 @@ import org.skywalking.apm.collector.storage.service.DAOService;
  * @author peng-yongsheng
  * @since v3.0-2017
  */
-public abstract class AbstractLocalAsyncWorker<INPUT extends Data, OUTPUT extends Data> extends AbstractWorker<INPUT, OUTPUT> implements QueueExecutor<INPUT> {
+public abstract class AbstractLocalAsyncWorker<INPUT, OUTPUT> extends AbstractWorker<INPUT, OUTPUT> implements QueueExecutor<INPUT> {
 
     public AbstractLocalAsyncWorker(DAOService daoService, CacheServiceManager cacheServiceManager) {
         super(daoService, cacheServiceManager);

apm-collector/apm-collector-stream/collector-stream-provider/src/main/java/org/skywalking/apm/collector/stream/worker/base/AbstractLocalAsyncWorkerProvider.java

@@ -19,7 +19,6 @@
 package org.skywalking.apm.collector.stream.worker.base;
 
 import org.skywalking.apm.collector.cache.CacheServiceManager;
-import org.skywalking.apm.collector.core.data.Data;
 import org.skywalking.apm.collector.queue.base.QueueEventHandler;
 import org.skywalking.apm.collector.queue.base.QueueExecutor;
 import org.skywalking.apm.collector.queue.service.QueueCreatorService;
@@ -28,7 +27,7 @@ import org.skywalking.apm.collector.storage.service.DAOService;
 /**
  * @author peng-yongsheng
  */
-public abstract class AbstractLocalAsyncWorkerProvider<INPUT extends Data, OUTPUT extends Data, WORKER_TYPE extends AbstractLocalAsyncWorker<INPUT, OUTPUT> & QueueExecutor<INPUT>> extends AbstractWorkerProvider<INPUT, OUTPUT, WORKER_TYPE> {
+public abstract class AbstractLocalAsyncWorkerProvider<INPUT, OUTPUT, WORKER_TYPE extends AbstractLocalAsyncWorker<INPUT, OUTPUT> & QueueExecutor<INPUT>> extends AbstractWorkerProvider<INPUT, OUTPUT, WORKER_TYPE> {
 
     public abstract int queueSize();
 
@@ -41,12 +40,8 @@ public abstract class AbstractLocalAsyncWorkerProvider<INPUT extends Data, OUTPU
     }
 
     @Override
-    final public WorkerRef create(WorkerCreateListener workerCreateListener) throws ProviderNotFoundException {
-<<<<<<< HEAD
-        WorkerType localAsyncWorker = workerInstance(getDaoService(), getCacheServiceManager());
-=======
-        WORKER_TYPE localAsyncWorker = workerInstance(daoService);
->>>>>>> 0c17906c3c1c41752e1ec38b37d9e0dec22503ca
+    public final WorkerRef create(WorkerCreateListener workerCreateListener) throws ProviderNotFoundException {
+        WORKER_TYPE localAsyncWorker = workerInstance(getDaoService(), getCacheServiceManager());
         workerCreateListener.addWorker(localAsyncWorker);
         QueueEventHandler<INPUT> queueEventHandler = queueCreatorService.create(queueSize(), localAsyncWorker);
         return new LocalAsyncWorkerRef<>(localAsyncWorker, queueEventHandler);

apm-collector/apm-collector-stream/collector-stream-provider/src/main/java/org/skywalking/apm/collector/stream/worker/base/AbstractRemoteWorkerProvider.java

@@ -49,11 +49,7 @@ public abstract class AbstractRemoteWorkerProvider<INPUT extends Data, OUTPUT ex
      * worker instance, when the worker provider not find then Throw this Exception.
      */
     @Override final public WorkerRef create(WorkerCreateListener workerCreateListener) {
-<<<<<<< HEAD
-        WorkerType remoteWorker = workerInstance(getDaoService(), getCacheServiceManager());
-=======
-        WORKER_TYPE remoteWorker = workerInstance(daoService);
->>>>>>> 0c17906c3c1c41752e1ec38b37d9e0dec22503ca
+        WORKER_TYPE remoteWorker = workerInstance(getDaoService(), getCacheServiceManager());
         workerCreateListener.addWorker(remoteWorker);
         RemoteWorkerRef<INPUT, OUTPUT> workerRef = new RemoteWorkerRef<>(remoteWorker);
         return workerRef;

apm-collector/apm-collector-stream/collector-stream-provider/src/main/java/org/skywalking/apm/collector/stream/worker/base/AbstractWorker.java

@@ -19,7 +19,6 @@
 package org.skywalking.apm.collector.stream.worker.base;
 
 import org.skywalking.apm.collector.cache.CacheServiceManager;
-import org.skywalking.apm.collector.core.data.Data;
 import org.skywalking.apm.collector.core.graph.Next;
 import org.skywalking.apm.collector.core.graph.NodeProcessor;
 import org.skywalking.apm.collector.storage.service.DAOService;
@@ -29,7 +28,7 @@ import org.slf4j.LoggerFactory;
 /**
  * @author peng-yongsheng
  */
-public abstract class AbstractWorker<INPUT extends Data, OUTPUT extends Data> implements NodeProcessor<INPUT, OUTPUT> {
+public abstract class AbstractWorker<INPUT, OUTPUT> implements NodeProcessor<INPUT, OUTPUT> {
 
     private final Logger logger = LoggerFactory.getLogger(AbstractWorker.class);
 

apm-collector/apm-collector-stream/collector-stream-provider/src/main/java/org/skywalking/apm/collector/stream/worker/base/AbstractWorkerProvider.java

@@ -19,14 +19,12 @@
 package org.skywalking.apm.collector.stream.worker.base;
 
 import org.skywalking.apm.collector.cache.CacheServiceManager;
-import org.skywalking.apm.collector.core.data.Data;
 import org.skywalking.apm.collector.storage.service.DAOService;
 
 /**
  * @author peng-yongsheng
  */
-<<<<<<< HEAD
-public abstract class AbstractWorkerProvider<INPUT extends Data, OUTPUT extends Data, WorkerType extends AbstractWorker<INPUT, OUTPUT>> implements Provider {
+public abstract class AbstractWorkerProvider<INPUT, OUTPUT, WORKER_TYPE extends AbstractWorker<INPUT, OUTPUT>> implements Provider {
 
     private final DAOService daoService;
     private final CacheServiceManager cacheServiceManager;
@@ -44,9 +42,5 @@ public abstract class AbstractWorkerProvider<INPUT extends Data, OUTPUT extends
         return cacheServiceManager;
     }
 
-    public abstract WorkerType workerInstance(DAOService daoService, CacheServiceManager cacheServiceManager);
-=======
-public abstract class AbstractWorkerProvider<INPUT extends Data, OUTPUT extends Data, WORKER_TYPE extends AbstractWorker<INPUT, OUTPUT>> implements Provider {
-    public abstract WORKER_TYPE workerInstance(DAOService daoService);
->>>>>>> 0c17906c3c1c41752e1ec38b37d9e0dec22503ca
+    public abstract WORKER_TYPE workerInstance(DAOService daoService, CacheServiceManager cacheServiceManager);
 }

apm-collector/apm-collector-stream/collector-stream-provider/src/main/java/org/skywalking/apm/collector/stream/worker/base/LocalAsyncWorkerRef.java

@@ -18,19 +18,17 @@
 
 package org.skywalking.apm.collector.stream.worker.base;
 
-import org.skywalking.apm.collector.core.data.Data;
 import org.skywalking.apm.collector.core.graph.NodeProcessor;
 import org.skywalking.apm.collector.queue.base.QueueEventHandler;
 
 /**
  * @author peng-yongsheng
  */
-public class LocalAsyncWorkerRef<INPUT extends Data, OUTPUT extends Data> extends WorkerRef<INPUT, OUTPUT> {
+public class LocalAsyncWorkerRef<INPUT, OUTPUT> extends WorkerRef<INPUT, OUTPUT> {
 
     private final QueueEventHandler<INPUT> queueEventHandler;
 
-    LocalAsyncWorkerRef(NodeProcessor<INPUT, OUTPUT> destinationHandler,
-        QueueEventHandler<INPUT> queueEventHandler) {
+    LocalAsyncWorkerRef(NodeProcessor<INPUT, OUTPUT> destinationHandler, QueueEventHandler<INPUT> queueEventHandler) {
         super(destinationHandler);
         this.queueEventHandler = queueEventHandler;
     }

apm-collector/apm-collector-stream/collector-stream-provider/src/main/java/org/skywalking/apm/collector/stream/worker/base/WorkerRef.java

@@ -18,14 +18,13 @@
 
 package org.skywalking.apm.collector.stream.worker.base;
 
-import org.skywalking.apm.collector.core.data.Data;
 import org.skywalking.apm.collector.core.graph.NodeProcessor;
 import org.skywalking.apm.collector.core.graph.WayToNode;
 
 /**
  * @author peng-yongsheng
  */
-public abstract class WorkerRef<INPUT extends Data, OUTPUT extends Data> extends WayToNode<INPUT, OUTPUT> {
+public abstract class WorkerRef<INPUT, OUTPUT> extends WayToNode<INPUT, OUTPUT> {
     WorkerRef(NodeProcessor<INPUT, OUTPUT> destinationHandler) {
         super(destinationHandler);
     }