From 3f30e6dca251eaab7ba45760f62f9fd8267cd450 Mon Sep 17 00:00:00 2001
From: peng-yongsheng <8082209@qq.com>
Date: Sun, 7 Jan 2018 11:25:33 +0800
Subject: [PATCH] =?UTF-8?q?Modify=20base=20worker=20modal=E2=80=99s=20gene?=
 =?UTF-8?q?ric=20type=20definition.?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../worker/model/base/AbstractLocalAsyncWorker.java    |  4 ++--
 .../model/base/AbstractLocalAsyncWorkerProvider.java   |  4 ++--
 .../worker/model/base/AbstractRemoteWorker.java        |  4 ++--
 .../model/base/AbstractRemoteWorkerProvider.java       |  4 ++--
 .../analysis/worker/model/base/AbstractWorker.java     |  5 ++---
 .../worker/model/base/AbstractWorkerProvider.java      |  5 ++---
 .../worker/model/base/LocalAsyncWorkerRef.java         | 10 ++++++----
 .../analysis/worker/model/base/RemoteWorkerRef.java    |  4 ++--
 .../worker/model/base/WorkerCreateListener.java        |  2 +-
 .../analysis/worker/model/base/WorkerRef.java          |  2 +-
 10 files changed, 22 insertions(+), 22 deletions(-)

diff --git a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractLocalAsyncWorker.java b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractLocalAsyncWorker.java
index 73a48df6c..d85ac9f5d 100644
--- a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractLocalAsyncWorker.java
+++ b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractLocalAsyncWorker.java
@@ -18,7 +18,7 @@
 
 package org.apache.skywalking.apm.collector.analysis.worker.model.base;
 
-import org.apache.skywalking.apm.collector.core.data.EndOfBatchQueueMessage;
+import org.apache.skywalking.apm.collector.core.data.QueueData;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 
 /**
@@ -28,7 +28,7 @@ import org.apache.skywalking.apm.collector.core.module.ModuleManager;
  * @author peng-yongsheng
  * @since v3.0-2017
  */
