-Adding custom thread pool for async calls

-Adding a method to call each server in different threads
-Adding implementation for allOf() and anyOf() methods of CompletableFuture Api
-Adding MovieClient service to demonstrate different calls
-Adding Movie and Review models
-Adding third-party movies-restful-api.jar lib to run movie rest services, can run by:"libs/java -jar  movies-restful-api.jar"

Author: Milad Mofidi

libs/movies-restful-api.jar

@@ -54,6 +54,36 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-core</artifactId>
+            <version>2.14.1</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-webflux</artifactId>
+            <version>5.2.9.RELEASE</version>
+        </dependency>
+
+
+
+        <!--<dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-webflux</artifactId>
+            <version>3.0.0</version>
+        </dependency>-->
+
+
+        <!--        <dependency>
+                    <groupId>org.springframework.boot</groupId>
+                    <artifactId>spring-boot-starter-webflux</artifactId>
+                    <version>2.7.4</version>
+                </dependency>-->
+
+
+
+
     </dependencies>
 
 </project>

pom.xml

@@ -0,0 +1,89 @@
+package org.example.apiclient;
+
+import org.example.domain.movie.Movie;
+import org.example.domain.movie.MovieInfo;
+import org.example.domain.movie.Review;
+import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.util.UriComponentsBuilder;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+
+/**
+ * @author milad mofidi
+ * email: [email protected]
+ * user: miladm on 2/18/2023
+ */
+public class MoviesClient
+{
+    private final WebClient webClient;
+
+    public MoviesClient(WebClient webClient)
+    {
+        this.webClient = webClient;
+    }
+    //Implementing in blocking way
+    public Movie retrieveMovie(Long movieInfoId){
+        
+        var movieInfo = invokeMovieInfo(movieInfoId);
+        var reviews =invokeReviewsService(movieInfoId);
+        return new Movie(movieInfo, reviews);
+    }
+
+
+    //Implementing in non-blocking way
+    public CompletableFuture<Movie> retrieveMovie_Cf_async(Long movieInfoId){
+
+        var movieInfo = CompletableFuture.supplyAsync( () ->invokeMovieInfo(movieInfoId)) ;
+        var reviews =CompletableFuture.supplyAsync( () ->invokeReviewsService(movieInfoId)) ;
+        return movieInfo
+                .thenCombine(reviews, (movieInfoResult, reviewsResult) ->{
+                    return new Movie(movieInfoResult, reviewsResult);
+    });
+
+    }
+    //Calling traditional retrieve movie method to make list of loaded movies
+    public List<Movie> retrieveMovies(List<Long> movieInfoIds){
+        return movieInfoIds.stream().map(movieInfoId -> retrieveMovie(movieInfoId)).collect(Collectors.toList());
+    }
+
+    //Calling completable-future retrieve movie method to make list of loaded movies using retrieveMovie_Cf_async() method
+    public List<Movie> retrieveMovieList_Cf_async(List<Long> movieInfoIds){
+        var movieFutures = movieInfoIds.stream().map(this::retrieveMovie_Cf_async).
+                collect(Collectors.toList());
+        return movieFutures.
+                stream().
+                map(CompletableFuture::join)
+                .collect(Collectors.toList());
+    }
+
+    //Using "allOf()" method to improve performance, this method will wait until all items get processed
+    public List<Movie> retrieveMovieList_Cf_async_allOf(List<Long> movieInfoIds){
+        var movieFutures = movieInfoIds.stream().map(this::retrieveMovie_Cf_async).
+                collect(Collectors.toList());
+
+        var cfAllOf = CompletableFuture.allOf(movieFutures.toArray(new CompletableFuture[movieFutures.size()]));
+        return cfAllOf.thenApply(v -> movieFutures
+                .stream()
+                .map(CompletableFuture::join)
+                .collect(Collectors.toList()))
+                .join();
+    }
+
+    //Implementing the web client call in blocking way
+    private MovieInfo invokeMovieInfo(Long movieInfoId)
+    {
+        var movieInfoUrlPath = "/v1/reviews?movieInfoId=2";
+        return webClient.get().uri(movieInfoUrlPath, movieInfoId).retrieve().bodyToMono(MovieInfo.class).block();
+    }
+
+    //Implementing the web client call in blocking way
+    private List<Review> invokeReviewsService(Long movieInfoId)
+    {
+        ///v1/reviews?movieInfoId=2
+        var reviewUri = UriComponentsBuilder.fromUriString("/v1/reviews").queryParam("movieInfoId", movieInfoId).buildAndExpand().toString();
+        return webClient.get().uri(reviewUri).retrieve().bodyToFlux(Review.class).collectList().block();
+    }
+
+}

src/main/java/org/example/apiclient/MoviesClient.java

@@ -0,0 +1,17 @@
+package org.example.domain.movie;
+
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class Movie {
+
+    private MovieInfo movieInfo;
+    private List<Review> reviewList;
+}

src/main/java/org/example/domain/movie/Movie.java

@@ -0,0 +1,21 @@
+package org.example.domain.movie;
+
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.time.LocalDate;
+import java.util.List;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class MovieInfo {
+    private String movieInfoId;
+    private String name;
+    private Integer year;
+
+    private List<String> cast;
+    private LocalDate release_date;
+}

src/main/java/org/example/domain/movie/MovieInfo.java

@@ -0,0 +1,17 @@
+package org.example.domain.movie;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class Review {
+
+    private String reviewId;
+    private Long movieInfoId;
+    private String comment;
+    //@Min(value = 0L, message = "rating.negative : rating is negative and please pass a non-negative value")
+    private Double rating;
+}