actor provider

skywalking-collector/pom.xml

@@ -19,11 +19,6 @@
 	</properties>
 
 	<dependencies>
-		<dependency>
-			<groupId>com.a.eye</groupId>
-			<artifactId>skywalking-sniffer-mock</artifactId>
-			<version>${project.version}</version>
-		</dependency>
 		<dependency>
 			<groupId>com.typesafe.akka</groupId>
 			<artifactId>akka-cluster_2.11</artifactId>
@@ -49,5 +44,12 @@
 			<artifactId>leveldb</artifactId>
 			<version>0.9</version>
 		</dependency>
+
+		<dependency>
+			<groupId>com.a.eye</groupId>
+			<artifactId>skywalking-sniffer-mock</artifactId>
+			<version>${project.version}</version>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 </project>

skywalking-collector/skywalking-collector-cluster/pom.xml

@@ -27,5 +27,10 @@
             <artifactId>spring-context-support</artifactId>
             <version>${project.spring.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.a.eye</groupId>
+            <artifactId>skywalking-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
 </project>
\ No newline at end of file

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/cluster/base/AbstractUntypedActor.java

+package com.a.eye.skywalking.collector.cluster.base;
+
+import akka.actor.ActorSystem;
+import akka.actor.UntypedActor;
+
+/**
+ * @author pengys5
+ */
+public abstract class AbstractUntypedActor extends UntypedActor {
+}

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/cluster/base/IActorProvider.java

+package com.a.eye.skywalking.collector.cluster.base;
+
+import akka.actor.ActorSystem;
+import com.a.eye.skywalking.collector.cluster.config.CollectorConfig;
+
+/**
+ * @author pengys5
+ */
+public interface IActorProvider {
+    public String actorName();
+
+    public void createActor(ActorSystem system);
+
+    public void actorOf(ActorSystem system, String actorInClusterName);
+}

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/cluster/config/CollectorConfig.java

@@ -11,5 +11,9 @@ public class CollectorConfig {
         public static String hostname = "127.0.0.1";
         public static String port = "2551";
         public static String cluster = "127.0.0.1:2551";
+
+        public static class Actor {
+            public static int ActorManagerActor_Num = 2;
+        }
     }
 }

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/cluster/config/CollectorConfigInitializer.java

 package com.a.eye.skywalking.collector.cluster.config;
 
-import com.a.eye.skywalking.api.conf.Config;
 import com.a.eye.skywalking.api.logging.api.ILog;
 import com.a.eye.skywalking.api.logging.api.LogManager;
 import com.a.eye.skywalking.api.util.ConfigInitializer;

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/cluster/manager/ActorCreator.java

+package com.a.eye.skywalking.collector.cluster.manager;
+
+import akka.actor.ActorSystem;
+import com.a.eye.skywalking.collector.cluster.base.IActorProvider;
+
+import java.util.ServiceLoader;
+
+/**
+ * @author pengys5
+ */
+public enum ActorCreator {
+    INSTANCE;
+
+    public void create(ActorSystem system) {
+        ServiceLoader<IActorProvider> serviceLoader = ServiceLoader.load(IActorProvider.class);
+        for (IActorProvider service : serviceLoader) {
+            service.createActor(system);
+        }
+    }
+}

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/cluster/manager/ActorManagerActor.java

 package com.a.eye.skywalking.collector.cluster.manager;
 
 import akka.actor.Terminated;
-import akka.actor.UntypedActor;
+import com.a.eye.skywalking.collector.cluster.base.AbstractUntypedActor;
+import com.a.eye.skywalking.collector.cluster.base.IActorProvider;
 import com.a.eye.skywalking.collector.cluster.message.ActorRegisterMessage;
 
 /**
  * Created by Administrator on 2017/2/21 0021.
  */
