diff --git a/pom.xml b/pom.xml index a71630d289d3..bd3086577753 100644 --- a/pom.xml +++ b/pom.xml @@ -260,6 +260,7 @@ rate-limiting-pattern fallback onion-architecture + scatter-gather diff --git a/scatter-gather/README.md b/scatter-gather/README.md new file mode 100644 index 000000000000..01e8e3a7e406 --- /dev/null +++ b/scatter-gather/README.md @@ -0,0 +1,243 @@ +--- +title: "Scatter-Gather Pattern in Java: Broadcasting One Request and Aggregating Many Replies" +shortTitle: Scatter-Gather +description: "Learn the Scatter-Gather pattern in Java. Send one request to several independent services at once, gather the replies that arrive in time, and aggregate them into a single answer while tolerating slow or failed recipients." +category: Concurrency +language: en +tag: + - Asynchronous + - Decoupling + - Integration + - Messaging + - Scalability +--- + +## Also known as + +* Broadcast and Aggregate +* Request-Reply Broadcast + +## Intent of Scatter-Gather Design Pattern + +Send the same request to a number of independent recipients concurrently, collect the replies that arrive within a deadline, and combine them into one result. The caller pays roughly the latency of the slowest tolerated recipient instead of the sum of all latencies, and a single slow or failing recipient does not prevent an answer. + +## Detailed Explanation of Scatter-Gather Pattern with Real-World Examples + +Real-world example + +> A travel site has to show the price of a hotel stay. It does not own a single price list; several rate providers each have their own. Asking them one after another would make the page as slow as all of them together, and one provider that is down would block the result. Instead the site scatters the same request to every provider at the same time, waits a few hundred milliseconds for their replies, drops the ones that did not answer in time, and shows the cheapest of the quotes it gathered. + +In plain words + +> Ask everybody the same question at once, wait a bounded time, and aggregate the answers you got. + +Enterprise Integration Patterns says + +> Use a Scatter-Gather that broadcasts a message to multiple recipients and re-aggregates the responses back into a single message. + +How it differs from Fan-Out/Fan-In + +> [Fan-Out/Fan-In](../fanout-fanin) splits one job into sub-tasks of the same kind and needs all of them back to build the result. Scatter-Gather sends the same message to different, independent services, expects heterogeneous replies, and is designed to produce a result even when some recipients are slow or unavailable. + +Sequence diagram + +```mermaid +sequenceDiagram + participant Client + participant ScatterGather + participant Atlas as Atlas Hotels + participant Harbor as Harbor Stays + participant Sleepy as Sleepy Suites (slow) + participant Flaky as Flaky Inns (down) + + Client->>ScatterGather: scatter(request) + par broadcast + ScatterGather->>Atlas: quote(request) + ScatterGather->>Harbor: quote(request) + ScatterGather->>Sleepy: quote(request) + ScatterGather->>Flaky: quote(request) + end + Atlas-->>ScatterGather: 387.00 + Harbor-->>ScatterGather: 295.50 + Flaky-->>ScatterGather: error + Note over ScatterGather,Sleepy: timeout expires, reply dropped + ScatterGather->>ScatterGather: gather() keeps 2 of 4 replies + ScatterGather->>Client: aggregate() returns Harbor Stays 295.50 +``` + +## Programmatic Example of Scatter-Gather Pattern in Java + +The request is a plain immutable value. Every recipient receives exactly the same instance. + +```java +public record RateRequest(String city, LocalDate checkIn, int nights) {} + +public record RateQuote(String provider, BigDecimal total) {} +``` + +Each recipient implements `RateProvider`. In a real system these would be remote services with their own latency and failure profile. The demo ships a fast in-memory provider, a decorator that delays any provider, and a provider that is down. + +```java +public interface RateProvider { + String name(); + + RateQuote quote(RateRequest request); +} +``` + +The `Aggregator` reduces the gathered replies. Because it is a separate strategy the same coordinator can serve callers that want the cheapest quote, an average, or the full list. + +```java +@FunctionalInterface +public interface Aggregator { + R aggregate(List replies); + + static Aggregator> cheapestQuote() { + return replies -> replies.stream().min(Comparator.comparing(RateQuote::total)); + } +} +``` + +`ScatterGather` implements the three phases. The scatter phase submits one asynchronous call per provider and attaches the timeout to each future. Nothing is awaited yet. + +```java +public List scatter(RateRequest request, List providers) { + var pending = new ArrayList(); + for (var provider : providers) { + var reply = + CompletableFuture.supplyAsync(() -> provider.quote(request), executor) + .orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS); + pending.add(new PendingReply(provider, reply)); + } + return pending; +} +``` + +The gather phase waits until every future has settled, then keeps the successful replies. Timeouts and failures are logged and dropped, which is what allows the caller to get an answer from the providers that did respond. + +```java +public List gather(List pending) { + CompletableFuture.allOf(pending.stream().map(PendingReply::reply).toArray(CompletableFuture[]::new)) + .exceptionally(ex -> null) + .join(); + var quotes = new ArrayList(); + for (var entry : pending) { + try { + quotes.add(entry.reply().join()); + } catch (CompletionException e) { + if (e.getCause() instanceof TimeoutException) { + LOGGER.warn("Dropping {}: no reply within {} ms", entry.provider().name(), timeout.toMillis()); + } else { + LOGGER.warn("Dropping {}: {}", entry.provider().name(), e.getCause().getMessage()); + } + } + } + return quotes; +} +``` + +A convenience method chains the phases together. + +```java +public R scatterGather( + RateRequest request, List providers, Aggregator aggregator) { + return aggregator.aggregate(gather(scatter(request, providers))); +} +``` + +The demo application asks four providers for the same three-night stay. One provider needs two seconds while the gather timeout is 300 milliseconds, and one provider is down. The site still answers with the cheapest of the two quotes it gathered. + +```java +var request = new RateRequest("Lisbon", LocalDate.of(2026, 10, 3), 3); +var providers = + List.of( + new InMemoryRateProvider("Atlas Hotels", new BigDecimal("129.00")), + new InMemoryRateProvider("Harbor Stays", new BigDecimal("98.50")), + new DelayedRateProvider( + new InMemoryRateProvider("Sleepy Suites", new BigDecimal("75.00")), + Duration.ofSeconds(2)), + new FailingRateProvider("Flaky Inns")); + +try (var scatterGather = + new ScatterGather(Executors.newFixedThreadPool(providers.size()), Duration.ofMillis(300))) { + var pending = scatterGather.scatter(request, providers); + var quotes = scatterGather.gather(pending); + reportBestOffer(Aggregator.cheapestQuote().aggregate(quotes)); +} +``` + +The result is reported by a small helper so the empty case is handled explicitly. + +```java +static void reportBestOffer(Optional best) { + best.ifPresentOrElse( + offer -> LOGGER.info("Best offer: {} at {}", offer.provider(), offer.total()), + () -> LOGGER.info("No provider answered in time")); +} +``` + +Running the program produces output similar to the following. + +``` +Scatter phase: broadcasting the same request to every provider +Scattering request for 3 nights in Lisbon to 4 providers +Atlas Hotels quotes 387.00 for 3 nights in Lisbon +Harbor Stays quotes 295.50 for 3 nights in Lisbon +Gather phase: collecting replies that arrive within the timeout +Sleepy Suites is slow and will need 2000 ms to answer +Gathered quote 387.00 from Atlas Hotels +Gathered quote 295.50 from Harbor Stays +Dropping Sleepy Suites: no reply within 300 ms +Dropping Flaky Inns: Flaky Inns is unavailable +Gathered 2 of 4 replies +Aggregate phase: choosing the cheapest of 2 quotes +Best offer: Harbor Stays at 295.50 +``` + +## Class diagram + +See [scatter-gather.urm.puml](./etc/scatter-gather.urm.puml) for the PlantUML class diagram. + +## When to Use the Scatter-Gather Pattern in Java + +* The same question has to be answered by several independent services, such as price comparison, search federation, or quorum reads. +* The latency of asking recipients sequentially is unacceptable. +* A partial answer built from the recipients that replied in time is more valuable than no answer. +* The recipients are unknown or change at runtime, so the caller should only depend on a common contract. + +## Real-World Applications of Scatter-Gather Pattern in Java + +* Travel and shopping comparison sites that query many suppliers for the same product. +* Distributed search engines that broadcast a query to every index shard and merge the ranked results. +* Quorum reads in replicated data stores that ask several replicas and accept the first consistent majority. +* [Apache Camel Scatter-Gather EIP](https://camel.apache.org/components/latest/eips/scatter-gather.html) +* [Spring Integration Scatter-Gather](https://docs.spring.io/spring-integration/reference/scatter-gather.html) +* [Akka scatter-gather with `ask` and `Future.sequence`](https://doc.akka.io/docs/akka/current/futures.html) + +## Benefits and Trade-offs of Scatter-Gather Pattern + +Benefits: + +* **Lower latency**: recipients are called concurrently, so the caller waits for the slowest tolerated reply rather than the sum of all replies. +* **Resilience**: a timeout bounds the wait and a failed recipient is simply left out of the aggregate. +* **Decoupling**: the caller depends only on the recipient contract and the aggregation strategy, not on the number or identity of recipients. + +Trade-offs: + +* **Partial results**: the caller must be able to live with an answer built from a subset of recipients, and the aggregator must handle an empty set. +* **Resource usage**: every request occupies one thread or connection per recipient; a dropped reply may still be computed by the recipient. +* **Tuning**: the timeout is a compromise between completeness and responsiveness and usually needs measurement to get right. + +## Related Java Design Patterns + +* [Fan-Out/Fan-In](../fanout-fanin): splits one task into homogeneous sub-tasks and waits for all of them; Scatter-Gather broadcasts one request to heterogeneous recipients and tolerates missing replies. +* [Microservices Aggregator](../microservices-aggregrator): a service that composes the responses of several downstream services; Scatter-Gather is a way to fetch those responses concurrently. +* [Async Method Invocation](../async-method-invocation): the mechanism used to call each recipient without blocking the caller. +* [Promise](../promise): each pending reply is a promise that either completes with a quote or fails. +* Timeout: bounds how long the gather phase waits for each recipient. + +## References and Credits + +* [Enterprise Integration Patterns](https://www.amazon.com/gp/product/0321200683) (Gregor Hohpe and Bobby Woolf) +* [Scatter-Gather at enterpriseintegrationpatterns.com](https://www.enterpriseintegrationpatterns.com/patterns/messaging/BroadcastAggregate.html) +* [Java Concurrency in Practice](https://www.amazon.com/gp/product/0321349601) (Brian Goetz) diff --git a/scatter-gather/etc/scatter-gather.urm.puml b/scatter-gather/etc/scatter-gather.urm.puml new file mode 100644 index 000000000000..42a7a26d5b2c --- /dev/null +++ b/scatter-gather/etc/scatter-gather.urm.puml @@ -0,0 +1,74 @@ +@startuml +package com.iluwatar.scattergather { + class RateRequest { + + RateRequest(city : String, checkIn : LocalDate, nights : int) + + city() : String + + checkIn() : LocalDate + + nights() : int + } + class RateQuote { + + RateQuote(provider : String, total : BigDecimal) + + provider() : String + + total() : BigDecimal + } + interface RateProvider { + + name() : String {abstract} + + quote(request : RateRequest) : RateQuote {abstract} + } + class InMemoryRateProvider { + - name : String + - nightlyRate : BigDecimal + + InMemoryRateProvider(name : String, nightlyRate : BigDecimal) + + name() : String + + quote(request : RateRequest) : RateQuote + } + class DelayedRateProvider { + - delegate : RateProvider + - delay : Duration + + DelayedRateProvider(delegate : RateProvider, delay : Duration) + + name() : String + + quote(request : RateRequest) : RateQuote + } + class FailingRateProvider { + - name : String + + FailingRateProvider(name : String) + + name() : String + + quote(request : RateRequest) : RateQuote + } + interface Aggregator { + + aggregate(replies : List) : R {abstract} + + cheapestQuote() : Aggregator> {static} + } + class PendingReply { + + PendingReply(provider : RateProvider, reply : CompletableFuture) + + provider() : RateProvider + + reply() : CompletableFuture + } + class ScatterGather { + - executor : ExecutorService + - timeout : Duration + + ScatterGather(executor : ExecutorService, timeout : Duration) + + scatter(request : RateRequest, providers : List) : List + + gather(pending : List) : List + + scatterGather(request : RateRequest, providers : List, aggregator : Aggregator) : R + + close() : void + } + class App { + + App() + + main(args : String[]) : void + ~ reportBestOffer(best : Optional) : void {static} + } +} +InMemoryRateProvider ..|> RateProvider +DelayedRateProvider ..|> RateProvider +DelayedRateProvider --> RateProvider +FailingRateProvider ..|> RateProvider +RateProvider ..> RateRequest +RateProvider ..> RateQuote +PendingReply --> RateProvider +PendingReply ..> RateQuote +ScatterGather ..> PendingReply +ScatterGather ..> Aggregator +ScatterGather --> "*" RateProvider +App ..> ScatterGather +@enduml diff --git a/scatter-gather/pom.xml b/scatter-gather/pom.xml new file mode 100644 index 000000000000..34012dd88a31 --- /dev/null +++ b/scatter-gather/pom.xml @@ -0,0 +1,70 @@ + + + + 4.0.0 + + com.iluwatar + java-design-patterns + 1.26.0-SNAPSHOT + + scatter-gather + + + org.slf4j + slf4j-api + + + ch.qos.logback + logback-classic + + + org.junit.jupiter + junit-jupiter-engine + test + + + + + + org.apache.maven.plugins + maven-assembly-plugin + + + + + + com.iluwatar.scattergather.App + + + + + + + + + diff --git a/scatter-gather/src/main/java/com/iluwatar/scattergather/Aggregator.java b/scatter-gather/src/main/java/com/iluwatar/scattergather/Aggregator.java new file mode 100644 index 000000000000..4cac13e31747 --- /dev/null +++ b/scatter-gather/src/main/java/com/iluwatar/scattergather/Aggregator.java @@ -0,0 +1,54 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.scattergather; + +import java.util.Comparator; +import java.util.List; +import java.util.Optional; + +/** + * Reduces the gathered replies into a single result. The aggregation strategy is pluggable so the + * same scatter and gather machinery can serve callers that want the cheapest quote, the average + * price, or the full list. + * + * @param the type of the gathered replies + * @param the type of the aggregated result + */ +@FunctionalInterface +public interface Aggregator { + + /** + * Combines the gathered replies. + * + * @param replies the replies that arrived in time, possibly empty + * @return the aggregated result + */ + R aggregate(List replies); + + /** Returns an aggregator that picks the quote with the lowest total price. */ + static Aggregator> cheapestQuote() { + return replies -> replies.stream().min(Comparator.comparing(RateQuote::total)); + } +} diff --git a/scatter-gather/src/main/java/com/iluwatar/scattergather/App.java b/scatter-gather/src/main/java/com/iluwatar/scattergather/App.java new file mode 100644 index 000000000000..21136f6e574b --- /dev/null +++ b/scatter-gather/src/main/java/com/iluwatar/scattergather/App.java @@ -0,0 +1,88 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.scattergather; + +import java.math.BigDecimal; +import java.time.Duration; +import java.time.LocalDate; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.Executors; +import lombok.extern.slf4j.Slf4j; + +/** + * The Scatter-Gather pattern sends one request to several independent recipients at the same time, + * gathers whatever replies arrive within a deadline, and aggregates them into a single answer. It + * is the integration counterpart of the fan-out/fan-in pattern: fan-out/fan-in splits one job into + * sub-tasks of the same kind, whereas scatter-gather broadcasts the same message to different + * services and has to cope with some of them being slow or unavailable. + * + *