-public abstract class AbstractLocalAsyncWorker<INPUT extends EndOfBatchQueueMessage, OUTPUT extends EndOfBatchQueueMessage> extends AbstractWorker<INPUT, OUTPUT> {
+public abstract class AbstractLocalAsyncWorker<INPUT extends QueueData, OUTPUT extends QueueData> extends AbstractWorker<INPUT, OUTPUT> {
 
     public AbstractLocalAsyncWorker(ModuleManager moduleManager) {
         super(moduleManager);
diff --git a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractLocalAsyncWorkerProvider.java b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractLocalAsyncWorkerProvider.java
index 54d92ef92..472f4d173 100644
--- a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractLocalAsyncWorkerProvider.java
+++ b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractLocalAsyncWorkerProvider.java
@@ -18,14 +18,14 @@
 
 package org.apache.skywalking.apm.collector.analysis.worker.model.base;
 
-import org.apache.skywalking.apm.collector.core.data.EndOfBatchQueueMessage;
+import org.apache.skywalking.apm.collector.core.data.QueueData;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.commons.datacarrier.DataCarrier;
 
 /**
  * @author peng-yongsheng
  */
-public abstract class AbstractLocalAsyncWorkerProvider<INPUT extends EndOfBatchQueueMessage, OUTPUT extends EndOfBatchQueueMessage, WORKER_TYPE extends AbstractLocalAsyncWorker<INPUT, OUTPUT>> extends AbstractWorkerProvider<INPUT, OUTPUT, WORKER_TYPE> {
+public abstract class AbstractLocalAsyncWorkerProvider<INPUT extends QueueData, OUTPUT extends QueueData, WORKER_TYPE extends AbstractLocalAsyncWorker<INPUT, OUTPUT>> extends AbstractWorkerProvider<INPUT, OUTPUT, WORKER_TYPE> {
 
     public abstract int queueSize();
 
diff --git a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractRemoteWorker.java b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractRemoteWorker.java
index 6f13e83b4..42769c464 100644
--- a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractRemoteWorker.java
+++ b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractRemoteWorker.java
@@ -18,7 +18,7 @@
 
 package org.apache.skywalking.apm.collector.analysis.worker.model.base;
 
-import org.apache.skywalking.apm.collector.core.data.AbstractData;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.remote.service.Selector;
 
@@ -31,7 +31,7 @@ import org.apache.skywalking.apm.collector.remote.service.Selector;
  * @author peng-yongsheng
  * @since v3.0-2017
  */
-public abstract class AbstractRemoteWorker<INPUT extends AbstractData, OUTPUT extends AbstractData> extends AbstractWorker<INPUT, OUTPUT> {
+public abstract class AbstractRemoteWorker<INPUT extends RemoteData, OUTPUT extends RemoteData> extends AbstractWorker<INPUT, OUTPUT> {
 
     public AbstractRemoteWorker(ModuleManager moduleManager) {
         super(moduleManager);
diff --git a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractRemoteWorkerProvider.java b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractRemoteWorkerProvider.java
index 339b489f9..e007524d9 100644
--- a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractRemoteWorkerProvider.java
+++ b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractRemoteWorkerProvider.java
@@ -18,7 +18,7 @@
 
 package org.apache.skywalking.apm.collector.analysis.worker.model.base;
 
-import org.apache.skywalking.apm.collector.core.data.AbstractData;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
 
@@ -30,7 +30,7 @@ import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
  * @author peng-yongsheng
  * @since v3.0-2017
  */
-public abstract class AbstractRemoteWorkerProvider<INPUT extends AbstractData, OUTPUT extends AbstractData, WORKER_TYPE extends AbstractRemoteWorker<INPUT, OUTPUT>> extends AbstractWorkerProvider<INPUT, OUTPUT, WORKER_TYPE> {
+public abstract class AbstractRemoteWorkerProvider<INPUT extends RemoteData, OUTPUT extends RemoteData, WORKER_TYPE extends AbstractRemoteWorker<INPUT, OUTPUT>> extends AbstractWorkerProvider<INPUT, OUTPUT, WORKER_TYPE> {
 
     private final RemoteSenderService remoteSenderService;
     private final int graphId;
diff --git a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractWorker.java b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractWorker.java
index da4efdbbc..e3b697ba2 100644
--- a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractWorker.java
+++ b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractWorker.java
@@ -18,7 +18,6 @@
 
 package org.apache.skywalking.apm.collector.analysis.worker.model.base;
 
-import org.apache.skywalking.apm.collector.core.data.EndOfBatchQueueMessage;
 import org.apache.skywalking.apm.collector.core.graph.Next;
 import org.apache.skywalking.apm.collector.core.graph.NodeProcessor;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
@@ -28,13 +27,13 @@ import org.slf4j.LoggerFactory;
 /**
  * @author peng-yongsheng
  */
-public abstract class AbstractWorker<INPUT extends EndOfBatchQueueMessage, OUTPUT extends EndOfBatchQueueMessage> implements NodeProcessor<INPUT, OUTPUT> {
+public abstract class AbstractWorker<INPUT, OUTPUT> implements NodeProcessor<INPUT, OUTPUT> {
 
     private final Logger logger = LoggerFactory.getLogger(AbstractWorker.class);
 
     private final ModuleManager moduleManager;
 
-    public AbstractWorker(ModuleManager moduleManager) {
+    AbstractWorker(ModuleManager moduleManager) {
         this.moduleManager = moduleManager;
     }
 
diff --git a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractWorkerProvider.java b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractWorkerProvider.java
index 8ebc62d70..b312af532 100644
--- a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractWorkerProvider.java
+++ b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractWorkerProvider.java
@@ -18,17 +18,16 @@
 
 package org.apache.skywalking.apm.collector.analysis.worker.model.base;
 
-import org.apache.skywalking.apm.collector.core.data.EndOfBatchQueueMessage;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 
 /**
  * @author peng-yongsheng
  */
-public abstract class AbstractWorkerProvider<INPUT extends EndOfBatchQueueMessage, OUTPUT extends EndOfBatchQueueMessage, WORKER_TYPE extends AbstractWorker<INPUT, OUTPUT>> implements Provider {
+public abstract class AbstractWorkerProvider<INPUT, OUTPUT, WORKER_TYPE extends AbstractWorker<INPUT, OUTPUT>> implements Provider {
 
     private final ModuleManager moduleManager;
 
-    public AbstractWorkerProvider(ModuleManager moduleManager) {
+    AbstractWorkerProvider(ModuleManager moduleManager) {
         this.moduleManager = moduleManager;
     }
 
diff --git a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/LocalAsyncWorkerRef.java b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/LocalAsyncWorkerRef.java
index 58d831a20..da2e8107b 100644
--- a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/LocalAsyncWorkerRef.java
+++ b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/LocalAsyncWorkerRef.java
@@ -20,8 +20,9 @@ package org.apache.skywalking.apm.collector.analysis.worker.model.base;
 
 import java.util.Iterator;
 import java.util.List;
-import org.apache.skywalking.apm.collector.core.data.EndOfBatchQueueMessage;
+import org.apache.skywalking.apm.collector.core.data.QueueData;
 import org.apache.skywalking.apm.collector.core.graph.NodeProcessor;
+import org.apache.skywalking.apm.collector.core.queue.EndOfBatchContext;
 import org.apache.skywalking.apm.commons.datacarrier.DataCarrier;
 import org.apache.skywalking.apm.commons.datacarrier.consumer.IConsumer;
 import org.slf4j.Logger;
@@ -30,7 +31,7 @@ import org.slf4j.LoggerFactory;
 /**
  * @author peng-yongsheng
  */
-public class LocalAsyncWorkerRef<INPUT extends EndOfBatchQueueMessage, OUTPUT extends EndOfBatchQueueMessage> extends WorkerRef<INPUT, OUTPUT> implements IConsumer<INPUT> {
+public class LocalAsyncWorkerRef<INPUT extends QueueData, OUTPUT extends QueueData> extends WorkerRef<INPUT, OUTPUT> implements IConsumer<INPUT> {
 
     private final Logger logger = LoggerFactory.getLogger(LocalAsyncWorkerRef.class);
 
@@ -40,7 +41,7 @@ public class LocalAsyncWorkerRef<INPUT extends EndOfBatchQueueMessage, OUTPUT ex
         super(destinationHandler);
     }
 
-    public void setQueueEventHandler(DataCarrier<INPUT> dataCarrier) {
+    void setQueueEventHandler(DataCarrier<INPUT> dataCarrier) {
         this.dataCarrier = dataCarrier;
     }
 
@@ -52,7 +53,7 @@ public class LocalAsyncWorkerRef<INPUT extends EndOfBatchQueueMessage, OUTPUT ex
             INPUT input = inputIterator.next();
             i++;
             if (i == data.size()) {
-                input.setEndOfBatch(true);
+                input.getEndOfBatchContext().setEndOfBatch(true);
             }
             out(input);
         }
@@ -69,6 +70,7 @@ public class LocalAsyncWorkerRef<INPUT extends EndOfBatchQueueMessage, OUTPUT ex
     }
 
     @Override protected void in(INPUT input) {
+        input.setEndOfBatchContext(new EndOfBatchContext(false));
         dataCarrier.produce(input);
     }
 
diff --git a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/RemoteWorkerRef.java b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/RemoteWorkerRef.java
index a5ec2b515..28e384740 100644
--- a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/RemoteWorkerRef.java
+++ b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/RemoteWorkerRef.java
@@ -18,7 +18,7 @@
 
 package org.apache.skywalking.apm.collector.analysis.worker.model.base;
 
-import org.apache.skywalking.apm.collector.core.data.AbstractData;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -26,7 +26,7 @@ import org.slf4j.LoggerFactory;
 /**
  * @author peng-yongsheng
  */
-public class RemoteWorkerRef<INPUT extends AbstractData, OUTPUT extends AbstractData> extends WorkerRef<INPUT, OUTPUT> {
+public class RemoteWorkerRef<INPUT extends RemoteData, OUTPUT extends RemoteData> extends WorkerRef<INPUT, OUTPUT> {
 
     private final Logger logger = LoggerFactory.getLogger(RemoteWorkerRef.class);
 
diff --git a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/WorkerCreateListener.java b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/WorkerCreateListener.java
index 23e602d28..fe1a48f72 100644
--- a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/WorkerCreateListener.java
+++ b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/WorkerCreateListener.java
@@ -33,7 +33,7 @@ public class WorkerCreateListener {
         this.persistenceWorkers = new ArrayList<>();
     }
 
-    public void addWorker(AbstractWorker worker) {
+    void addWorker(AbstractWorker worker) {
         if (worker instanceof PersistenceWorker) {
             persistenceWorkers.add((PersistenceWorker)worker);
         }
diff --git a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/WorkerRef.java b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/WorkerRef.java
index b1fa00b5d..3dbb71d27 100644
--- a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/WorkerRef.java
+++ b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/WorkerRef.java
@@ -24,7 +24,7 @@ import org.apache.skywalking.apm.collector.core.graph.WayToNode;
 /**
  * @author peng-yongsheng
  */
-public abstract class WorkerRef<INPUT, OUTPUT> extends WayToNode<INPUT, OUTPUT> {
+abstract class WorkerRef<INPUT, OUTPUT> extends WayToNode<INPUT, OUTPUT> {
     WorkerRef(NodeProcessor<INPUT, OUTPUT> destinationHandler) {
         super(destinationHandler);
     }