Getting Started the Stream API

String[] arr = new String[]{"a", "b", "c"};
// String[] arr = Arrays.asList("a", "b", "c");
Stream<String> stream = Arrays.stream(arr);

List<String> list = Arrays.asList("a", "b", "c");
Stream<String> stream1 = list.stream();
Stream<String> stream2 = list.parallelStream(); // 병렬 처리 스트림

// Null-Safe
public Stream<String> streamOf(Collection<String> list) {
  returen list == null || list.isEmpty() ? Stream.empty() : list.stream();
}

Stream<String> s = Stream.<String>builder()
                         .add("a")
                         .add("b")
                         .add("c")
                         .build();

Stream.generate(() -> "str").limit(5);

Stream<Integer> s = Stream.iterate(10, n -> n + 1).limit(5); // 10, 11, 12, 13, 14

IntStream s1 = IntStream.range(1, 5); // 1, 2, 3, 4
LongStream s2 = LongStream.rangeClosed(1, 5); // 1, 2, 3, 4, 5
DoubleStream s3 = DoubleStream.of(0, 4, 3); // 0.0, 4.0, 3.0
DoubleStream s4 = new Random().doubles(3)
// boxing
Stream<Integer> s5 = IntStream.range(1, 5).boxed();

IntStream.rangeClosed(2, 12) // 2 ~ 12
         .mapToObj(month -> new Month(month))
         .collect(Collectors.toList());

IntStream s1 = "abc".chars();
Stream<String> s2 = Pattern.compile("\\|").splitAsStream("a|b|c");

Stream<String> s1 = Files.lines(Paths.get("~/words.txt"), StandardCharsets.UTF_8);
Stream<Path> s2 = Files.list(Paths.get("~/Documents"));

Stream<String> s = list.parallelStream();

boolean isParallel = s.isParallel(); // Check parallel stream

s.sequential(); // Change parallel to sequential stream

s.parallel();  // Change sequential to parallel stream

System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20");

Stream<Integer> s1 = Stream.of(1, 3, 5);
Stream<Integer> s2 = Stream.of(2, 4, 6);

Stream<Integer> resultingStream = Stream.concat(s1, s2);

Stream<Integer> s1 = Stream.of(1, 3, 5);
Stream<Integer> s2 = Stream.of(2, 4, 6);
Stream<Integer> s2 = Stream.of(18, 15, 36);

Stream<Integer> resultingStream = Stream.concat(Stream.concat(s1, s2), s3);
// using flatMap()
// Stream<Integer> resultingStream = Stream.of(s1, s2, s3).flatMap(i -> i);

long count = list.stream()
                 .filter(n -> n > 100) // (1)
                 .distinct()
                 .count(); // (2)

@Test
void futureTest() throws InterruptedException, ExecutionException {
    System.out.println("Thread#" + Thread.currentThread().getId());

    ExecutorService executor = Executors.newFixedThreadPool(1); // (1)
    Future<Integer> future = executor.submit(() -> {
        TimeUnit.SECONDS.sleep(1);
        System.out.println("Thread#" + Thread.currentThread().getId());
        return 123;
    });

    System.out.println("future done? " + future.isDone());

    Integer result = future.get(); // (2)

    System.out.println("future done? " + future.isDone());
    System.out.println("result:" + result);
}

@Test
void async() throws InterruptedException {
    AsyncRestTemplate asyncRestTemplate = new AsyncRestTemplate();

    ListenableFuture<ResponseEntity<Map>> entity =
        asyncRestTemplate.getForEntity("https://httpbin.org/get", Map.class);

    entity.addCallback(
        result -> System.out.println(result.getStatusCode()), // (1)
        err -> System.out.println(Arrays.toString(err.getStackTrace()))); // (2)

    System.out.println("test1");
    TimeUnit.SECONDS.sleep(5); // (3)
    System.out.println("test2");
}

public Future<String> getStringAsync() throws InterruptedException {
    CompletableFuture<String> cf = new CompletableFuture<>();
    Executors.newCachedThreadPool().submit(() -> {
        TimeUnit.SECONDS.sleep(1);
        cf.complete("Hello");
        return null;
    });
    return cf;
}

@Test
public void completableFutureExample() throws InterruptedException {
    System.out.println("Thread#" + Thread.currentThread().getId() + ": 1");

    CompletableFuture<String> cf = (CompletableFuture)getAsync();
    cf.thenAccept(s -> { // (1)
        System.out.println("Thread#" + Thread.currentThread().getId() + ": 2");
        System.out.println(s);
    });

    System.out.println("Thread#" + Thread.currentThread().getId() + ": 3");
    TimeUnit.SECONDS.sleep(3);

    System.out.println("Thread#" + Thread.currentThread().getId() + ": 4");
}
// Output:
//   Thread#1: 1
//   Thread#1: 3 (2)
//   Thread#12: 2 (3)
//   Hello
//   Thread#1: 4 (4)

