00.At-Least-Once Delivery

class Msg implements Serializable {
  private static final long serialVersionUID = 1L;
  public final long deliveryId;
  public final String s;

  public Msg(long deliveryId, String s) {
    this.deliveryId = deliveryId;
    this.s = s;
  }
}

class Confirm implements Serializable {
  private static final long serialVersionUID = 1L;
  public final long deliveryId;

  public Confirm(long deliveryId) {
    this.deliveryId = deliveryId;
  }
}

class MsgSent implements Serializable {
  private static final long serialVersionUID = 1L;
  public final String s;

  public MsgSent(String s) {
    this.s = s;
  }
}
class MsgConfirmed implements Serializable {
  private static final long serialVersionUID = 1L;
  public final long deliveryId;

  public MsgConfirmed(long deliveryId) {
    this.deliveryId = deliveryId;
  }
}

@Component
@Scope("prototype")
public class MyPersistentActor extends AbstractPersistentActorWithAtLeastOnceDelivery {
	  private final ActorSelection destination;
	  
	  private int	msgCnt = 0;	  

	  public MyPersistentActor(ActorSelection destination) {		  
	      this.destination =destination;
	  }

	  @Override public String persistenceId() {
	    return "persistence-id";
	  }

	  @Override
	  public Receive createReceive() {
	    return receiveBuilder().
	      match(String.class, s -> {
	    	//메시지가 전송중
	        persist(new MsgSent(s), evt -> updateState(evt));	        	        
	      }).
	      match(Confirm.class, confirm -> {
	    	this.msgCnt++;
	    	  
	    	if(this.msgCnt % 2 == 0) {
		    	//의도적으로 짝수메시지만 확인함...
		        persist(new MsgConfirmed(confirm.deliveryId), evt -> updateState(evt));		        
		        System.out.println("msg ok :" + confirm.deliveryId  );
	    	}else {
	    		//의도적 으로 메시지 확인을 드롭...(다음번에 성공되어야함)
	    		System.out.println("drop msg for test:" + confirm.deliveryId );
	    	}
	      }).
	      build();
	  }

	  @Override
	  public Receive createReceiveRecover() {
	    return receiveBuilder().
	        match(Object.class, evt -> updateState(evt)).build();
	  }

	  void updateState(Object event) {
	    if (event instanceof MsgSent) {
	      final MsgSent evt = (MsgSent) event;
	      deliver(destination, deliveryId -> new Msg(deliveryId, evt.s));
	    } else if (event instanceof MsgConfirmed) {
	      final MsgConfirmed evt = (MsgConfirmed) event;
	      confirmDelivery(evt.deliveryId);
	    }
	  }
}

@Component
@Scope("prototype")
public class MyDestination extends AbstractActor {
	  @Override
	  public Receive createReceive() {
	    return receiveBuilder()
	      .match(Msg.class, msg -> {
	    	//메시지를 받고 할일을 하고..
	    	//....	    	  
	    	  
	    	//수신 메시지 확인됨을 보냄..
	        getSender().tell(new Confirm(msg.deliveryId), getSelf());
	      })
	      .build();
	  }
}

	protected void persistenceMessage(ActorSystem system,SpringExtension ext)  {
	    new TestKit(system) {{
	    	ActorRef probe = getRef();
	    	//메시지 목적대상 액터 생성
	    	final ActorRef myDestination = system.actorOf(ext.props("myDestination"),"myDes");	    	
	    	final ActorSelection myDes = system.actorSelection("/user/myDes");
	    	
	    	//재전송가능한 메시지 발송기 생성 ( 옵션은 목적대상 액터선택자 ) 
	        final ActorRef myPersistentActor = system.actorOf( ext.props("myPersistentActor",myDes ), "myPersistentActor"  );
	        
	        for(int i=0;i<10;i++) {
	        	myPersistentActor.tell( "hi " + i ,ActorRef.noSender() );	        	
	        }
	        
	        //장애 시나리오를 가정하고,메시지가 모두 처리되는 시간 25초로 가정
	        expectNoMessage(java.time.Duration.ofSeconds(25));	        
	    }};
	}