package mrhaki;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.function.Consumer;

class Async {

    private final ExecutorService executorService;
    private final WeatherService weatherService;

    Async(ExecutorService executorService, WeatherService weatherService) {
        this.executorService = executorService;
        this.weatherService = weatherService;
    }

    // Find temperature for a city using WeatherService and
    // assign result using callback argument.
    void findTemperature(String city, Consumer<Result> callback) {
        // Here we do an asynchronous call using ExecutorService and
        // a Runnable lambda.
        // We're assigning the result of calling WeatherService using
        // the callback argument.
        executorService.submit(() -> {
            int temperature = weatherService.getTemperature(city);
            var result = new Result(city, temperature);
            callback.accept(result);
        });
    }

    // Find temperature for multiple cities using WeatherService and
    // assign result using callback argument.
    void findTemperatures(List<String> cities, Consumer<Result> callback) {
        cities.forEach(city -> findTemperature(city, callback));
    }

    record Result(String city, int temperature) {
    }

    interface WeatherService {
        int getTemperature(String city);
    }
}

package mrhaki

import spock.lang.Specification
import spock.lang.Subject
import spock.util.concurrent.BlockingVariable
import spock.util.concurrent.BlockingVariables

import java.util.concurrent.Executors

class AsyncSpec extends Specification {

    // Provide a stub for the WeatherService interface.
    // Return 21 on the first call and 18 on subsequent calls.
    private Async.WeatherService weatherService = Stub() {
        getTemperature(_ as String) >>> [21, 18]
    }

    // We want to test the class Async
    @Subject
    private Async async = new Async(Executors.newFixedThreadPool(2), weatherService)

    void "findTemperature should return expected temperature"() {
        given:
        // We define a BlockingVariable to store the result in the callback,
        // so we can wait for the value in the then: block and
        // asssert the value when it becomes available.
        def result = new BlockingVariable<Async.Result>()

        when:
        // We invoke the async method and in the callback use
        // our BlockingVariable to set the result.
        async.findTemperature("Tilburg") { Async.Result temp ->
            // Set the result to the BlockingVariable.
            result.set(temp)
        }

        then:
        // Now we wait until the result is available with the
        // blocking call get().
        // Default waiting time is 1 second. We can change that
        // by providing the number of seconds as argument
        // to the BlockingVariable constructor.
        // E.g. new BlockingVariable<Long>(3) to wait for 3 seconds.
        result.get() == new Async.Result("Tilburg", 21)
    }

    void "findTemperatures should return expected temperatures"() {
        given:
        // With type BlockingVariables we can wait for multiple values.
        // Each value must be assigned to a unique key.
        def result = new BlockingVariables(5)

        when:
        async.findTemperatures(["Tilburg", "Amsterdam"]) { Async.Result temp ->
            // Set the result with a key to the BlockingVariables result variable.
            // We can story multiple results in one BlockingVariables.
            result[temp.city()] = temp.temperature()
        }

        then:
        // We wait for the results key by key.
        // We cannot rely that the result are available in the
        // same order as the passed input arguments Tilburg and Amsterdam
        // as the call will be asynchronous.
        // But using BlockingVariables we dont' have to care,
        // we simply request the value for a key and the code will
        // block until it is available.
        result["Amsterdam"] == 18
        result["Tilburg"] == 21
    }
}

package mrhaki;

import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;

class AsyncWeather {

    private final ExecutorService executorService;
    private final WeatherService weatherService;

    private final Map<String, Integer> results = new ConcurrentHashMap<>();

    AsyncWeather(ExecutorService executorService, WeatherService weatherService) {
        this.executorService = executorService;
        this.weatherService = weatherService;
    }

    // Invoke the WeatherService in a new thread and store result in results.
    void findTemperature(String city) {
        executorService.submit(() -> results.put(city, weatherService.getTemperature(city)));
    }

    // Invoke the WeatherService in a new thread for each city and store result in results.
    void findTemperatures(String... cities) {
        Arrays.stream(cities)
              .parallel()
              .forEach(this::findTemperature);
    }

    // Get the temperature. Value can be null when the WeatherService call is not finished yet.
    int getTemperature(String city) {
        return results.get(city);
    }

    interface WeatherService {
        int getTemperature(String city);
    }
}

package mrhaki

import spock.lang.Specification
import spock.lang.Subject
import spock.util.concurrent.PollingConditions

import java.util.concurrent.Executors

class AsyncPollingSpec extends Specification {

    // Provide a stub for the WeatherService interface.
    // Return 21 when city is Tilburg and 18 for other cities.
    private AsyncWeather.WeatherService weatherService = Stub() {
        getTemperature(_ as String) >> { args ->
            if ("Tilburg" == args[0]) 21 else 18
        }
    }

    // We want to test the class AsyncWeather
    @Subject
    private AsyncWeather async = new AsyncWeather(Executors.newFixedThreadPool(2), weatherService)

    void "findTemperature and getTemperature should return expected temperature"() {
        when:
        // We invoke the async method.
        async.findTemperature("Tilburg")

        then:
        // Now we wait until the results are set.
        // By default we wait for at  most 1 second,
        // but we can configure some extra properties like
        // timeout, delay, initial delay and factor to increase delays.
        // E.g. new PollingConditions(timeout: 5, initialDelay: 0.5, delay: 0.5)
        new PollingConditions().eventually {
            // Although we are in a then block, we must
            // use the assert keyword in our eventually
            // Closure code.
            assert async.getTemperature("Tilburg") == 21
        }
    }

    void "findTemperatures and getTemperature shoud return expected temperatures"() {
        when:
        // We invoke the async method.
        async.findTemperatures("Tilburg", "Amsterdam")

        then:
        // Instead of using eventually we can use within
        // with a given timeout we want the conditions to
        // be available for assertions.
        new PollingConditions().within(3) {
            // Although we are in a then block, we must
            // use the assert keyword in our within
            // Closure code.
            assert async.getTemperature("Amsterdam") == 18
            assert async.getTemperature("Tilburg") == 21
        }
    }
}