import akka.japi.pf.ReceiveBuilder; //导入依赖的package包/类
public Guest(ActorRef waiter, Coffee favoriteCoffee, FiniteDuration finishCoffeeDuration, int caffeineLimit) {
    this.waiter = waiter;
    this.favoriteCoffee = favoriteCoffee;
    this.finishCoffeeDuration = finishCoffeeDuration;
    this.caffeineLimit = caffeineLimit;
    orderFavoriteCoffee();

    receive(ReceiveBuilder.
            match(Waiter.CoffeeServed.class, coffeeServed -> coffeeServed.coffee.equals(favoriteCoffee), coffeeServed -> {
                coffeeCount++;
                log().info("Enjoying my {} yummy {}!", coffeeCount, coffeeServed.coffee);
                scheduleCoffeeFinished();
            }).
            match(Waiter.CoffeeServed.class, coffeeServed -> {
                log().info("Expected a {}, but got a {}!", favoriteCoffee, coffeeServed.coffee);
                waiter.tell(new Waiter.Complaint(favoriteCoffee), self());
            }).
            match(CoffeeFinished.class, coffeeFinished -> coffeeCount > this.caffeineLimit, coffeeFinished -> {
                throw new CaffeineException();
            }).
            match(CoffeeFinished.class, coffeeFinished ->
                    orderFavoriteCoffee()
            ).
            matchAny(this::unhandled).build()
    );
}
 

import akka.japi.pf.ReceiveBuilder; //导入依赖的package包/类
public Waiter(ActorRef coffeeHouse, ActorRef barista, int maxComplaintCount) {
    this.coffeeHouse = coffeeHouse;
    this.barista = barista;
    this.maxComplaintCount = maxComplaintCount;

    receive(ReceiveBuilder.
            match(ServeCoffee.class, serveCoffee ->
                    this.coffeeHouse.tell(new CoffeeHouse.ApproveCoffee(serveCoffee.coffee, sender()), self())
            ).
            match(Barista.CoffeePrepared.class, coffeePrepared ->
                    coffeePrepared.guest.tell(new CoffeeServed(coffeePrepared.coffee), self())
            ).
            match(Complaint.class, complaint -> complaintCount == this.maxComplaintCount, complaint -> {
                throw new FrustratedException(complaint.coffee, sender());
            }).
            match(Complaint.class, complaint -> {
                complaintCount++;
                this.barista.tell(new Barista.PrepareCoffee(complaint.coffee, sender()), self());
            }).
            matchAny(this::unhandled).build()
    );
}
 

import akka.japi.pf.ReceiveBuilder; //导入依赖的package包/类
public CoffeeHouse(int caffeineLimit) {
    log().debug("CoffeeHouse Open");
    this.caffeineLimit = caffeineLimit;

    receive(ReceiveBuilder.
            match(CreateGuest.class, createGuest -> {
                final ActorRef guest = createGuest(createGuest.favoriteCoffee, createGuest.caffeineLimit);
                addGuestToBookkeeper(guest);
                context().watch(guest);
            }).
            match(ApproveCoffee.class, this::coffeeApproved, approveCoffee ->
                    barista.forward(new Barista.PrepareCoffee(approveCoffee.coffee, approveCoffee.guest), context())
            ).
            match(ApproveCoffee.class, approveCoffee -> {
                log().info("Sorry, {}, but you have reached your limit.", approveCoffee.guest.path().name());
                context().stop(approveCoffee.guest);
            }).
            match(Terminated.class, terminated -> {
                log().info("Thanks, {}, for being our guest!", terminated.getActor());
                removeGuestFromBookkeeper(terminated.getActor());
            }).
            matchAny(this::unhandled).build()
    );
}
 