This demo models a travel site that asks four hotel rate providers for the price of the same + * stay. One provider is slower than the gather timeout and one is down. The site still answers, + * using the quotes from the two healthy providers, and the {@link Aggregator} picks the cheapest. + */ +@Slf4j +public class App { + + /** + * Program entry point. + * + * @param args command line arguments, unused + */ + public static void main(String[] args) { + var request = new RateRequest("Lisbon", LocalDate.of(2026, 10, 3), 3); + var providers = + List.of( + new InMemoryRateProvider("Atlas Hotels", new BigDecimal("129.00")), + new InMemoryRateProvider("Harbor Stays", new BigDecimal("98.50")), + new DelayedRateProvider( + new InMemoryRateProvider("Sleepy Suites", new BigDecimal("75.00")), + Duration.ofSeconds(2)), + new FailingRateProvider("Flaky Inns")); + + try (var scatterGather = + new ScatterGather(Executors.newFixedThreadPool(providers.size()), Duration.ofMillis(300))) { + LOGGER.info("Scatter phase: broadcasting the same request to every provider"); + var pending = scatterGather.scatter(request, providers); + + LOGGER.info("Gather phase: collecting replies that arrive within the timeout"); + var quotes = scatterGather.gather(pending); + + LOGGER.info("Aggregate phase: choosing the cheapest of {} quotes", quotes.size()); + reportBestOffer(Aggregator.cheapestQuote().aggregate(quotes)); + } + } + + /** + * Logs the aggregated result, or the fact that no provider answered in time. + * + * @param best the cheapest gathered quote, empty when every provider failed or timed out + */ + static void reportBestOffer(Optional best) { + best.ifPresentOrElse( + offer -> LOGGER.info("Best offer: {} at {}", offer.provider(), offer.total()), + () -> LOGGER.info("No provider answered in time")); + } +} diff --git a/scatter-gather/src/main/java/com/iluwatar/scattergather/DelayedRateProvider.java b/scatter-gather/src/main/java/com/iluwatar/scattergather/DelayedRateProvider.java new file mode 100644 index 000000000000..88ca680fe195 --- /dev/null +++ b/scatter-gather/src/main/java/com/iluwatar/scattergather/DelayedRateProvider.java @@ -0,0 +1,67 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.scattergather; + +import java.time.Duration; +import lombok.extern.slf4j.Slf4j; + +/** + * Wraps another provider and delays its reply, simulating a slow remote service. When the delay + * exceeds the gather timeout the reply is dropped and the remaining quotes are used instead. + */ +@Slf4j +public class DelayedRateProvider implements RateProvider { + + private final RateProvider delegate; + private final Duration delay; + + /** + * Creates a provider that answers only after the given delay. + * + * @param delegate the provider that produces the actual quote + * @param delay how long to wait before delegating + */ + public DelayedRateProvider(RateProvider delegate, Duration delay) { + this.delegate = delegate; + this.delay = delay; + } + + @Override + public String name() { + return delegate.name(); + } + + @Override + public RateQuote quote(RateRequest request) { + LOGGER.info("{} is slow and will need {} ms to answer", name(), delay.toMillis()); + try { + Thread.sleep(delay); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(name() + " was interrupted before answering", e); + } + return delegate.quote(request); + } +} diff --git a/scatter-gather/src/main/java/com/iluwatar/scattergather/FailingRateProvider.java b/scatter-gather/src/main/java/com/iluwatar/scattergather/FailingRateProvider.java new file mode 100644 index 000000000000..f2fa3cdbc1fc --- /dev/null +++ b/scatter-gather/src/main/java/com/iluwatar/scattergather/FailingRateProvider.java @@ -0,0 +1,53 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.scattergather; + +/** + * A provider that is currently unavailable. Its failure must not prevent the caller from receiving + * the quotes of the healthy providers. + */ +public class FailingRateProvider implements RateProvider { + + private final String name; + + /** + * Creates a provider that always fails. + * + * @param name the provider name + */ + public FailingRateProvider(String name) { + this.name = name; + } + + @Override + public String name() { + return name; + } + + @Override + public RateQuote quote(RateRequest request) { + throw new IllegalStateException(name + " is unavailable"); + } +} diff --git a/scatter-gather/src/main/java/com/iluwatar/scattergather/InMemoryRateProvider.java b/scatter-gather/src/main/java/com/iluwatar/scattergather/InMemoryRateProvider.java new file mode 100644 index 000000000000..71571143ac8e --- /dev/null +++ b/scatter-gather/src/main/java/com/iluwatar/scattergather/InMemoryRateProvider.java @@ -0,0 +1,59 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.scattergather; + +import java.math.BigDecimal; +import lombok.extern.slf4j.Slf4j; + +/** A fast provider that answers immediately with a fixed nightly rate. */ +@Slf4j +public class InMemoryRateProvider implements RateProvider { + + private final String name; + private final BigDecimal nightlyRate; + + /** + * Creates a provider with a fixed price per night. + * + * @param name the provider name + * @param nightlyRate the price charged for one night + */ + public InMemoryRateProvider(String name, BigDecimal nightlyRate) { + this.name = name; + this.nightlyRate = nightlyRate; + } + + @Override + public String name() { + return name; + } + + @Override + public RateQuote quote(RateRequest request) { + var total = nightlyRate.multiply(BigDecimal.valueOf(request.nights())); + LOGGER.info("{} quotes {} for {} nights in {}", name, total, request.nights(), request.city()); + return new RateQuote(name, total); + } +} diff --git a/scatter-gather/src/main/java/com/iluwatar/scattergather/RateProvider.java b/scatter-gather/src/main/java/com/iluwatar/scattergather/RateProvider.java new file mode 100644 index 000000000000..d5cee3448dbd --- /dev/null +++ b/scatter-gather/src/main/java/com/iluwatar/scattergather/RateProvider.java @@ -0,0 +1,44 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.scattergather; + +/** + * A recipient of the scattered request. In a real system each provider would be a separate remote + * service with its own latency and failure profile, which is exactly why the caller cannot assume + * that every reply arrives, or arrives in time. + */ +public interface RateProvider { + + /** Returns the provider's name, used to identify its reply in logs and quotes. */ + String name(); + + /** + * Produces a quote for the given request. + * + * @param request the stay to quote + * @return the provider's offer + */ + RateQuote quote(RateRequest request); +} diff --git a/scatter-gather/src/main/java/com/iluwatar/scattergather/RateQuote.java b/scatter-gather/src/main/java/com/iluwatar/scattergather/RateQuote.java new file mode 100644 index 000000000000..964531395611 --- /dev/null +++ b/scatter-gather/src/main/java/com/iluwatar/scattergather/RateQuote.java @@ -0,0 +1,36 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.scattergather; + +import java.math.BigDecimal; + +/** + * A single reply gathered from one {@link RateProvider}. The gather phase collects these and the + * {@link Aggregator} reduces them into one answer for the caller. + * + * @param provider the name of the provider that produced the quote + * @param total the total price for the whole stay + */ +public record RateQuote(String provider, BigDecimal total) {} diff --git a/scatter-gather/src/main/java/com/iluwatar/scattergather/RateRequest.java b/scatter-gather/src/main/java/com/iluwatar/scattergather/RateRequest.java new file mode 100644 index 000000000000..f6d86c731f87 --- /dev/null +++ b/scatter-gather/src/main/java/com/iluwatar/scattergather/RateRequest.java @@ -0,0 +1,46 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.scattergather; + +import java.time.LocalDate; + +/** + * The request that is scattered unchanged to every {@link RateProvider}. Each recipient receives + * the same message, which is what distinguishes Scatter-Gather from patterns that split work into + * different sub-tasks. + * + * @param city the destination city + * @param checkIn the first night of the stay + * @param nights how many nights the guest stays + */ +public record RateRequest(String city, LocalDate checkIn, int nights) { + + /** Validates the request so that recipients never have to. */ + public RateRequest { + if (nights <= 0) { + throw new IllegalArgumentException("nights must be positive"); + } + } +} diff --git a/scatter-gather/src/main/java/com/iluwatar/scattergather/ScatterGather.java b/scatter-gather/src/main/java/com/iluwatar/scattergather/ScatterGather.java new file mode 100644 index 000000000000..f3e116bba3a8 --- /dev/null +++ b/scatter-gather/src/main/java/com/iluwatar/scattergather/ScatterGather.java @@ -0,0 +1,154 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.scattergather; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import lombok.extern.slf4j.Slf4j; + +/** + * Coordinates the three phases of the pattern. + * + *

    + *
  1. Scatter: the same request is sent to every provider concurrently. + *
  2. Gather: replies are collected until each one has either arrived, failed, or exceeded + * the timeout. Late and failed replies are logged and dropped so a single slow provider + * cannot hold up the whole answer. + *
  3. Aggregate: the gathered replies are reduced by an {@link Aggregator}. + *
+ * + *

The class owns the executor it was given and shuts it down on {@link #close()}. + */ +@Slf4j +public class ScatterGather implements AutoCloseable { + + /** + * A reply that is still in flight after the scatter phase. + * + * @param provider the provider the request was sent to + * @param reply the future that completes with the provider's quote, or exceptionally + */ + public record PendingReply(RateProvider provider, CompletableFuture reply) {} + + private final ExecutorService executor; + private final Duration timeout; + + /** + * Creates a coordinator. + * + * @param executor runs the calls to the providers; it is shut down when this object is closed + * @param timeout how long the gather phase waits for each reply + */ + public ScatterGather(ExecutorService executor, Duration timeout) { + this.executor = executor; + this.timeout = timeout; + } + + /** + * Scatter phase: sends the request to every provider without waiting for any reply. + * + * @param request the request to broadcast + * @param providers the recipients + * @return one pending reply per provider, in the same order as the providers + */ + public List scatter(RateRequest request, List providers) { + LOGGER.info( + "Scattering request for {} nights in {} to {} providers", + request.nights(), + request.city(), + providers.size()); + var pending = new ArrayList(); + for (var provider : providers) { + var reply = + CompletableFuture.supplyAsync(() -> provider.quote(request), executor) + .orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS); + pending.add(new PendingReply(provider, reply)); + } + return pending; + } + + /** + * Gather phase: waits until every pending reply has settled and keeps the successful ones. + * + * @param pending the replies produced by {@link #scatter} + * @return the quotes that arrived in time, possibly fewer than the number of providers + */ + public List gather(List pending) { + CompletableFuture.allOf( + pending.stream().map(PendingReply::reply).toArray(CompletableFuture[]::new)) + .exceptionally(ex -> null) + .join(); + var quotes = new ArrayList(); + for (var entry : pending) { + try { + var quote = entry.reply().join(); + LOGGER.info("Gathered quote {} from {}", quote.total(), entry.provider().name()); + quotes.add(quote); + } catch (CompletionException e) { + if (e.getCause() instanceof TimeoutException) { + LOGGER.warn( + "Dropping {}: no reply within {} ms", entry.provider().name(), timeout.toMillis()); + } else { + LOGGER.warn("Dropping {}: {}", entry.provider().name(), e.getCause().getMessage()); + } + } + } + LOGGER.info("Gathered {} of {} replies", quotes.size(), pending.size()); + return quotes; + } + + /** + * Runs all three phases: scatter, gather, and aggregate. + * + * @param request the request to broadcast + * @param providers the recipients + * @param aggregator reduces the gathered quotes + * @param the aggregated result type + * @return the aggregated result + */ + public R scatterGather( + RateRequest request, List providers, Aggregator aggregator) { + return aggregator.aggregate(gather(scatter(request, providers))); + } + + /** Stops the executor, interrupting providers that are still working on a dropped request. */ + @Override + public void close() { + executor.shutdownNow(); + try { + if (!executor.awaitTermination(1, TimeUnit.SECONDS)) { + LOGGER.warn("Executor did not terminate within one second"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } +} diff --git a/scatter-gather/src/test/java/com/iluwatar/scattergather/AggregatorTest.java b/scatter-gather/src/test/java/com/iluwatar/scattergather/AggregatorTest.java new file mode 100644 index 000000000000..611e7338a363 --- /dev/null +++ b/scatter-gather/src/test/java/com/iluwatar/scattergather/AggregatorTest.java @@ -0,0 +1,54 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.scattergather; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.math.BigDecimal; +import java.util.List; +import org.junit.jupiter.api.Test; + +class AggregatorTest { + + @Test + void shouldPickLowestTotal() { + var quotes = + List.of( + new RateQuote("a", new BigDecimal("200.00")), + new RateQuote("b", new BigDecimal("99.99")), + new RateQuote("c", new BigDecimal("100.00"))); + + var best = Aggregator.cheapestQuote().aggregate(quotes); + + assertTrue(best.isPresent()); + assertEquals("b", best.get().provider()); + } + + @Test + void shouldReturnEmptyForNoQuotes() { + assertTrue(Aggregator.cheapestQuote().aggregate(List.of()).isEmpty()); + } +} diff --git a/scatter-gather/src/test/java/com/iluwatar/scattergather/AppTest.java b/scatter-gather/src/test/java/com/iluwatar/scattergather/AppTest.java new file mode 100644 index 000000000000..588c5021382c --- /dev/null +++ b/scatter-gather/src/test/java/com/iluwatar/scattergather/AppTest.java @@ -0,0 +1,57 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.scattergather; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import java.math.BigDecimal; +import java.util.Optional; +import org.junit.jupiter.api.Test; + +class AppTest { + + @Test + void shouldLaunchApp() { + assertDoesNotThrow(() -> App.main(new String[] {})); + } + + @Test + void shouldBeInstantiable() { + assertNotNull(new App(), "App should be instantiable"); + } + + @Test + void shouldReportBestOfferWhenPresent() { + var best = Optional.of(new RateQuote("Harbor Stays", new BigDecimal("295.50"))); + + assertDoesNotThrow(() -> App.reportBestOffer(best)); + } + + @Test + void shouldReportWhenNoProviderAnswered() { + assertDoesNotThrow(() -> App.reportBestOffer(Optional.empty())); + } +} diff --git a/scatter-gather/src/test/java/com/iluwatar/scattergather/RateProviderTest.java b/scatter-gather/src/test/java/com/iluwatar/scattergather/RateProviderTest.java new file mode 100644 index 000000000000..b0f1bf16d853 --- /dev/null +++ b/scatter-gather/src/test/java/com/iluwatar/scattergather/RateProviderTest.java @@ -0,0 +1,71 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.scattergather; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.math.BigDecimal; +import java.time.Duration; +import java.time.LocalDate; +import org.junit.jupiter.api.Test; + +class RateProviderTest { + + private static final RateRequest REQUEST = + new RateRequest("Madrid", LocalDate.of(2026, 3, 10), 4); + + @Test + void inMemoryProviderMultipliesNightlyRateByNights() { + var provider = new InMemoryRateProvider("inn", new BigDecimal("25.50")); + + assertEquals("inn", provider.name()); + assertEquals(new RateQuote("inn", new BigDecimal("102.00")), provider.quote(REQUEST)); + } + + @Test + void delayedProviderDelegatesAfterWaiting() { + var provider = + new DelayedRateProvider( + new InMemoryRateProvider("slow", new BigDecimal("10.00")), Duration.ofMillis(10)); + + assertEquals("slow", provider.name()); + assertEquals(new RateQuote("slow", new BigDecimal("40.00")), provider.quote(REQUEST)); + } + + @Test + void failingProviderThrows() { + var provider = new FailingRateProvider("down"); + + assertEquals("down", provider.name()); + assertThrows(IllegalStateException.class, () -> provider.quote(REQUEST)); + } + + @Test + void requestRejectsNonPositiveNights() { + assertThrows( + IllegalArgumentException.class, () -> new RateRequest("Rome", LocalDate.of(2026, 1, 1), 0)); + } +} diff --git a/scatter-gather/src/test/java/com/iluwatar/scattergather/ScatterGatherTest.java b/scatter-gather/src/test/java/com/iluwatar/scattergather/ScatterGatherTest.java new file mode 100644 index 000000000000..5fead3458b6a --- /dev/null +++ b/scatter-gather/src/test/java/com/iluwatar/scattergather/ScatterGatherTest.java @@ -0,0 +1,205 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.scattergather; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.math.BigDecimal; +import java.time.Duration; +import java.time.LocalDate; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +class ScatterGatherTest { + + private static final RateRequest REQUEST = new RateRequest("Porto", LocalDate.of(2026, 5, 1), 2); + private static final Duration TIMEOUT = Duration.ofMillis(300); + + private final CountDownLatch gate = new CountDownLatch(1); + private ExecutorService executor; + private ScatterGather scatterGather; + + @BeforeEach + void setUp() { + executor = Executors.newFixedThreadPool(4); + scatterGather = new ScatterGather(executor, TIMEOUT); + } + + @AfterEach + void tearDown() { + gate.countDown(); + scatterGather.close(); + } + + @Test + void shouldGatherEveryReplyWhenAllProvidersAnswer() { + var providers = List.of(provider("A", "120.00"), provider("B", "80.00")); + + var quotes = scatterGather.gather(scatterGather.scatter(REQUEST, providers)); + + assertEquals(2, quotes.size()); + assertEquals(new RateQuote("A", new BigDecimal("240.00")), quotes.get(0)); + assertEquals(new RateQuote("B", new BigDecimal("160.00")), quotes.get(1)); + } + + @Test + void shouldDropProviderThatMissesTheTimeout() { + var providers = List.of(provider("fast", "100.00"), blockedProvider("stuck")); + + var quotes = scatterGather.gather(scatterGather.scatter(REQUEST, providers)); + + assertEquals(List.of(new RateQuote("fast", new BigDecimal("200.00"))), quotes); + } + + @Test + void shouldDropProviderThatFails() { + var providers = List.of(new FailingRateProvider("down"), provider("up", "50.00")); + + var quotes = scatterGather.gather(scatterGather.scatter(REQUEST, providers)); + + assertEquals(List.of(new RateQuote("up", new BigDecimal("100.00"))), quotes); + } + + @Test + void shouldAggregateCheapestQuote() { + var providers = + List.of( + provider("pricey", "300.00"), provider("cheap", "90.00"), provider("mid", "150.00")); + + var best = scatterGather.scatterGather(REQUEST, providers, Aggregator.cheapestQuote()); + + assertTrue(best.isPresent()); + assertEquals("cheap", best.get().provider()); + assertEquals(new BigDecimal("180.00"), best.get().total()); + } + + @Test + void shouldReturnEmptyResultWhenNoProviderAnswers() { + var providers = + List.of(new FailingRateProvider("down"), blockedProvider("stuck")); + + var best = scatterGather.scatterGather(REQUEST, providers, Aggregator.cheapestQuote()); + + assertTrue(best.isEmpty()); + } + + @Test + void shouldShutDownExecutorOnClose() { + scatterGather.scatter(REQUEST, List.of(blockedProvider("stuck"))); + + scatterGather.close(); + + assertTrue(executor.isShutdown()); + assertTrue(executor.isTerminated()); + } + + @Test + void shouldPreserveInterruptFlagWhenCloseIsInterrupted() { + var stubborn = new CountDownLatch(1); + var ownExecutor = Executors.newSingleThreadExecutor(); + var subject = new ScatterGather(ownExecutor, Duration.ofSeconds(10)); + subject.scatter(REQUEST, List.of(interruptIgnoringProvider("stubborn", stubborn))); + try { + Thread.currentThread().interrupt(); + + subject.close(); + + assertTrue(Thread.interrupted(), "close must re-set the interrupt flag it swallowed"); + assertTrue(ownExecutor.isShutdown()); + } finally { + stubborn.countDown(); + } + } + + @Test + void shouldReturnFromCloseWhenTaskIgnoresInterrupts() { + var stubborn = new CountDownLatch(1); + var ownExecutor = Executors.newSingleThreadExecutor(); + var subject = new ScatterGather(ownExecutor, Duration.ofSeconds(10)); + subject.scatter(REQUEST, List.of(interruptIgnoringProvider("stubborn", stubborn))); + try { + subject.close(); + + assertTrue(ownExecutor.isShutdown()); + assertFalse(ownExecutor.isTerminated(), "the stubborn task is still running after close"); + } finally { + stubborn.countDown(); + } + } + + private static RateProvider provider(String name, String nightlyRate) { + return new InMemoryRateProvider(name, new BigDecimal(nightlyRate)); + } + + /** A provider that does not answer until the test releases the gate. */ + private RateProvider blockedProvider(String name) { + return new RateProvider() { + @Override + public String name() { + return name; + } + + @Override + public RateQuote quote(RateRequest request) { + try { + gate.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("interrupted", e); + } + return new RateQuote(name, BigDecimal.ONE); + } + }; + } + + /** A provider that keeps waiting on the latch even when its thread is interrupted. */ + private static RateProvider interruptIgnoringProvider(String name, CountDownLatch latch) { + return new RateProvider() { + @Override + public String name() { + return name; + } + + @Override + public RateQuote quote(RateRequest request) { + while (true) { + try { + latch.await(); + return new RateQuote(name, BigDecimal.ONE); + } catch (InterruptedException ignored) { + // deliberately keeps waiting to simulate a task that does not honour interrupts + } + } + } + }; + } +}