import com.lightbend.lagom.javadsl.api.ServiceCall; //导入依赖的package包/类
@Override
public ServiceCall<LiveChirpsRequest, Source<Chirp, ?>> getLiveChirps() {
    return req -> chirps.getRecentChirps(req.userIds).thenApply(recentChirps -> {
        List<Source<Chirp, ?>> sources = new ArrayList<>();
        for (String userId : req.userIds) {
            sources.add(topic.subscriber(userId));
        }
        HashSet<String> users = new HashSet<>(req.userIds);
        Source<Chirp, ?> publishedChirps = Source.from(sources).flatMapMerge(sources.size(), s -> s)
                .filter(c -> users.contains(c.userId));

        // We currently ignore the fact that it is possible to get duplicate chirps
        // from the recent and the topic. That can be solved with a de-duplication stage.
        return Source.from(recentChirps).concat(publishedChirps);
    });
}
 

import com.lightbend.lagom.javadsl.api.ServiceCall; //导入依赖的package包/类
/**
* Get all registered Cargo
*
* @return
*/
   @Override
   public ServiceCall<NotUsed, NotUsed, PSequence<Cargo>> getAllRegistrations() {
       return (userId, req) -> {
           CompletionStage<PSequence<Cargo>> result = db.selectAll("SELECT cargoid, name, description, owner, destination FROM cargo")
                   .thenApply(rows -> {
                       List<Cargo> cargos = rows.stream().map(row -> Cargo.of(row.getString("cargoid"),
                               row.getString("name"),
                               row.getString("description"),
                               row.getString("owner"),
                               row.getString("destination"))).collect(Collectors.toList());
                       return TreePVector.from(cargos);
                   });
           return result;
       };
   }
 

import com.lightbend.lagom.javadsl.api.ServiceCall; //导入依赖的package包/类
@Override
public ServiceCall<HistoricalChirpsRequest, Source<Chirp, ?>> getHistoricalChirps() {
  return req -> {
    List<Source<Chirp, ?>> sources = new ArrayList<>();
    for (String uid : req.getUserIds()) {
        Source<Chirp, NotUsed> select = db
          .select("SELECT * FROM chirp WHERE userId = ? AND timestamp >= ? ORDER BY timestamp ASC", uid,
              req.getFromTime().toEpochMilli())
          .map(this::mapChirp);
      sources.add(select);
    }
      // Chirps from one user are ordered by timestamp, but chirps from different
      // users are not ordered. That can be improved by implementing a smarter
      // merge that takes the timestamps into account.
    Source<Chirp, ?> result = Source.from(sources).flatMapMerge(sources.size(), s -> s);
    return CompletableFuture.completedFuture(result);
  };
}
 

import com.lightbend.lagom.javadsl.api.ServiceCall; //导入依赖的package包/类
@Override
public ServiceCall<Shipment, Done> provisionItems() {
    return shipment ->
            refFor(shipment.getId())
                    .ask(new IncreaseInventory(
                            shipment.getName(),
                            shipment.getId(),
                            shipment.getCount()))
                    .thenApply(ignored -> Done.getInstance())
            ;
}
 

import com.lightbend.lagom.javadsl.api.ServiceCall; //导入依赖的package包/类
public ServiceCall<NotUsed, String> hello(String id) {
  return request -> {
    // Look up the hello world entity for the given ID.
    PersistentEntityRef<HelloCommand> ref = persistentEntityRegistry.refFor(HelloEntity.class, id);
    // Ask the entity the Hello command.
    return ref.ask(new Hello(id, Optional.empty()));
  };
}
 

import com.lightbend.lagom.javadsl.api.ServiceCall; //导入依赖的package包/类
public ServiceCall<GreetingMessage, Done> useGreeting(String id) {
  return request -> {
    // Look up the hello world entity for the given ID.
    PersistentEntityRef<HelloCommand> ref = persistentEntityRegistry.refFor(HelloEntity.class, id);
    // Tell the entity to use the greeting name specified.
    return ref.ask(new UseGreetingMessage(request.message));
  };

}
 

import com.lightbend.lagom.javadsl.api.ServiceCall; //导入依赖的package包/类
@Override
public ServiceCall<NotUsed, User> getUser(String userId) {
  return request -> {
    return friendEntityRef(userId).ask(new GetUser()).thenApply(reply -> {
      if (reply.user.isPresent())
        return reply.user.get();
      else
        throw new NotFound("user " + userId + " not found");
    });
  };
}
 

import com.lightbend.lagom.javadsl.api.ServiceCall; //导入依赖的package包/类
@Override
public ServiceCall<User, NotUsed> createUser() {
  return request -> {
    return friendEntityRef(request.userId).ask(new CreateUser(request))
        .thenApply(ack -> NotUsed.getInstance());
  };
}
 

import com.lightbend.lagom.javadsl.api.ServiceCall; //导入依赖的package包/类
@Override
public ServiceCall<FriendId, NotUsed> addFriend(String userId) {
  return request -> {
    return friendEntityRef(userId).ask(new AddFriend(request.friendId))
        .thenApply(ack -> NotUsed.getInstance());
  };
}
 

import com.lightbend.lagom.javadsl.api.ServiceCall; //导入依赖的package包/类
@Override
public ServiceCall<NotUsed, PSequence<String>> getFollowers(String userId) {
  return req -> {
    CompletionStage<PSequence<String>> result = db.selectAll("SELECT * FROM follower WHERE userId = ?", userId)
      .thenApply(rows -> {
      List<String> followers = rows.stream().map(row -> row.getString("followedBy")).collect(Collectors.toList());
      return TreePVector.from(followers);
    });
    return result;
  };
}
 

