import org.pcollections.TreePVector; //导入依赖的package包/类
@Test
public void shouldIncludeSomeOldChirpsInLiveFeed() throws Exception {
    ChirpService chirpService = server.client(ChirpService.class);

    Chirp chirp1 = new Chirp("usr3", "hi 1");
    chirpService.addChirp("usr3").invoke(chirp1).toCompletableFuture().get(3, SECONDS);

    Chirp chirp2 = new Chirp("usr4", "hi 2");
    chirpService.addChirp("usr4").invoke(chirp2).toCompletableFuture().get(3, SECONDS);

    LiveChirpsRequest request = new LiveChirpsRequest(TreePVector.<String>empty().plus("usr3").plus("usr4"));

    eventually(FiniteDuration.create(10, SECONDS), () -> {
        Source<Chirp, ?> chirps = chirpService.getLiveChirps().invoke(request).toCompletableFuture().get(3, SECONDS);
        Probe<Chirp> probe = chirps.runWith(TestSink.probe(server.system()), server.materializer());
        probe.request(10);
        probe.expectNextUnordered(chirp1, chirp2);

        Chirp chirp3 = new Chirp("usr4", "hi 3");
        chirpService.addChirp("usr4").invoke(chirp3).toCompletableFuture().get(3, SECONDS);
        probe.expectNext(chirp3);

        probe.cancel();
    });
}
 

import org.pcollections.TreePVector; //导入依赖的package包/类
@Test
public void shouldRetrieveOldChirps() throws Exception {
    ChirpService chirpService = server.client(ChirpService.class);

    Chirp chirp1 = new Chirp("usr5", "msg 1");
    chirpService.addChirp("usr5").invoke(chirp1).toCompletableFuture().get(3, SECONDS);

    Chirp chirp2 = new Chirp("usr6", "msg 2");
    chirpService.addChirp("usr6").invoke(chirp2).toCompletableFuture().get(3, SECONDS);

    HistoricalChirpsRequest request = new HistoricalChirpsRequest(Instant.now().minusSeconds(20),
            TreePVector.<String>empty().plus("usr5").plus("usr6"));

    eventually(FiniteDuration.create(10, SECONDS), () -> {
        Source<Chirp, ?> chirps = chirpService.getHistoricalChirps().invoke(request).toCompletableFuture().get(3, SECONDS);
        Probe<Chirp> probe = chirps.runWith(TestSink.probe(server.system()), server.materializer());
        probe.request(10);
        probe.expectNextUnordered(chirp1, chirp2);
        probe.expectComplete();
    });
}
 

import org.pcollections.TreePVector; //导入依赖的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 org.pcollections.TreePVector; //导入依赖的package包/类
@Test
public void shouldIncludeSomeOldChirpsInLiveFeed() throws Exception {
  ChirpService chirpService = server.client(ChirpService.class);

  Chirp chirp1 = Chirp.of("usr3", "hi 1");
  chirpService.addChirp("usr3").invoke(chirp1).toCompletableFuture().get(3, SECONDS);

  Chirp chirp2 = Chirp.of("usr4", "hi 2");
  chirpService.addChirp("usr4").invoke(chirp2).toCompletableFuture().get(3, SECONDS);

  LiveChirpsRequest request = LiveChirpsRequest.of(TreePVector.<String>empty().plus("usr3").plus("usr4"));
  Source<Chirp, ?> chirps = chirpService.getLiveChirps("user3").invoke(request).toCompletableFuture().get(3, SECONDS);
  Probe<Chirp> probe = chirps.runWith(TestSink.probe(server.system()), server.materializer());
  probe.request(10);
  probe.expectNextUnordered(chirp1, chirp2);

  Chirp chirp3 = Chirp.of("usr4", "hi 3");
  chirpService.addChirp("usr4").invoke(chirp3).toCompletableFuture().get(3, SECONDS);
  probe.expectNext(chirp3);

  probe.cancel();
}
 