import akka.japi.pf.ReceiveBuilder; //导入依赖的package包/类
@Test
public void shouldRestartWaiterAndResendPrepareCoffeeToBaristaOnFailure() {
    new JavaTestKit(system) {{
        createActor(CoffeeHouse.class, "resend-prepare-coffee", () -> new CoffeeHouse(Integer.MAX_VALUE) {
            @Override
            protected ActorRef createBarista() {
                return getRef();
            }

            @Override
            protected ActorRef createWaiter() { //stubbing out the waiter actor to always throw exception
                return context().actorOf(Props.create(AbstractActor.class, () -> new AbstractActor() {{
                    receive(
                            ReceiveBuilder.matchAny(o -> {
                                throw new Waiter.FrustratedException(new Coffee.Akkaccino(), system.deadLetters());
                            }).build());
                }}), "waiter");
            }
        });
        ActorRef waiter = expectActor(this, "/user/resend-prepare-coffee/waiter");
        waiter.tell("Blow up", ActorRef.noSender());
        expectMsgEquals(new Barista.PrepareCoffee(new Coffee.Akkaccino(), system.deadLetters()));
    }};
}
 

import akka.japi.pf.ReceiveBuilder; //导入依赖的package包/类
public Waiter(ActorRef coffeeHouse, ActorRef barista, int maxComplaintCount) {
        this.coffeeHouse = coffeeHouse;
        this.barista = barista;
        this.maxComplaintCount = maxComplaintCount;

        receive(ReceiveBuilder.
                match(ServeCoffee.class, serveCoffee ->
                        this.coffeeHouse.tell(new CoffeeHouse.ApproveCoffee(serveCoffee.coffee, sender()), self())
                ).
                match(Barista.CoffeePrepared.class, coffeePrepared ->
                        coffeePrepared.guest.tell(new CoffeeServed(coffeePrepared.coffee), self())
                ).
//                match(Complaint.class, complaint -> complaintCount == this.maxComplaintCount, complaint -> {
//                    throw new FrustratedException(complaint.coffee, sender());
//                }).
                match(Complaint.class, complaint -> {
                    complaintCount++;
                    this.barista.tell(new Barista.PrepareCoffee(complaint.coffee, sender()), self());
                }).
                matchAny(this::unhandled).build()
        );
    }
 

import akka.japi.pf.ReceiveBuilder; //导入依赖的package包/类
public Guest(ActorRef waiter, Coffee favoriteCoffee, FiniteDuration finishCoffeeDuration) {
    this.waiter = waiter;
    this.favoriteCoffee = favoriteCoffee;
    this.finishCoffeeDuration = finishCoffeeDuration;
    orderFavoriteCoffee();

    receive(ReceiveBuilder.
            match(Waiter.CoffeeServed.class, coffeeServed -> {
                coffeeCount++;
                log().info("Enjoying my {} yummy {}!", coffeeCount, coffeeServed.coffee);
                scheduleCoffeeFinished();
            }).
            match(CoffeeFinished.class, coffeeFinished ->
                    orderFavoriteCoffee()
            ).
            matchAny(this::unhandled).build()
    );
}
 

import akka.japi.pf.ReceiveBuilder; //导入依赖的package包/类
public Guest(ActorRef waiter, Coffee favoriteCoffee, FiniteDuration finishCoffeeDuration, int caffeineLimit) {
    this.waiter = waiter;
    this.favoriteCoffee = favoriteCoffee;
    this.finishCoffeeDuration = finishCoffeeDuration;
    this.caffeineLimit = caffeineLimit;
    orderFavoriteCoffee();

    receive(ReceiveBuilder.
            match(Waiter.CoffeeServed.class, coffeeServed -> {
                coffeeCount++;
                log().info("Enjoying my {} yummy {}!", coffeeCount, coffeeServed.coffee);
                scheduleCoffeeFinished();
            }).
            match(CoffeeFinished.class, coffeeFinished -> coffeeCount > this.caffeineLimit, coffeeFinished -> {
                throw new CaffeineException();
            }).
            match(CoffeeFinished.class, coffeeFinished ->
                    orderFavoriteCoffee()
            ).
            matchAny(this::unhandled).build()
    );
}
 

