Java Generate/Merge Files(2)Akka and Actor

Java Generate/Merge Files(2)Akka and Actor

root guardian is the father of all actors,
User Actor is the father of all user actors, path is /user
System Actor is the father of all system actors, path is /system

actorOf - create a new Actor
actorSelection - select and search Actor
actorFor - use actorSelection

AKKA Configuration application.conf
akka.actor.deployment {
  user/sourceRouter {
    router = smallest-mailbox-pool
    resizer {
      lower-bound = 2
      upper-bound = 15
    }
  }
  user/jobIDRouter {
    router = smallest-mailbox-pool
    resizer {
      lower-bound = 20
      upper-bound = 100
    }
  }
}

XML related AKKA Config
<bean id="actorSystemFactory" class="com.sillycat.feeds2g.services.base.ActorSystemFactory"
init-method="init" />

<bean id="sourceIDsExportActor" class="com.sillycat.feeds2g.services.actors.SourceIDsExportActor"


scope="prototype">
<property name="sourceDAO" ref="sourceDAO" />

</bean>
<bean id="jobIDExportActor" class="com.sillycat.feeds2g.services.actors.JobIDExportActor"


scope="prototype">
<property name="redisService" ref="redisService" />

</bean>

<bean id="referenceIDsExportActor" class="com.sillycat.feeds2g.services.actors.ReferenceIDsExportActor"


scope="prototype">
<property name="redisService" ref="redisService" />
</bean>

They are just normal spring typical configurations. Some base java factory classes.
package com.sillycat.feeds2g.services.base;

import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import akka.actor.ActorSystem;

public class ActorSystemFactory implements ApplicationContextAware {

    private ApplicationContext applicationContext;

    private ActorSystem system;

    public void setApplicationContext(ApplicationContext context) throws BeansException {
        applicationContext = context;
    }

    public void init() {
        system = ActorSystem.create("feeds2g");
        // initialize the application context in the Akka Spring Extension
        SpringExtension.SpringExtProvider.get(system).initialize(applicationContext);
    }

    public ActorSystem getActorSystem() {
        return system;
    }

}

Actor spring integration support
package com.sillycat.feeds2g.services.base;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import akka.actor.Actor;
import akka.actor.IndirectActorProducer;

public class SpringActorProducer implements IndirectActorProducer {
    final ClassPathXmlApplicationContext applicationContext;
    final String actorBeanName;

    public SpringActorProducer(ApplicationContext applicationContext, String actorBeanName) {
        this.applicationContext = (ClassPathXmlApplicationContext) applicationContext;
        this.actorBeanName = actorBeanName;
    }

    @Override
    public Actor produce() {
        applicationContext.refresh();
        return (Actor) applicationContext.getBean(actorBeanName);
    }

    @SuppressWarnings("unchecked")
    @Override
    public Class<? extends Actor> actorClass() {
        applicationContext.refresh();
        return (Class<? extends Actor>) applicationContext.getType(actorBeanName);
    }
}

package com.sillycat.feeds2g.services.base;

import org.springframework.context.ApplicationContext;

import akka.actor.AbstractExtensionId;
import akka.actor.ExtendedActorSystem;
import akka.actor.Extension;
import akka.actor.Props;

public class SpringExtension extends AbstractExtensionId<SpringExtension.SpringExt> {

    /**
     * The identifier used to access the SpringExtension.
     */
    public static SpringExtension SpringExtProvider = new SpringExtension();

    /**
     * Is used by Akka to instantiate the Extension identified by this
     * ExtensionId, internal use only.
     */
    @Override
    public SpringExt createExtension(ExtendedActorSystem system) {
        return new SpringExt();
    }

    /**
     * The Extension implementation.
     */
    public static class SpringExt implements Extension {
        private volatile ApplicationContext applicationContext;

        /**
         * Used to initialize the Spring application context for the extension.
         *
         * @param applicationContext
         */
        public void initialize(ApplicationContext applicationContext) {
            this.applicationContext = applicationContext;
        }

        /**
         * Create a Props for the specified actorBeanName using the
         * SpringActorProducer class.
         *
         * @param actorBeanName
         *            The name of the actor bean to create Props for
         * @return a Props that will create the named actor bean using Spring
         */
        public Props props(String actorBeanName) {
            return Props.create(SpringActorProducer.class, applicationContext, actorBeanName);
        }
    }
}

Related pom.xml configuration to load the dependencies and build.
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
                      http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.sillycat</groupId>