import org.pcollections.TreePVector; //导入依赖的package包/类
@Test
public void shouldRetrieveOldChirps() throws Exception {
  ChirpService chirpService = server.client(ChirpService.class);

  Chirp chirp1 = Chirp.of("usr5", "msg 1");
  chirpService.addChirp("usr5").invoke(chirp1).toCompletableFuture().get(3, SECONDS);

  Chirp chirp2 = Chirp.of("usr6", "msg 2");
  chirpService.addChirp("usr6").invoke(chirp2).toCompletableFuture().get(3, SECONDS);

  HistoricalChirpsRequest request = HistoricalChirpsRequest.of(Instant.now().minusSeconds(20),
      TreePVector.<String>empty().plus("usr5").plus("usr6"));
  Source<Chirp, ?> chirps = chirpService.getHistoricalChirps().invoke(request).toCompletableFuture().get(3, SECONDS);
  Probe<Chirp> probe = chirps.runWith(TestSink.probe(server.system()), server.materializer());
  probe.request(10);
  probe.expectNextUnordered(chirp1, chirp2);
  probe.expectComplete();
}
 

import org.pcollections.TreePVector; //导入依赖的package包/类
@Override
public PVector<V> applyUnsafe(JsonParser p) throws IOException {
    List<V> values = new LinkedList<>();

    /*
     * If we are in the root context (not inside an object or list) then we need to consume the next token
     * before attempting to call child deserialisers.
     */
    if (p.getParsingContext().inRoot()) {
        if (p.nextToken() == JsonToken.END_ARRAY) {
            return TreePVector.empty();
        }
    }

    if (JsonToken.VALUE_NULL != p.getCurrentToken()) {
        /*
         * When the parser has hit the end of input nextToken() will always return null.
         * So need to prevent infinite loops we check the parser closed flag.
         */
        while (!p.isClosed() && (p.nextToken() != JsonToken.END_ARRAY)) {
            values.add(itemDeserialiser.apply(p));
        }
    }
    return TreePVector.from(values);
}
 

import org.pcollections.TreePVector; //导入依赖的package包/类
public BasketState removeItem(UUID itemId, int removedCount) {
    int totalCount = items.stream().filter(dbi-> dbi.getId().equals(itemId)).mapToInt(DomainBasketItem::getNumberOfItems).sum();
    PSequence<DomainBasketItem> newItems =
            TreePVector.from(
                    items.stream().filter(dbi-> !dbi.getId().equals(itemId)).collect(Collectors.toList())
            ).plus(new DomainBasketItem(itemId, totalCount-removedCount));
    return new BasketState(
            newItems,
            alreadyOrdered);
}
 

import org.pcollections.TreePVector; //导入依赖的package包/类
@Test
public void addItems() {
    UUID bid = UUID.randomUUID();
    UUID orangeId = UUID.randomUUID();
    UUID bananaId = UUID.randomUUID();


    BasketCommand.AddItem cmd1 = new BasketCommand.AddItem(orangeId, 3);
    BasketCommand.AddItem cmd2 = new BasketCommand.AddItem(bananaId, 1);
    BasketCommand.GetBasket cmd3 = BasketCommand.GetBasket.INSTANCE;

    PersistentEntityTestDriver<BasketCommand, PEBasketEvent, BasketState> driver =
            new PersistentEntityTestDriver<>(system, new BasketEntity(), bid.toString());


    // add some oranges
    Outcome<PEBasketEvent, BasketState> outcome1 = driver.run(cmd1);
    assertEquals(Done.getInstance(), outcome1.getReplies().get(0));
    assertEquals(Collections.emptyList(), outcome1.issues());

    // and then add some bananas. Finally get the basket.
    Outcome<PEBasketEvent, BasketState> outcome3 = driver.run(cmd2, cmd3);

    PSequence<DomainBasketItem> items = TreePVector.<DomainBasketItem>empty()
            .plus(new DomainBasketItem(orangeId, 3))
            .plus(new DomainBasketItem(bananaId, 1));
    BasketState expected = new BasketState(items, false);
    assertEquals(expected, outcome3.getReplies().get(1));
    assertEquals(Collections.emptyList(), outcome1.issues());

}
 

