Analysis register module finished.

apm-collector/apm-collector-analysis/analysis-register/register-define/src/main/java/org/apache/skywalking/apm/collector/analysis/register/define/graph/GraphIdDefine.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.analysis.register.define.graph;
+
+/**
+ * @author peng-yongsheng
+ */
+public class GraphIdDefine {
+    public static final int APPLICATION_REGISTER_GRAPH_ID = 200;
+    public static final int INSTANCE_REGISTER_GRAPH_ID = 201;
+    public static final int SERVICE_NAME_REGISTER_GRAPH_ID = 202;
+}

apm-collector/apm-collector-analysis/analysis-register/register-define/src/main/java/org/apache/skywalking/apm/collector/analysis/register/define/graph/WorkerIdDefine.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.analysis.register.define.graph;
+
+/**
+ * @author peng-yongsheng
+ */
+public class WorkerIdDefine {
+    public static final int APPLICATION_REGISTER_REMOTE_WORKER = 200;
+    public static final int APPLICATION_REGISTER_SERIAL_WORKER = 201;
+    public static final int INSTANCE_REGISTER_REMOTE_WORKER = 202;
+    public static final int INSTANCE_REGISTER_SERIAL_WORKER = 203;
+    public static final int SERVICE_NAME_REGISTER_REMOTE_WORKER = 204;
+    public static final int SERVICE_NAME_REGISTER_SERIAL_WORKER = 205;
+}

apm-collector/apm-collector-analysis/analysis-register/register-provider/pom.xml

@@ -36,5 +36,20 @@
             <artifactId>register-define</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>collector-cache-define</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>collector-storage-define</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>analysis-worker-model</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
 </project>
\ No newline at end of file

apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/AnalysisRegisterModuleProvider.java

@@ -23,6 +23,9 @@ import org.apache.skywalking.apm.collector.analysis.register.define.AnalysisRegi
 import org.apache.skywalking.apm.collector.analysis.register.define.service.IApplicationIDService;
 import org.apache.skywalking.apm.collector.analysis.register.define.service.IInstanceIDService;
 import org.apache.skywalking.apm.collector.analysis.register.define.service.IServiceNameService;
+import org.apache.skywalking.apm.collector.analysis.register.provider.register.ApplicationRegisterGraph;
+import org.apache.skywalking.apm.collector.analysis.register.provider.register.InstanceRegisterGraph;
+import org.apache.skywalking.apm.collector.analysis.register.provider.register.ServiceNameRegisterGraph;
 import org.apache.skywalking.apm.collector.analysis.register.provider.service.ApplicationIDService;
 import org.apache.skywalking.apm.collector.analysis.register.provider.service.InstanceIDService;
 import org.apache.skywalking.apm.collector.analysis.register.provider.service.ServiceNameService;
@@ -52,7 +55,14 @@ public class AnalysisRegisterModuleProvider extends ModuleProvider {
     }
 
     @Override public void start(Properties config) throws ServiceNotProvidedException {
+        ApplicationRegisterGraph applicationRegisterGraph = new ApplicationRegisterGraph(getManager());
+        applicationRegisterGraph.create();
 
+        InstanceRegisterGraph instanceRegisterGraph = new InstanceRegisterGraph(getManager());
+        instanceRegisterGraph.create();
+
+        ServiceNameRegisterGraph serviceNameRegisterGraph = new ServiceNameRegisterGraph(getManager());
+        serviceNameRegisterGraph.create();
     }
 
     @Override public void notifyAfterCompleted() throws ServiceNotProvidedException {

apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/ApplicationRegisterGraph.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.analysis.register.provider.register;
+
+import org.apache.skywalking.apm.collector.analysis.register.define.graph.GraphIdDefine;
+import org.apache.skywalking.apm.collector.core.graph.GraphManager;
+import org.apache.skywalking.apm.collector.core.module.ModuleManager;
+import org.apache.skywalking.apm.collector.remote.RemoteModule;
+import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
+import org.apache.skywalking.apm.collector.storage.table.register.Application;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ApplicationRegisterGraph {
+
+    private final ModuleManager moduleManager;
+
+    public ApplicationRegisterGraph(ModuleManager moduleManager) {
+        this.moduleManager = moduleManager;
+    }
+
+    public void create() {
+        RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
+
+        GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.APPLICATION_REGISTER_GRAPH_ID, Application.class)
+            .addNode(new ApplicationRegisterRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.APPLICATION_REGISTER_GRAPH_ID).create(null))
+            .addNext(new ApplicationRegisterSerialWorker.Factory(moduleManager).create(null));
+    }
+}

apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/register/ApplicationRegisterRemoteWorker.java → apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/ApplicationRegisterRemoteWorker.java

@@ -16,16 +16,16 @@
  *
  */
 
+package org.apache.skywalking.apm.collector.analysis.register.provider.register;
 
-package org.apache.skywalking.apm.collector.agent.stream.worker.register;
-
+import org.apache.skywalking.apm.collector.analysis.register.define.graph.WorkerIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
-import org.apache.skywalking.apm.collector.storage.table.register.Application;
-import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider;
 import org.apache.skywalking.apm.collector.remote.service.Selector;
-import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
-import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
+import org.apache.skywalking.apm.collector.storage.table.register.Application;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -41,7 +41,7 @@ public class ApplicationRegisterRemoteWorker extends AbstractRemoteWorker<Applic
     }
 
     @Override public int id() {
-        return 10006;
+        return WorkerIdDefine.APPLICATION_REGISTER_REMOTE_WORKER;
     }
 
     @Override protected void onWork(Application message) throws WorkerException {

apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/register/ApplicationRegisterSerialWorker.java → apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/ApplicationRegisterSerialWorker.java

@@ -16,21 +16,19 @@
  *
  */
 
+package org.apache.skywalking.apm.collector.analysis.register.provider.register;
 
-package org.apache.skywalking.apm.collector.agent.stream.worker.register;
-
-import org.apache.skywalking.apm.collector.agent.stream.IdAutoIncrement;
+import org.apache.skywalking.apm.collector.analysis.register.define.graph.WorkerIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorker;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
 import org.apache.skywalking.apm.collector.cache.CacheModule;
+import org.apache.skywalking.apm.collector.cache.service.ApplicationCacheService;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.core.util.Const;
+import org.apache.skywalking.apm.collector.storage.StorageModule;
 import org.apache.skywalking.apm.collector.storage.dao.IApplicationRegisterDAO;
 import org.apache.skywalking.apm.collector.storage.table.register.Application;
-import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorker;
-import org.apache.skywalking.apm.collector.cache.service.ApplicationCacheService;
-import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
-import org.apache.skywalking.apm.collector.storage.StorageModule;
-import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
-import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -51,7 +49,7 @@ public class ApplicationRegisterSerialWorker extends AbstractLocalAsyncWorker<Ap
     }
 
     @Override public int id() {
-        return 101;
+        return WorkerIdDefine.APPLICATION_REGISTER_SERIAL_WORKER;
     }
 
     @Override protected void onWork(Application application) throws WorkerException {
@@ -84,8 +82,8 @@ public class ApplicationRegisterSerialWorker extends AbstractLocalAsyncWorker<Ap
 
     public static class Factory extends AbstractLocalAsyncWorkerProvider<Application, Application, ApplicationRegisterSerialWorker> {
 
-        public Factory(ModuleManager moduleManager, QueueCreatorService<Application> queueCreatorService) {
-            super(moduleManager, queueCreatorService);
+        public Factory(ModuleManager moduleManager) {
+            super(moduleManager);
         }
 
         @Override public ApplicationRegisterSerialWorker workerInstance(ModuleManager moduleManager) {

apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/IdAutoIncrement.java → apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/IdAutoIncrement.java

@@ -17,7 +17,7 @@
  */
 
 
-package org.apache.skywalking.apm.collector.agent.stream;
+package org.apache.skywalking.apm.collector.analysis.register.provider.register;
 
 /**
  * @author peng-yongsheng

apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/InstanceRegisterGraph.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.analysis.register.provider.register;
+
+import org.apache.skywalking.apm.collector.analysis.register.define.graph.GraphIdDefine;
+import org.apache.skywalking.apm.collector.core.graph.GraphManager;
+import org.apache.skywalking.apm.collector.core.module.ModuleManager;
+import org.apache.skywalking.apm.collector.remote.RemoteModule;
+import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
+import org.apache.skywalking.apm.collector.storage.table.register.Instance;
+
+/**
+ * @author peng-yongsheng
+ */
+public class InstanceRegisterGraph {
+
+    private final ModuleManager moduleManager;
+
+    public InstanceRegisterGraph(ModuleManager moduleManager) {
+        this.moduleManager = moduleManager;
+    }
+
+    public void create() {
+        RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
+
+        GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.INSTANCE_REGISTER_GRAPH_ID, Instance.class)
+            .addNode(new InstanceRegisterRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.INSTANCE_REGISTER_GRAPH_ID).create(null))
+            .addNext(new InstanceRegisterSerialWorker.Factory(moduleManager).create(null));
+    }
+}

apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/register/InstanceRegisterRemoteWorker.java → apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/InstanceRegisterRemoteWorker.java

@@ -16,16 +16,16 @@
  *
  */
 
+package org.apache.skywalking.apm.collector.analysis.register.provider.register;
 
-package org.apache.skywalking.apm.collector.agent.stream.worker.register;
-
+import org.apache.skywalking.apm.collector.analysis.register.define.graph.WorkerIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
 import org.apache.skywalking.apm.collector.remote.service.Selector;
 import org.apache.skywalking.apm.collector.storage.table.register.Instance;
-import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
-import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider;
-import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -37,7 +37,7 @@ public class InstanceRegisterRemoteWorker extends AbstractRemoteWorker<Instance,
     private final Logger logger = LoggerFactory.getLogger(InstanceRegisterRemoteWorker.class);
 
     @Override public int id() {
-        return 10001;
+        return WorkerIdDefine.INSTANCE_REGISTER_REMOTE_WORKER;
     }
 
     InstanceRegisterRemoteWorker(ModuleManager moduleManager) {

apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/register/InstanceRegisterSerialWorker.java → apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/InstanceRegisterSerialWorker.java

@@ -16,19 +16,18 @@
  *
  */
 
+package org.apache.skywalking.apm.collector.analysis.register.provider.register;
 
-package org.apache.skywalking.apm.collector.agent.stream.worker.register;
-
+import org.apache.skywalking.apm.collector.analysis.register.define.graph.WorkerIdDefine;
 import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorker;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
 import org.apache.skywalking.apm.collector.cache.CacheModule;
 import org.apache.skywalking.apm.collector.cache.service.InstanceCacheService;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
-import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
 import org.apache.skywalking.apm.collector.storage.StorageModule;
 import org.apache.skywalking.apm.collector.storage.dao.IInstanceRegisterDAO;
 import org.apache.skywalking.apm.collector.storage.table.register.Instance;
-import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
-import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -49,7 +48,7 @@ public class InstanceRegisterSerialWorker extends AbstractLocalAsyncWorker<Insta
     }
 
     @Override public int id() {
-        return 102;
+        return WorkerIdDefine.INSTANCE_REGISTER_SERIAL_WORKER;
     }
 
     @Override protected void onWork(Instance instance) throws WorkerException {
@@ -83,8 +82,8 @@ public class InstanceRegisterSerialWorker extends AbstractLocalAsyncWorker<Insta
 
     public static class Factory extends AbstractLocalAsyncWorkerProvider<Instance, Instance, InstanceRegisterSerialWorker> {
 
-        public Factory(ModuleManager moduleManager, QueueCreatorService<Instance> queueCreatorService) {
-            super(moduleManager, queueCreatorService);
+        public Factory(ModuleManager moduleManager) {
+            super(moduleManager);
         }
 
         @Override public InstanceRegisterSerialWorker workerInstance(ModuleManager moduleManager) {

apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/ServiceNameRegisterGraph.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.analysis.register.provider.register;
+
+import org.apache.skywalking.apm.collector.analysis.register.define.graph.GraphIdDefine;
+import org.apache.skywalking.apm.collector.core.graph.GraphManager;
+import org.apache.skywalking.apm.collector.core.module.ModuleManager;
+import org.apache.skywalking.apm.collector.remote.RemoteModule;
+import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
+import org.apache.skywalking.apm.collector.storage.table.register.ServiceName;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ServiceNameRegisterGraph {
+
+    private final ModuleManager moduleManager;
+
+    public ServiceNameRegisterGraph(ModuleManager moduleManager) {
+        this.moduleManager = moduleManager;
+    }
+
+    public void create() {
+        RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
+
+        GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.SERVICE_NAME_REGISTER_GRAPH_ID, ServiceName.class)
+            .addNode(new ServiceNameRegisterRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.SERVICE_NAME_REGISTER_GRAPH_ID).create(null))
+            .addNext(new ServiceNameRegisterSerialWorker.Factory(moduleManager).create(null));
+    }
+}

apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/register/ServiceNameRegisterRemoteWorker.java → apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/ServiceNameRegisterRemoteWorker.java

@@ -16,16 +16,16 @@
  *
  */
 
+package org.apache.skywalking.apm.collector.analysis.register.provider.register;
 
-package org.apache.skywalking.apm.collector.agent.stream.worker.register;
-
+import org.apache.skywalking.apm.collector.analysis.register.define.graph.WorkerIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
-import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider;
 import org.apache.skywalking.apm.collector.remote.service.Selector;
 import org.apache.skywalking.apm.collector.storage.table.register.ServiceName;
-import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
-import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -41,7 +41,7 @@ public class ServiceNameRegisterRemoteWorker extends AbstractRemoteWorker<Servic
     }
 
     @Override public int id() {
-        return 10000;
+        return WorkerIdDefine.SERVICE_NAME_REGISTER_REMOTE_WORKER;
     }
 
     @Override protected void onWork(ServiceName serviceName) throws WorkerException {

apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/register/ServiceNameRegisterSerialWorker.java → apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/ServiceNameRegisterSerialWorker.java

@@ -16,21 +16,19 @@
  *
  */
 
+package org.apache.skywalking.apm.collector.analysis.register.provider.register;
 
-package org.apache.skywalking.apm.collector.agent.stream.worker.register;
-
-import org.apache.skywalking.apm.collector.agent.stream.IdAutoIncrement;
+import org.apache.skywalking.apm.collector.analysis.register.define.graph.WorkerIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorker;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
 import org.apache.skywalking.apm.collector.cache.CacheModule;
 import org.apache.skywalking.apm.collector.cache.service.ServiceIdCacheService;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.core.util.Const;
-import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
-import org.apache.skywalking.apm.collector.storage.dao.IServiceNameRegisterDAO;
-import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorker;
-import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
 import org.apache.skywalking.apm.collector.storage.StorageModule;
+import org.apache.skywalking.apm.collector.storage.dao.IServiceNameRegisterDAO;
 import org.apache.skywalking.apm.collector.storage.table.register.ServiceName;
-import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -51,7 +49,7 @@ public class ServiceNameRegisterSerialWorker extends AbstractLocalAsyncWorker<Se
     }
 
     @Override public int id() {
-        return 100;
+        return WorkerIdDefine.SERVICE_NAME_REGISTER_SERIAL_WORKER;
     }
 
     @Override protected void onWork(ServiceName serviceName) throws WorkerException {
@@ -87,8 +85,8 @@ public class ServiceNameRegisterSerialWorker extends AbstractLocalAsyncWorker<Se
 
     public static class Factory extends AbstractLocalAsyncWorkerProvider<ServiceName, ServiceName, ServiceNameRegisterSerialWorker> {
 
-        public Factory(ModuleManager moduleManager, QueueCreatorService<ServiceName> queueCreatorService) {
-            super(moduleManager, queueCreatorService);
+        public Factory(ModuleManager moduleManager) {
+            super(moduleManager);
         }
 
         @Override public ServiceNameRegisterSerialWorker workerInstance(ModuleManager moduleManager) {

apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/service/ApplicationIDService.java

@@ -18,10 +18,12 @@
 
 package org.apache.skywalking.apm.collector.analysis.register.provider.service;
 
+import org.apache.skywalking.apm.collector.analysis.register.define.graph.GraphIdDefine;
 import org.apache.skywalking.apm.collector.analysis.register.define.service.IApplicationIDService;
 import org.apache.skywalking.apm.collector.cache.CacheModule;
 import org.apache.skywalking.apm.collector.cache.service.ApplicationCacheService;
 import org.apache.skywalking.apm.collector.core.graph.Graph;
+import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.core.util.ObjectUtils;
 import org.apache.skywalking.apm.collector.storage.table.register.Application;
@@ -45,7 +47,7 @@ public class ApplicationIDService implements IApplicationIDService {
 
     private Graph<Application> getApplicationRegisterGraph() {
         if (ObjectUtils.isEmpty(applicationRegisterGraph)) {
-//            this.applicationRegisterGraph = GraphManager.INSTANCE.createIfAbsent(RegisterStreamGraphDefine.APPLICATION_REGISTER_GRAPH_ID, Application.class);
+            this.applicationRegisterGraph = GraphManager.INSTANCE.findGraph(GraphIdDefine.APPLICATION_REGISTER_GRAPH_ID, Application.class);
         }
         return this.applicationRegisterGraph;
     }

apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/service/InstanceIDService.java

@@ -18,10 +18,12 @@
 
 package org.apache.skywalking.apm.collector.analysis.register.provider.service;
 
+import org.apache.skywalking.apm.collector.analysis.register.define.graph.GraphIdDefine;
 import org.apache.skywalking.apm.collector.analysis.register.define.service.IInstanceIDService;
 import org.apache.skywalking.apm.collector.cache.CacheModule;
 import org.apache.skywalking.apm.collector.cache.service.InstanceCacheService;
 import org.apache.skywalking.apm.collector.core.graph.Graph;
+import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.core.util.ObjectUtils;
 import org.apache.skywalking.apm.collector.storage.StorageModule;
@@ -55,7 +57,7 @@ public class InstanceIDService implements IInstanceIDService {
 
     private Graph<Instance> getInstanceRegisterGraph() {
         if (ObjectUtils.isEmpty(instanceRegisterGraph)) {
-//            this.instanceRegisterGraph = GraphManager.INSTANCE.createIfAbsent(RegisterStreamGraphDefine.INSTANCE_REGISTER_GRAPH_ID, Instance.class);
+            this.instanceRegisterGraph = GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.INSTANCE_REGISTER_GRAPH_ID, Instance.class);
         }
         return instanceRegisterGraph;
     }

apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/service/ServiceNameService.java

@@ -18,10 +18,12 @@
 
 package org.apache.skywalking.apm.collector.analysis.register.provider.service;
 
+import org.apache.skywalking.apm.collector.analysis.register.define.graph.GraphIdDefine;
 import org.apache.skywalking.apm.collector.analysis.register.define.service.IServiceNameService;
 import org.apache.skywalking.apm.collector.cache.CacheModule;
 import org.apache.skywalking.apm.collector.cache.service.ServiceIdCacheService;
 import org.apache.skywalking.apm.collector.core.graph.Graph;
+import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.core.util.ObjectUtils;
 import org.apache.skywalking.apm.collector.storage.table.register.ServiceName;
@@ -52,7 +54,7 @@ public class ServiceNameService implements IServiceNameService {
 
     private Graph<ServiceName> getServiceNameRegisterGraph() {
         if (ObjectUtils.isEmpty(serviceNameRegisterGraph)) {
-//            this.serviceNameRegisterGraph = GraphManager.INSTANCE.createIfAbsent(RegisterStreamGraphDefine.SERVICE_NAME_REGISTER_GRAPH_ID, ServiceName.class);
+            this.serviceNameRegisterGraph = GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.SERVICE_NAME_REGISTER_GRAPH_ID, ServiceName.class);
         }
         return serviceNameRegisterGraph;
     }

apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractRemoteWorkerProvider.java

@@ -48,7 +48,7 @@ public abstract class AbstractRemoteWorkerProvider<INPUT extends Data, OUTPUT ex
      * @return The created worker reference. See {@link RemoteWorkerRef} worker instance, when the worker provider not
      * find then Throw this Exception.
      */
-    @Override final public RemoteWorkerRef create(WorkerCreateListener workerCreateListener) {
+    @Override final public RemoteWorkerRef<INPUT, OUTPUT> create(WorkerCreateListener workerCreateListener) {
         WORKER_TYPE remoteWorker = workerInstance(getModuleManager());
         workerCreateListener.addWorker(remoteWorker);
         return new RemoteWorkerRef<>(remoteWorker, remoteSenderService, graphId);