Fault tolerance code sample with AKKA and JAVA.

DivisionActor.java

/**
 * 
 */
package com.akkasample.actor;

import com.akkasample.bean.DivOperands;

import akka.actor.UntypedActor;
import scala.Option;

/**
 * This actor handles the division math operation 
 * @author Rakesh
 *
 */
public class DivisionActor extends UntypedActor {
	
	/**
	 * Actor method call before restart of the actor.
	 */
	@Override
	public void preRestart(Throwable reason, Option<Object> message) throws Exception {
		System.out.println("Pre restart called");
		super.preRestart(reason, message);
	}
	
	/**
	 * Method call before actor onReceive method call.
	 */
	@Override
	public void preStart() throws Exception {
		System.out.println("Pre start called");
		super.preStart();
	}
	
	
	/**
	 * Actor implemented method to handle the messages
	 */

	/* (non-Javadoc)
	 * @see akka.actor.UntypedActor#onReceive(java.lang.Object)
	 */
	@Override
	public void onReceive(Object object) throws Exception {
		if(object instanceof DivOperands){
			DivOperands divOperands=(DivOperands)object;
			Long opr1 = Long.parseLong(divOperands.getOpr1());
			Long opr2 = Long.parseLong(divOperands.getOpr2());
			System.out.println(opr1/opr2);
			
		}else{
			unhandled(object);
		}
	}

}

FaultToleranceMain.java

package com.akkasample;

import com.akkasample.actor.MathOperation;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;

/**
 * Main method to run the example
 * @author Rakesh
 *
 */
public class FaultToleranceMain {

	public static void main(String[] args) {
		ActorSystem system = ActorSystem.create();
		ActorRef mathOp = system.actorOf(Props.create(MathOperation.class),"MathOperation");
		mathOp.tell("start", ActorRef.noSender());
		system.shutdown();
	}

}

MathOperation.java

/**
 * 
 */
package com.akkasample.actor;

import java.util.concurrent.TimeUnit;

import com.akkasample.bean.DivOperands;

import akka.actor.ActorRef;
import akka.actor.OneForOneStrategy;
import akka.actor.Props;
import akka.actor.SupervisorStrategy;
import akka.actor.SupervisorStrategy.Directive;
import akka.actor.UntypedActor;
import akka.japi.Function;
import scala.concurrent.duration.Duration;

/**
 * Math operation Actor which supervise its child actors by handling various exception conditions.
 * @author Rakesh
 *
 */
public class MathOperation extends UntypedActor {
	ActorRef divActorRef = this.context().actorOf(Props.create(DivisionActor.class),"DivisionActor");
	/*
	 * Strategy to handle the exceptions occure inside the Division actor.
	 */
	SupervisorStrategy strategy = new OneForOneStrategy(3, Duration.create(1,TimeUnit.SECONDS),new Function<Throwable, Directive>(){

		public Directive apply(Throwable throwable) throws Exception {
			if(throwable instanceof ArithmeticException || throwable instanceof NumberFormatException){
				System.out.println("Resume called");
				return SupervisorStrategy.resume();
			}else{
				System.out.println("Stop called");
				return SupervisorStrategy.stop();
			}
			
		}
		
	});
	
	/**
	 * Set the supervisor strategy.
	 */
	@Override
	public SupervisorStrategy supervisorStrategy() {
		return strategy;
	};
	/**
	 * Message handler method of the actor.
	 */
	@Override
	public void onReceive(Object arg0) throws Exception {
		
		divActorRef.tell(new DivOperands("10", "0"), getSelf());// causes the Arithmetic exception / by zero.
		divActorRef.tell(new DivOperands("10.", "0"), getSelf()); // causes the NumberFormat exception for "10."
		divActorRef.tell(new DivOperands("10", null), getSelf()); // causes the Number format exception for null
		
		
	}

}