<artifactId>feeds-2g</artifactId>
<version>1.0</version>
<description>feeds next generation</description>
<name>feeds next generation</name>
<packaging>jar</packaging>
<properties>
<springframework.version>4.3.7.RELEASE</springframework.version>
<jackson.version>2.8.7</jackson.version>
<akka.version></akka.version>
</properties>
<dependencies>
<!-- akka -->
<dependency>
<groupId>com.typesafe.akka</groupId>
<artifactId>akka-actor_2.11</artifactId>
<version>2.4.17</version>
</dependency>
<!-- myIbatis -->
<dependency>
<groupId>org.mybatis</groupId>
<artifactId>mybatis-spring</artifactId>
<version>1.3.1</version>
</dependency>
<!-- logging -->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.25</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>jcl-over-slf4j</artifactId>
<version>1.7.25</version>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>1.2.3</version>
</dependency>
<!-- REDIS -->
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>2.9.0</version>
</dependency>
<!-- spring framework -->
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>${springframework.version}</version>
<exclusions>
<exclusion>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
</exclusion>
</exclusions>
</dependency>
<!-- apache -->
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.5</version>
</dependency>
<!-- JSON -->
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-core</artifactId>
<version>${jackson.version}</version>
</dependency>
<!-- testing -->
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>${springframework.version}</version>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-s3</artifactId>
<version>1.11.113</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.6.1</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>2.4.1</version>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
<archive>
<manifest>
<mainClass>com.sillycat.feeds2g.ExecutorApp</mainClass>
</manifest>
</archive>
</configuration>
<executions>
<execution>
<id>assemble-all</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>3.0.0</version>
<executions>
<execution>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<transformers>
<transformer
implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>META-INF/spring.handlers</resource>
</transformer>
<transformer
implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>META-INF/spring.schemas</resource>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

One typical Actor

package com.sillycat.feeds2g.services.actors;

import java.util.Arrays;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.sillycat.feeds2g.models.ReferenceIDsResult;
import com.sillycat.feeds2g.models.messages.CampaignIDMessage;
import com.sillycat.feeds2g.models.messages.ReferenceIDsMessage;
import com.sillycat.feeds2g.models.messages.SourceIDMessage;
import com.sillycat.feeds2g.services.RedisService;

import akka.actor.ActorRef;
import akka.actor.ActorSelection;
import akka.actor.UntypedActor;

public class ReferenceIDsExportActor extends UntypedActor {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    private RedisService redisService;

    private ActorSelection jobIDRouter = getContext().actorSelection("akka://feeds2g/user/jobIDRouter");

    public void setRedisService(RedisService redisService) {
        this.redisService = redisService;
    }

    public void onReceive(Object msg) throws Throwable {
        if (msg instanceof CampaignIDMessage) {
            // SSCAN REDIS to find the referenceIDs
            logger.debug("ReferenceIDsExportActor get message " + msg);

        } else if (msg instanceof SourceIDMessage) {
            // SSCAN REDIS to find the referenceIDs
            logger.debug("ReferenceIDsExportActor get message " + msg);
            SourceIDMessage sourceIDMessage = (SourceIDMessage) msg;
            Integer sourceID = sourceIDMessage.getSourceID();
            logger.info("System ReferenceIDsExportActor start to export jobs for sourceID = " + sourceID);
            String cursor = "0";
            do {
                ReferenceIDsResult result = redisService.fetchReferenceIDsBySource(sourceID, cursor);
                cursor = result.getCursor();
                String referenceIDString = result.getReferenceIDString();
                String[] referenceArray = referenceIDString.split(",");
                List<String> referenceIDs = Arrays.asList(referenceArray);

                jobIDRouter.tell(new ReferenceIDsMessage(sourceID, referenceIDs), ActorRef.noSender());

            } while (!"0".equals(cursor));
            logger.info("System ReferenceIDsExportActor finished sscanning all referenceIDs for sourceID = " + sourceID);
        } else {
            unhandled(msg);
        }
    }

}

References:
http://sunxiang0918.cn/2016/01/10/Akka-in-JAVA-1/
http://sunxiang0918.cn/2016/01/13/Akka-in-JAVA-2/
http://sunxiang0918.cn/2016/01/18/Akka-in-JAVA-3/
http://sunxiang0918.cn/2016/02/10/Akka-in-JAVA-4/
demo
https://github.com/sunxiang0918/AkkaDemo

http://doc.akka.io/docs/akka-modules/1.3.1/modules/spring.html

http://stackoverflow.com/questions/11849254/akka-and-spring-configuration
https://github.com/aliakh/demo-akka-spring

https://github.com/XiaoMi/rose/tree/master/rose-example
http://doc.akka.io/docs/akka/current/general/configuration.html

Redis
https://redis.io/commands/rpoplpush
https://blog.logentries.com/2016/05/queuing-tasks-with-redis/