import org.pcollections.TreePVector; //导入依赖的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 org.pcollections.TreePVector; //导入依赖的package包/类
@Test
public void shouldBeAbleToCreateUsersAndConnectFriends() throws Exception {
  withServer(defaultSetup().withCassandra(true), server -> {
    FriendService friendService = server.client(FriendService.class);
    User usr1 = new User("usr1", "User 1");
    friendService.createUser().invoke(usr1).toCompletableFuture().get(10, SECONDS);
    User usr2 = new User("usr2", "User 2");
    friendService.createUser().invoke(usr2).toCompletableFuture().get(3, SECONDS);
    User usr3 = new User("usr3", "User 3");
    friendService.createUser().invoke(usr3).toCompletableFuture().get(3, SECONDS);

    friendService.addFriend("usr1").invoke(new FriendId(usr2.userId)).toCompletableFuture().get(3, SECONDS);
    friendService.addFriend("usr1").invoke(new FriendId(usr3.userId)).toCompletableFuture().get(3, SECONDS);

    User fetchedUsr1 = friendService.getUser("usr1").invoke().toCompletableFuture().get(3,
        SECONDS);
    assertEquals(usr1.userId, fetchedUsr1.userId);
    assertEquals(usr1.name, fetchedUsr1.name);
    assertEquals(TreePVector.empty().plus("usr2").plus("usr3"), fetchedUsr1.friends);

    eventually(FiniteDuration.create(10, SECONDS), () -> {
      PSequence<String> followers = friendService.getFollowers("usr2").invoke()
          .toCompletableFuture().get(3, SECONDS);
      assertEquals(TreePVector.empty().plus("usr1"), followers);
    });

  });
}
 

import org.pcollections.TreePVector; //导入依赖的package包/类
@Test
public void testCreateUserWithInitialFriends() {
  PersistentEntityTestDriver<FriendCommand, FriendEvent, FriendState> driver = new PersistentEntityTestDriver<>(
      system, new FriendEntity(), "user-1");

  TreePVector<String> friends = TreePVector.<String>empty().plus("bob").plus("peter");
  Outcome<FriendEvent, FriendState> outcome = driver.run(
      new CreateUser(new User("alice", "Alice", Optional.of(friends))));
  assertEquals(Done.getInstance(), outcome.getReplies().get(0));
  assertEquals("alice", ((UserCreated) outcome.events().get(0)).userId);
  assertEquals("bob", ((FriendAdded) outcome.events().get(1)).friendId);
  assertEquals("peter", ((FriendAdded) outcome.events().get(2)).friendId);
  assertEquals(Collections.emptyList(), driver.getAllIssues());
}
 

import org.pcollections.TreePVector; //导入依赖的package包/类
public CompletionStage<PSequence<Chirp>> getRecentChirps(PSequence<String> userIds) {
    CompletionStage<PSequence<Chirp>> results = CompletableFuture.completedFuture(TreePVector.empty());
    for (String userId : userIds) {
        results = results.thenCombine(getRecentChirps(userId), PSequence::plusAll);
    }

    return results.thenApply(this::limitRecentChirps);
}
 

import org.pcollections.TreePVector; //导入依赖的package包/类
private PSequence<Chirp> limitRecentChirps(PSequence<Chirp> all) {
    List<Chirp> limited = all.stream()
            .sorted(comparing((Chirp chirp) -> chirp.timestamp).reversed())
            .limit(NUM_RECENT_CHIRPS)
            .collect(Collectors.toCollection(ArrayList::new));

    Collections.reverse(limited);
    return TreePVector.from(limited);
}
 