@Service
public class ReceiptService {
    @Async
    public CompletableFuture<Receipt> getAsync(@NotNull String paymentId) {
        final Receipt receipt = restTemplate.getForObject(url, Receipt.class);

        return CompletableFuture.completableFuture(receipt);
    }
}

@RestController
public class ReceiptController {
    @GetMapping("/")
    public List<Receipt> getReceipts() {
        CompletableFuture<Receipt> receipt1 = receiptService.getAsync(id);
        ...
        CompletableFuture.allOf(receipt1, ...);

        return ...;
    }
}

private static final String HOST = "http://fakerestapi.azurewebsites.net/api";

@Test
public void webClientExample() {
    WebClient client = WebClient.create(HOST);

    WebClient.RequestBodySpec req1 = client
        .method(HttpMethod.GET)
        .uri("/Users/1")
        .contentType(MediaType.APPLICATION_JSON_UTF8)
        .accept(MediaType.APPLICATION_JSON_UTF8);
    WebClient.RequestBodySpec req2 = client
        .method(HttpMethod.GET)
        .uri("/Users/2");

    User user1 = req1.exchange().block().bodyToMono(User.class).block();
    User user2 = req2.exchange().block().bodyToMono(User.class).block();

    System.out.println(user1);
    System.out.println(user2);
}

private final static String SLOW_HOST =
    "http://slowwly.robertomurray.co.uk/delay/2000/url/http://fakerestapi.azurewebsites.net";

private final AsyncRestTemplate asyncRestTemplate = new AsyncRestTemplate();
private final RestTemplate restTemplate = new RestTemplate();

@Test
public void test() {
    final List<String> ids = Arrays.asList("1", "2", "3", "4", "5", "6", "7", "8", "9");
    final Map<String, Long> resultsTime = new HashMap<>(3);
    long startTime;

    // asyncRestTemplate
    startTime = System.nanoTime();
    System.out.println(asyncRestTemplate(ids));
    resultsTime.put("asyncRestTemplateElapsedTime", System.nanoTime() - startTime);

    // restTemplate
    startTime = System.nanoTime();
    System.out.println(restTemplate(ids));
    resultsTime.put("restTemplateElapsedTime", System.nanoTime() - startTime);

    // restTemplate with parallel
    startTime = System.nanoTime();
    System.out.println(restTemplateWithParallel(ids));
    resultsTime.put("restTemplateWithParallelElapsedTime", System.nanoTime() - startTime);

    // results
    final DecimalFormat formatter = new DecimalFormat("#,###ms");
    resultsTime.entrySet().stream()
        .sorted(Comparator.comparing(Map.Entry::getValue))
        .forEach(result -> System.out.println(
            result.getKey() + ": " + formatter.format(result.getValue() / 1_000_000)));
}

private List<User> asyncRestTemplate(List<String> ids) {
    List<CompletableFuture<ResponseEntity<User>>> res = new ArrayList<>(ids.size());

    // stream 으로 만들면 lazy하게 동작하면서 한번씩 API 요청?
    ids.forEach(id -> res.add(convert(asyncRestTemplate.getForEntity(
        SLOW_HOST + "/api/Users/" + id, User.class))));

    return res.stream()
        .map(CompletableFuture::join)
        .map(HttpEntity::getBody)
        .collect(Collectors.toList());
}

private List<User> restTemplate(List<String> ids) {
    List<User> users = ids.stream()
        .map(id -> restTemplate.getForEntity(SLOW_HOST + "/api/Users/" + id, User.class))
        .map(HttpEntity::getBody)
        .collect(Collectors.toList());
    return users;
}

private List<User> restTemplateWithParallel(List<String> ids) {
    List<User> users = ids.parallelStream()
        .map(id -> restTemplate.getForEntity(SLOW_HOST + "/api/Users/" + id, User.class))
        .map(HttpEntity::getBody)
        .collect(Collectors.toList());
    return users;
}

@GetMapping("/users/{userId}")
public UserDetail get(@PathVariable(value = "userId") String id) {
    try {
        return Stream.of(id)
            .map(userMapper::get)
            .map(validate) // throw InvalidException
            .map(user -> {
                final String uuid = user.getId();

                CompletableFuture<Accounts> fAccounts = accountConnector.getAccounts(uuid);
                CompletableFuture<Cards> fCards = cardConnector.getCards(uuid);
                CompletableFuture<Point> fPoint = pointConnector.getPoint(uuid);

                return CompletableFuture.allOf(accountsFuture, cardsFuture, pointFuture)
                    .thenApply(ignore -> UserDetail.builder()
                        .user(user)
                        .accounts(fAccounts.join())
                        .cards(fCards.join())
                        .point(fPoint.join())
                        .build())
                    .exceptionally(err -> {
                        throw new ConnectorException(err);
                    });
            })
            .map(CompletableFuture::join)
            .findFirst()
            .orElse(null);
    } catch (InvalidException | ConnectorException e) {
        log.debug(e.getMessage(), e);
        throw e;
    } catch (Exception e) {
        log.error(e.getMessage(), e);
        throw e;
    }
}