-public class ActorManagerActor extends UntypedActor {
+public class ActorManagerActor extends AbstractUntypedActor {
 
     @Override
     public void onReceive(Object message) throws Throwable {

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/cluster/manager/ActorManagerActorFactory.java

+package com.a.eye.skywalking.collector.cluster.manager;
+
+import akka.actor.ActorSystem;
+import akka.actor.Props;
+import com.a.eye.skywalking.collector.cluster.base.IActorProvider;
+import com.a.eye.skywalking.collector.cluster.config.CollectorConfig;
+
+/**
+ * @author pengys5
+ */
+public class ActorManagerActorFactory implements IActorProvider {
+
+    @Override
+    public String actorName() {
+        return "ActorManagerActor";
+    }
+
+    @Override
+    public void createActor(ActorSystem system) {
+        for (int i = 1; i <= CollectorConfig.Collector.Actor.ActorManagerActor_Num; i++) {
+            actorOf(system, actorName() + "_" + i);
+        }
+    }
+
+    @Override
+    public void actorOf(ActorSystem system, String actorInClusterName) {
+        system.actorOf(Props.create(ActorManagerActor.class), actorInClusterName);
+    }
+}

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/cluster/producer/TraceProducerApp.java

@@ -22,7 +22,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 import static akka.pattern.Patterns.ask;
 
 /**
- * {@link TraceProducerApp} is a producer for trace agent to send @link TraceSegment.
+ * {@link TraceProducerApp} is a producer for trace agent to send {@link TraceSegment}.
  * <p>
  * Created by pengys5 on 2017/2/17.
  */

skywalking-collector/skywalking-collector-cluster/src/main/resources/META-INF/services/com.a.eye.skywalking.collector.cluster.base.IActorProvider

+com.a.eye.skywalking.collector.cluster.manager.ActorManagerActorFactory
\ No newline at end of file

skywalking-collector/skywalking-collector-cluster/src/test/java/com/a/eye/skywalking/collector/cluster/ActorCreatorTestCase.java

+package com.a.eye.skywalking.collector.cluster;
+
+import akka.actor.ActorSystem;
+import akka.actor.Props;
+import com.a.eye.skywalking.collector.cluster.manager.ActorCreator;
+import com.a.eye.skywalking.collector.cluster.manager.ActorManagerActor;
+import org.junit.Test;
+
+import static org.mockito.Mockito.*;
+
+/**
+ * @author pengys5
+ */
+public class ActorCreatorTestCase {
+
+    @Test
+    public void testCreate() {
+        ActorSystem system = mock(ActorSystem.class);
+//        ActorCreator.INSTANCE.create(system, ActorManagerActor.class, 1);
+//        verify(system).actorOf(Props.create(ActorManagerActor.class), "ActorManagerActor");
+    }
+}

skywalking-collector/skywalking-collector-cluster/src/test/java/com/a/eye/skywalking/collector/cluster/ActorProviderTestCase.java

+package com.a.eye.skywalking.collector.cluster;
+
+import akka.actor.ActorSystem;
+import com.a.eye.skywalking.collector.cluster.manager.ActorManagerActorFactory;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+/**
+ * @author pengys5
+ */
+public class ActorProviderTestCase {
+
+    @Test
+    public void testActorName() {
+        ActorManagerActorFactory factory = new ActorManagerActorFactory();
+        String actorName = factory.actorName();
+        Assert.assertEquals("ActorManagerActor", actorName);
+    }
+
+    @Test
+    public void testCreateActor() {
+        ActorSystem system = Mockito.mock(ActorSystem.class);
+        ActorManagerActorFactory factory = new ActorManagerActorFactory();
+        factory.createActor(system);
+    }
+}

skywalking-collector/skywalking-collector-cluster/src/test/java/com/a/eye/skywalking/collector/cluster/CollectorConfigTestCase.java

@@ -7,8 +7,6 @@ import com.typesafe.config.Config;
 import org.junit.Assert;
 import org.junit.Test;
 
-import java.util.ArrayList;
-
 /**
  * Created by pengys5 on 2017/2/22 0022.
  */

skywalking-collector/skywalking-collector-worker/pom.xml

@@ -11,5 +11,11 @@
 
     <artifactId>skywalking-collector-worker</artifactId>
 
-
+    <dependencies>
+        <dependency>
+            <groupId>com.a.eye</groupId>
+            <artifactId>skywalking-collector-cluster</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+    </dependencies>
 </project>
\ No newline at end of file

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/CollectorBootStartUp.java

+package com.a.eye.skywalking.collector.worker;
+
+import akka.actor.ActorSystem;
+import akka.actor.Props;
+import com.a.eye.skywalking.collector.cluster.Const;
+import com.a.eye.skywalking.collector.cluster.consumer.TraceConsumerActor;
+
+/**
+ * @author pengys5
+ */
+public class CollectorBootStartUp {
+
+    public static void main(String[] args) {
+//        ActorSystem system = ActorSystem.create("ClusterSystem", config);
+//        system.actorOf(Props.create(TraceConsumerActor.class), Const.Trace_Consumer_Role);
+    }
+}

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/indicator/ApplicationDiscoverActor.java

+package com.a.eye.skywalking.collector.worker.indicator;
+
+import akka.actor.UntypedActor;
+import com.a.eye.skywalking.collector.cluster.base.AbstractUntypedActor;
+
+/**
+ * @author pengys5
+ */
+public class ApplicationDiscoverActor extends AbstractUntypedActor {
+
+    public static final String ActorName = "ApplicationDiscoverActor";
+
+    @Override
+    public String actorName() {
+        return ActorName;
+    }
+
+    @Override
+    public void onReceive(Object message) throws Throwable {
+
+    }
+}

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/indicator/TraceSegmentRelationActor.java

+package com.a.eye.skywalking.collector.worker.indicator;
+
+import akka.actor.UntypedActor;
+
+/**
+ * @author pengys5
+ */
+public class TraceSegmentRelationActor extends UntypedActor{
+    @Override
+    public void onReceive(Object message) throws Throwable {
+
+    }
+}