import akka.japi.pf.ReceiveBuilder; //导入依赖的package包/类
public CoffeeHouse(int caffeineLimit) {
        log().debug("CoffeeHouse Open");
        this.caffeineLimit = caffeineLimit;

        receive(ReceiveBuilder.
                match(CreateGuest.class, createGuest -> {
                    final ActorRef guest = createGuest(createGuest.favoriteCoffee);
//                    final ActorRef guest = createGuest(createGuest.favoriteCoffee, createGuest.caffeineLimit);
                    addGuestToBookkeeper(guest);
                    context().watch(guest);
                }).
                match(ApproveCoffee.class, this::coffeeApproved, approveCoffee ->
                        barista.forward(new Barista.PrepareCoffee(approveCoffee.coffee, approveCoffee.guest), context())
                ).
                match(ApproveCoffee.class, approveCoffee -> {
                    log().info("Sorry, {}, but you have reached your limit.", approveCoffee.guest.path().name());
                    context().stop(approveCoffee.guest);
                }).
                match(Terminated.class, terminated -> {
                    log().info("Thanks, {}, for being our guest!", terminated.getActor());
                    removeGuestFromBookkeeper(terminated.getActor());
                }).
                matchAny(this::unhandled).build()
        );
    }
 

import akka.japi.pf.ReceiveBuilder; //导入依赖的package包/类
public CoffeeHouse(int caffeineLimit) {
        log().debug("CoffeeHouse Open");
        this.caffeineLimit = caffeineLimit;

        receive(ReceiveBuilder.
                match(CreateGuest.class, createGuest -> {
                    final ActorRef guest = createGuest(createGuest.favoriteCoffee);
                    addGuestToBookkeeper(guest);
//                    context().watch(guest);
                }).
                match(ApproveCoffee.class, this::coffeeApproved, approveCoffee ->
                        barista.forward(new Barista.PrepareCoffee(approveCoffee.coffee, approveCoffee.guest), context())
                ).
                match(ApproveCoffee.class, approveCoffee -> {
                    log().info("Sorry, {}, but you have reached your limit.", approveCoffee.guest.path().name());
                    context().stop(approveCoffee.guest);
                }).
//                match(Terminated.class, terminated -> {
//                    log().info("Thanks, {}, for being our guest!", terminated.getActor());
//                    removeGuestFromBookkeeper(terminated.getActor());
//                }).
                matchAny(this::unhandled).build()
        );
    }
 

import akka.japi.pf.ReceiveBuilder; //导入依赖的package包/类
public CoffeeHouse(int caffeineLimit) {
    log().debug("CoffeeHouse Open");
    this.caffeineLimit = caffeineLimit;

    receive(ReceiveBuilder.
            match(CreateGuest.class, createGuest -> {
                final ActorRef guest = createGuest(createGuest.favoriteCoffee);
                addGuestToBookkeeper(guest);
            }).
            match(ApproveCoffee.class, this::coffeeApproved, approveCoffee ->
                    barista.forward(new Barista.PrepareCoffee(approveCoffee.coffee, approveCoffee.guest), context())
            ).
            match(ApproveCoffee.class, approveCoffee -> {
                log().info("Sorry, {}, but you have reached your limit.", approveCoffee.guest.path().name());
                context().stop(approveCoffee.guest);
            }).
            matchAny(this::unhandled).build()
    );
}
 

import akka.japi.pf.ReceiveBuilder; //导入依赖的package包/类
public Guest(ActorRef waiter, Coffee favoriteCoffee, FiniteDuration finishCoffeeDuration, int caffeineLimit) {
        this.waiter = waiter;
        this.favoriteCoffee = favoriteCoffee;
        this.finishCoffeeDuration = finishCoffeeDuration;
        this.caffeineLimit = caffeineLimit;
        orderFavoriteCoffee();

        receive(ReceiveBuilder.
                match(Waiter.CoffeeServed.class, coffeeServed -> coffeeServed.coffee.equals(favoriteCoffee), coffeeServed -> {
                    coffeeCount++;
                    log().info("Enjoying my {} yummy {}!", coffeeCount, coffeeServed.coffee);
                    scheduleCoffeeFinished();
                }).
//                match(Waiter.CoffeeServed.class, coffeeServed -> {
//                    log().info("Expected a {}, but got a {}!", favoriteCoffee, coffeeServed.coffee);
//                    waiter.tell(new Waiter.Complaint(favoriteCoffee), self());
//                }).
                match(CoffeeFinished.class, coffeeFinished -> coffeeCount > this.caffeineLimit, coffeeFinished -> {
                    throw new CaffeineException();
                }).
                match(CoffeeFinished.class, coffeeFinished ->
                        orderFavoriteCoffee()
                ).
                matchAny(this::unhandled).build()
        );
    }
 