import org.pcollections.TreePVector; //导入依赖的package包/类
@Test
public void shouldPublishShirpsToSubscribers() throws Exception {
    ChirpService chirpService = server.client(ChirpService.class);
    LiveChirpsRequest request = new LiveChirpsRequest(TreePVector.<String>empty().plus("usr1").plus("usr2"));
    Source<Chirp, ?> chirps1 = chirpService.getLiveChirps().invoke(request).toCompletableFuture().get(3, SECONDS);
    Probe<Chirp> probe1 = chirps1.runWith(TestSink.probe(server.system()), server.materializer());
    probe1.request(10);
    Source<Chirp, ?> chirps2 = chirpService.getLiveChirps().invoke(request).toCompletableFuture().get(3, SECONDS);
    Probe<Chirp> probe2 = chirps2.runWith(TestSink.probe(server.system()), server.materializer());
    probe2.request(10);

    Chirp chirp1 = new Chirp("usr1", "hello 1");
    chirpService.addChirp("usr1").invoke(chirp1).toCompletableFuture().get(3, SECONDS);
    probe1.expectNext(chirp1);
    probe2.expectNext(chirp1);

    Chirp chirp2 = new Chirp("usr1", "hello 2");
    chirpService.addChirp("usr1").invoke(chirp2).toCompletableFuture().get(3, SECONDS);
    probe1.expectNext(chirp2);
    probe2.expectNext(chirp2);

    Chirp chirp3 = new Chirp("usr2", "hello 3");
    chirpService.addChirp("usr2").invoke(chirp3).toCompletableFuture().get(3, SECONDS);
    probe1.expectNext(chirp3);
    probe2.expectNext(chirp3);

    probe1.cancel();
    probe2.cancel();
}
 

import org.pcollections.TreePVector; //导入依赖的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 org.pcollections.TreePVector; //导入依赖的package包/类
@Test
public void shouldBeAbleToCreateUsersAndConnectFriends() throws Exception {
  withServer(defaultSetup(), server -> {
    FriendService friendService = server.client(FriendService.class);
    User usr1 = User.of("usr1", "User 1");
    friendService.createUser().invoke(usr1).toCompletableFuture().get(10, SECONDS);
    User usr2 = User.of("usr2", "User 2");
    friendService.createUser().invoke(usr2).toCompletableFuture().get(3, SECONDS);
    User usr3 = User.of("usr3", "User 3");
    friendService.createUser().invoke(usr3).toCompletableFuture().get(3, SECONDS);

    friendService.addFriend("usr1").invoke(FriendId.of(usr2.getUserId())).toCompletableFuture().get(3, SECONDS);
    friendService.addFriend("usr1").invoke(FriendId.of(usr3.getUserId())).toCompletableFuture().get(3, SECONDS);

    User fetchedUsr1 = friendService.getUser("usr1").invoke().toCompletableFuture().get(3,
        SECONDS);
    assertEquals(usr1.getUserId(), fetchedUsr1.getUserId());
    assertEquals(usr1.getName(), fetchedUsr1.getName());
    assertEquals(TreePVector.empty().plus("usr2").plus("usr3"), fetchedUsr1.getFriends());

    eventually(FiniteDuration.create(10, SECONDS), () -> {
      PSequence<String> followers = friendService.getFollowers("usr2").invoke()
          .toCompletableFuture().get(3, SECONDS);
      assertEquals(TreePVector.empty().plus("usr1"), followers);
    });

  });
}
 

