ActionInvokeDistributedCommitedTransfer.java

package fucoin.actions.transaction;

import akka.actor.ActorRef;
import akka.actor.UntypedActorContext;
import fucoin.supervisor.DistributedCommitedTransferRequest;
import fucoin.supervisor.SuperVisorImpl;

public class ActionInvokeDistributedCommitedTransfer extends CoordinatorTransaction{

	private ActorRef source;
	private ActorRef target;
	private int amount;

	public ActionInvokeDistributedCommitedTransfer(ActorRef source,
			ActorRef target, int amount) {
		this.source=source;
		this.target=target;
		this.amount=amount;
	}

	@Override
	protected void onAction(ActorRef sender, ActorRef self,
			UntypedActorContext context, SuperVisorImpl superVisor) {
		log("invoke transaction "+source.path().name()+" sends "+amount+" to "+target.path().name());
		long timeout = System.currentTimeMillis()+500;
		DistributedCommitedTransferRequest ds = new DistributedCommitedTransferRequest(source,target,timeout);
		superVisor.addDistributedCommitedTransferRequest(ds);
		ActionPrepareDistributedCommittedTransfer apdct = new ActionPrepareDistributedCommittedTransfer(source,target,amount,timeout,ds.getId());
		for(ActorRef neighbor : superVisor.getKnownNeighbors().values()){
			neighbor.tell(apdct, self);
		}
	}

}