import akka.japi.pf.ReceiveBuilder; //导入依赖的package包/类
public Waiter(ActorRef coffeeHouse, ActorRef barista) {
        this.coffeeHouse = coffeeHouse;
        this.barista = barista;
//        this.maxComplaintCount = maxComplaintCount;

        receive(ReceiveBuilder.
                match(ServeCoffee.class, serveCoffee ->
                        this.coffeeHouse.tell(new CoffeeHouse.ApproveCoffee(serveCoffee.coffee, sender()), self())
                ).
                match(Barista.CoffeePrepared.class, coffeePrepared ->
                        coffeePrepared.guest.tell(new CoffeeServed(coffeePrepared.coffee), self())
                ).
//                match(Complaint.class, complaint -> complaintCount == this.maxComplaintCount, complaint -> {
//                    throw new FrustratedException();
//                }).
                match(Complaint.class, complaint -> {
                    complaintCount++;
                    this.barista.tell(new Barista.PrepareCoffee(complaint.coffee, sender()), self());
                }).
                matchAny(this::unhandled).build()
        );
    }
 

import akka.japi.pf.ReceiveBuilder; //导入依赖的package包/类
public ParseSupervisorActor() {
  receive(
      ReceiveBuilder.match(ParseMessageCreateCommand.class, this::handleParseMessageCreateCommand)
          .match(ParseMessageCreatedEvent.class, this::handleParseMessageCreatedEvent)
          .match(ParsedMessageEvent.class, this::handleParsedMessageEvent)
          .match(WorkflowFinishedEvent.class, this::handleWorkflowFinishedEvent)
          .matchAny(o -> LOG.warn("Unhandled message [{}]", o)).build());
}
 

import akka.japi.pf.ReceiveBuilder; //导入依赖的package包/类
public ParseMessageActor() {
  receive(ReceiveBuilder
      .match(ParseMessageCreateCommand.class,
          cmd -> getDBPersistenceActor().tell(cmd, getContext().parent()))
      .match(ParseMessageCommand.class, this::handleParseMessageCommand)
      .match(ConfigRetrievedEvent.class, this::handleConfigRetrievedEvent)
      .match(NotifyProcessEngineCommand.class, this::handleNotifyProcessEngineCommand)
      .match(NotifyConfigRetrievedEvent.class, this::handleNotifyConfigRetrievedEvent)
      .matchAny(o -> LOG.warn("Unhandled message [{}]", o)).build());
}
 

import akka.japi.pf.ReceiveBuilder; //导入依赖的package包/类
public ParsePersistenceActor() {
  receive(
      ReceiveBuilder.match(ParseMessageCreateCommand.class, this::handleParseMessageCreateCommand)
          .match(ConfigRetrievalCommand.class, this::handleConfigRetrievalCommand)
          .match(StoreInternalDataCommand.class, this::handleStoreInternalDataCommand)
          .match(NotifyProcessEngineCommand.class, this::handleNotifyProcessEngineCommand)
          .match(UpdateMessageStateCommand.class, this::handleUpdateMessageStateCommand)
          .matchAny(o -> LOG.warn("Unhandled message [{}]", o)).build());
}
 

import akka.japi.pf.ReceiveBuilder; //导入依赖的package包/类
@Override
public Receive createReceive() {
  return ReceiveBuilder.create()
          .match(
                  PaxosAPI.Propose.class,
                  propose -> propose.txid() == txid,
                  propose -> {
                    LOG.info("Received proposal={}", propose);
                    this.proposal = propose.value();
                    broadcastNextBallot(1);
                  }
          )
          .build();
}
 