import org.pcollections.TreePVector; //导入依赖的package包/类
@Test
public void testCreateUserWithInitialFriends() {
  PersistentEntityTestDriver<FriendCommand, FriendEvent, FriendState> driver = new PersistentEntityTestDriver<>(
      system, new FriendEntity(), "user-1");

  TreePVector<String> friends = TreePVector.<String>empty().plus("bob").plus("peter");
  Outcome<FriendEvent, FriendState> outcome = driver.run(
      CreateUser.of(AbstractUser.of("alice", "Alice", Optional.of(friends))));
  assertEquals(Done.getInstance(), outcome.getReplies().get(0));
  assertEquals("alice", ((UserCreated) outcome.events().get(0)).getUserId());
  assertEquals("bob", ((FriendAdded) outcome.events().get(1)).getFriendId());
  assertEquals("peter", ((FriendAdded) outcome.events().get(2)).getFriendId());
  assertEquals(Collections.emptyList(), driver.getAllIssues());
}
 

import org.pcollections.TreePVector; //导入依赖的package包/类
private CompletionStage<PSequence<Chirp>> recentChirps(PSequence<String> userIds) {
  int limit = 10;
  PSequence<CompletionStage<PSequence<Chirp>>> results = TreePVector.empty();
  for (String userId : userIds) {
    CompletionStage<PSequence<Chirp>> result = db
        .selectAll("SELECT * FROM chirp WHERE userId = ? ORDER BY timestamp DESC LIMIT ?", userId, limit)
        .thenApply(rows -> {
          List<Chirp> chirps = rows.stream().map(this::mapChirp).collect(Collectors.toList());
          return TreePVector.from(chirps);
        });
    results = results.plus(result);
  }

  CompletionStage<PSequence<Chirp>> combined = null;
  for (CompletionStage<PSequence<Chirp>> chirpsFromOneUser : results) {
    if (combined == null) {
      combined = chirpsFromOneUser;
    } else {
      combined = combined.thenCombine(chirpsFromOneUser, (a, b) -> a.plusAll(b));
    }
  }

  CompletionStage<PSequence<Chirp>> sortedLimited = combined.thenApply(all -> {
    List<Chirp> allSorted = new ArrayList<>(all);
    // reverse order
    Collections.sort(allSorted, (a, b) -> b.getTimestamp().compareTo(a.getTimestamp()));
    List<Chirp> limited = allSorted.stream().limit(limit).collect(Collectors.toList());
    List<Chirp> reversed = new ArrayList<>(limited);
    Collections.reverse(reversed);
    return TreePVector.from(reversed);
  });

  return sortedLimited;
}
 

import org.pcollections.TreePVector; //导入依赖的package包/类
@Test
public void shouldPublishShirpsToSubscribers() throws Exception {
  ChirpService chirpService = server.client(ChirpService.class);
  LiveChirpsRequest request = LiveChirpsRequest.of(TreePVector.<String>empty().plus("usr1").plus("usr2"));
  Source<Chirp, ?> chirps1 = chirpService.getLiveChirps("user1").invoke(request).toCompletableFuture().get(3, SECONDS);
  Probe<Chirp> probe1 = chirps1.runWith(TestSink.probe(server.system()), server.materializer());
  probe1.request(10);
  Source<Chirp, ?> chirps2 = chirpService.getLiveChirps("user1").invoke(request).toCompletableFuture().get(3, SECONDS);
  Probe<Chirp> probe2 = chirps2.runWith(TestSink.probe(server.system()), server.materializer());
  probe2.request(10);

  Chirp chirp1 = Chirp.of("usr1", "hello 1");
  chirpService.addChirp("usr1").invoke(chirp1).toCompletableFuture().get(3, SECONDS);
  probe1.expectNext(chirp1);
  probe2.expectNext(chirp1);

  Chirp chirp2 = Chirp.of("usr1", "hello 2");
  chirpService.addChirp("usr1").invoke(chirp2).toCompletableFuture().get(3, SECONDS);
  probe1.expectNext(chirp2);
  probe2.expectNext(chirp2);

  Chirp chirp3 = Chirp.of("usr2", "hello 3");
  chirpService.addChirp("usr2").invoke(chirp3).toCompletableFuture().get(3, SECONDS);
  probe1.expectNext(chirp3);
  probe2.expectNext(chirp3);

  probe1.cancel();
  probe2.cancel();
}
 

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