import com.lightbend.lagom.javadsl.api.ServiceCall; //导入依赖的package包/类
@Override
public ServiceCall<TestParams, NotUsed> startLoadHeadless() {
  return params -> {
    load(params).runWith(Sink.ignore(), materializer);
    return CompletableFuture.completedFuture(NotUsed.getInstance());
  };
}
 

import com.lightbend.lagom.javadsl.api.ServiceCall; //导入依赖的package包/类
@Override
public ServiceCall<Chirp, NotUsed> addChirp(String userId) {
    return chirp -> {
        if (!userId.equals(chirp.userId))
            throw new IllegalArgumentException("UserId " + userId + " did not match userId in " + chirp);

        return persistentEntities.refFor(ChirpTimelineEntity.class, userId)
                .ask(new AddChirp(chirp))
                .thenApply(done -> NotUsed.getInstance());
    };
}
 

import com.lightbend.lagom.javadsl.api.ServiceCall; //导入依赖的package包/类
@Override
public ServiceCall<HistoricalChirpsRequest, Source<Chirp, ?>> getHistoricalChirps() {
    return req -> {
        PSequence<String> userIds = req.userIds;
        long timestamp = req.fromTime.toEpochMilli();
        Source<Chirp, ?> result = chirps.getHistoricalChirps(userIds, timestamp);
        return CompletableFuture.completedFuture(result);
    };
}
 

import com.lightbend.lagom.javadsl.api.ServiceCall; //导入依赖的package包/类
@Override
public ServiceCall<NotUsed, Source<Chirp, ?>> getLiveActivityStream(String userId) {
  return req -> {
    return friendService.getUser(userId).invoke().thenCompose(user -> {
      PSequence<String> userIds = user.friends.plus(userId);
      LiveChirpsRequest chirpsReq =  new LiveChirpsRequest(userIds);
      // Note that this stream will not include changes to friend associates,
      // e.g. adding a new friend.
      CompletionStage<Source<Chirp, ?>> result = chirpService.getLiveChirps().invoke(chirpsReq);
      return result;
    });
  };
}
 

import com.lightbend.lagom.javadsl.api.ServiceCall; //导入依赖的package包/类
@Override
public ServiceCall<NotUsed, Source<Chirp, ?>> getHistoricalActivityStream(String userId) {
  return req ->
    friendService.getUser(userId).invoke().thenCompose(user -> {
      PSequence<String> userIds = user.friends.plus(userId);
      // FIXME we should use HistoricalActivityStreamReq request parameter
      Instant fromTime = Instant.now().minus(Duration.ofDays(7));
      HistoricalChirpsRequest chirpsReq = new HistoricalChirpsRequest(fromTime, userIds);
      CompletionStage<Source<Chirp, ?>> result = chirpService.getHistoricalChirps().invoke(chirpsReq);
      return result;
    });
}
 

import com.lightbend.lagom.javadsl.api.ServiceCall; //导入依赖的package包/类
@Override
public ServiceCall<NotUsed, User> getUser(String userId) {
  return req -> {
    if (userId.equals(usr1.userId))
      return completedFuture(usr1);
    else if (userId.equals(usr2.userId))
      return completedFuture(usr2);
    else
      throw new NotFound(userId);
  };
}
 

import com.lightbend.lagom.javadsl.api.ServiceCall; //导入依赖的package包/类
@Override
public ServiceCall<NotUsed, PSequence<String>> getFollowers(String userId) {
  return req -> {
    if (userId.equals(usr1.userId))
      return completedFuture(TreePVector.<String>empty());
    else if (userId.equals(usr2.userId))
      return completedFuture(TreePVector.<String>empty().plus("usr1"));
    else
      throw new NotFound(userId);
  };
}
 

import com.lightbend.lagom.javadsl.api.ServiceCall; //导入依赖的package包/类
@Override
public ServiceCall<LiveChirpsRequest, Source<Chirp, ?>> getLiveChirps() {
  return req -> {
    if (req.userIds.contains("usr2")) {
      Chirp c1 = new Chirp("usr2", "msg1");
      Chirp c2 = new Chirp("usr2", "msg2");
      return completedFuture(Source.from(Arrays.asList(c1, c2)));
    } else
      return completedFuture(Source.empty());
  };
}
 

import com.lightbend.lagom.javadsl.api.ServiceCall; //导入依赖的package包/类
@Override
public ServiceCall<HistoricalChirpsRequest, Source<Chirp, ?>> getHistoricalChirps() {
  return req -> {
    if (req.userIds.contains("usr2")) {
      Chirp c1 = new Chirp("usr2", "msg1");
      return completedFuture(Source.single(c1));
    } else
      return completedFuture(Source.empty());
  };
}
 

import com.lightbend.lagom.javadsl.api.ServiceCall; //导入依赖的package包/类
@Override
public ServiceCall<String, Leg, Done> addLeg() {
    return (id, request) -> {
        CompletionStage<Cargo> response = registrationService.getRegistration().invoke(request.getCargoId(), NotUsed.getInstance());
        PersistentEntityRef<ShippingCommand> itinerary = persistentEntityRegistry.refFor(ItineraryEntity.class, id);
        return itinerary.ask(AddLeg.of(request));
    };
}