import akka.japi.pf.ReceiveBuilder; //导入依赖的package包/类
private Receive nextBallotSent() {
  return ReceiveBuilder.create()
          .match(AlreadySucceed.class, this::onAlreadySucceed)
          .match(LastVote.class,
                  lastVote -> lastVote.ballotNumber() == lastTried && lastVote.txid() == txid,
                  lastVote -> {
                    LOG.info("Received last vote from {}", lastVote);
                    lastVotes.put(sender(), lastVote);

                    if (lastVotes.size() > priests.size() / 2) {
                      LOG.info("There is a quorum");
                      final LastVote winner = Collections.max(lastVotes.values());

                      final Object lockedValue;

                      if (winner.vote() == SpecialValues.BLANK) {
                        lockedValue = proposal;
                      } else if (winner.vote() == SpecialValues.OUTDATED_BALLOT_NUMBER) {
                        LOG.info("Seems that I have outdated ballot number");
                        broadcastNextBallot(winner.ballotNumber() + 1);
                        return;
                      } else {
                        lockedValue = winner.vote();
                      }

                      priests.forEach(p -> p.tell(new BeginBallot(txid, lastTried, lockedValue), self()));
                      getContext().become(beginBallotSent());
                    }
                  })
          .build();
}
 

import akka.japi.pf.ReceiveBuilder; //导入依赖的package包/类
private Receive beginBallotSent() {
  return ReceiveBuilder.create()
          .match(AlreadySucceed.class, this::onAlreadySucceed)
          .match(Voted.class,
                  voted -> voted.ballotNumber() == lastTried && voted.txid() == txid,
                  voted -> {
                    votes.add(voted);

                    if (votes.size() > priests.size() / 2) {
                      priests.forEach(p -> p.tell(new Success(txid, voted.vote()), self()));
                      context().stop(self());
                    }
                  })
          .build();
}
 

import akka.japi.pf.ReceiveBuilder; //导入依赖的package包/类
@Override
public Receive createReceive() {
  if (decision == null) {
    return ReceiveBuilder.create()
            .match(NextBallot.class, nextBallot -> {
              if (nextBallot.ballotNumber() > lastBallot) {
                lastBallot = nextBallot.ballotNumber();
                sender().tell(new LastVote(txid, lastBallot, lastVote), self());
              } else {
                sender().tell(new LastVote(txid, lastBallot, SpecialValues.OUTDATED_BALLOT_NUMBER), self());
                LOG.warning(
                        "NextBallot: Proposer has outdated ballotNumber number proposer={}, ballotNumber={}, currentBallot={}",
                        sender(), nextBallot.ballotNumber(), lastBallot
                );
              }
            })
            .match(BeginBallot.class, beginBallot -> {
              //FIXME: equal ballotNumber numbers may cause collisions, but they shouldn't pass NextBallot
              if (beginBallot.ballotNumber() == lastBallot) {
                lastVote = beginBallot.decree();
                sender().tell(new Voted(txid, beginBallot.ballotNumber(), beginBallot.decree()), self());
              } else {
                sender().tell(new Voted(txid, lastBallot, SpecialValues.OUTDATED_BALLOT_NUMBER), self());
                LOG.warning(
                        "BeginBallot: Proposer has outdated ballotNumber number proposer={}, ballotNumber={}, currentBallot={}",
                        sender(), beginBallot.ballotNumber(), lastBallot
                );
              }
            })
            .match(Success.class, success -> {
              LOG.info("Learned {} for txid={}", success.decree(), txid);
              decision = success.decree();
              subscriber.tell(new PaxosAPI.Decide(decision, txid), self());
              getContext().become(success());
            })
            .build();
  } else {
    return success();
  }
}
 

import akka.japi.pf.ReceiveBuilder; //导入依赖的package包/类
@Override
public Receive createReceive() {
  return ReceiveBuilder.create()
          .match(PaxosMessage.class, this::onPaxosMessage)
          .match(PaxosAPI.Decide.class, this::onDecide)
          .match(Broadcast.class, this::onBroadcast)
          .build();
}