/*
 * Copyright 2019 LINE Corporation
 *
 * LINE Corporation licenses this file to you under the Apache License,
 * version 2.0 (the "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at:
 *
 *   https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */
package com.linecorp.armeria.client.endpoint.healthcheck;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.awaitility.Awaitility.await;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;

import org.junit.jupiter.api.Test;

import com.google.common.collect.ImmutableList;

import com.linecorp.armeria.client.Endpoint;
import com.linecorp.armeria.client.endpoint.DynamicEndpointGroup;
import com.linecorp.armeria.common.util.AsyncCloseable;

class HealthCheckedEndpointGroupTest {
    @Test
    void disappearedEndpoint() {
        // Start with an endpoint group that has healthy 'foo'.
        final MockEndpointGroup delegate = new MockEndpointGroup();
        delegate.set(Endpoint.of("foo"));
        final AtomicReference<HealthCheckerContext> ctxCapture = new AtomicReference<>();

        try (HealthCheckedEndpointGroup group = new AbstractHealthCheckedEndpointGroupBuilder(delegate) {
            @Override
            protected Function<? super HealthCheckerContext, ? extends AsyncCloseable> newCheckerFactory() {
                return ctx -> {
                    ctxCapture.set(ctx);
                    ctx.updateHealth(1);
                    return () -> CompletableFuture.completedFuture(null);
                };
            }
        }.build()) {

            // Check the initial state.
            final HealthCheckerContext ctx = ctxCapture.get();
            assertThat(ctx).isNotNull();
            assertThat(ctx.endpoint()).isEqualTo(Endpoint.of("foo"));

            // 'foo' did not disappear yet, so the task must be accepted and run.
            final AtomicBoolean taskRun = new AtomicBoolean();
            ctx.executor().execute(() -> taskRun.set(true));
            await().untilAsserted(() -> assertThat(taskRun).isTrue());

            // Make 'foo' disappear.
            delegate.set();

            // 'foo' should not be healthy anymore.
            assertThat(group.endpoints()).isEmpty();

            // 'foo' should not be healthy even if `ctx.updateHealth()` was called.
            ctx.updateHealth(1);
            assertThat(group.endpoints()).isEmpty();
            assertThat(group.healthyEndpoints).isEmpty();

            // An attempt to schedule a new task for a disappeared endpoint must fail.
            assertThatThrownBy(() -> ctx.executor().execute(() -> {}))
                    .isInstanceOf(RejectedExecutionException.class)
                    .hasMessageContaining("destroyed");
        }
    }

    private static final class MockEndpointGroup extends DynamicEndpointGroup {
        void set(Endpoint... endpoints) {
            setEndpoints(ImmutableList.copyOf(endpoints));
        }
    }
}

@@ -854,24 +854,47 @@
 ity.await;%0A%0A
+import java.util.List;%0A
 import java.
@@ -928,16 +928,60 @@
 Future;%0A
+import java.util.concurrent.CountDownLatch;%0A
 import j
@@ -1028,16 +1028,54 @@
 eption;%0A
+import java.util.concurrent.TimeUnit;%0A
 import j
@@ -1401,16 +1401,64 @@
 tGroup;%0A
+import com.linecorp.armeria.common.CommonPools;%0A
 import c
@@ -1510,46 +1510,1845 @@
 e;%0A%0A
-class HealthCheckedEndpointGroupTest %7B
+import io.netty.channel.EventLoopGroup;%0A%0Aclass HealthCheckedEndpointGroupTest %7B%0A%0A    @Test%0A    void delegateUpdateCandidatesWhileCreatingHealthCheckedEndpointGroup() %7B%0A        final MockEndpointGroup delegate = new MockEndpointGroup();%0A        final CompletableFuture%3CList%3CEndpoint%3E%3E future = delegate.initialEndpointsFuture();%0A        future.complete(ImmutableList.of(Endpoint.of(%22127.0.0.1%22, 8080), Endpoint.of(%22127.0.0.1%22, 8081)));%0A%0A        final CountDownLatch latch = new CountDownLatch(1);%0A%0A        // Schedule the task which update the endpoint one second later to ensure that the change is happening%0A        // while creating the HealthCheckedEndpointGroup.%0A        final EventLoopGroup executors = CommonPools.workerGroup();%0A        executors.schedule(%0A                () -%3E %7B%0A                    delegate.set(Endpoint.of(%22127.0.0.1%22, 8082));%0A                    latch.countDown();%0A                %7D, 1, TimeUnit.SECONDS);%0A%0A        new AbstractHealthCheckedEndpointGroupBuilder(delegate) %7B%0A            @Override%0A            protected Function%3C? super HealthCheckerContext, ? extends AsyncCloseable%3E newCheckerFactory() %7B%0A                return (Function%3CHealthCheckerContext, AsyncCloseable%3E) ctx -%3E %7B%0A                    // Call updateHealth after the endpoint is changed so that%0A                    // snapshot.forEach(ctx -%3E ctx.initialCheckFuture.join()); performs the next action.%0A                    executors.schedule(() -%3E %7B%0A                        try %7B%0A                            latch.await();%0A                        %7D catch (InterruptedException e) %7B%0A                            // Ignore%0A                        %7D%0A                        ctx.updateHealth(1);%0A                    %7D, 2, TimeUnit.SECONDS);%0A                    return CompletableFuture::new;%0A                %7D;%0A            %7D%0A        %7D.build();%0A    %7D%0A
 %0A   
@@ -5180,16 +5180,29 @@
 (() -%3E %7B
+%0A            
 %7D))%0A    

Add a test for HealthCheckedEndpointGroup raising ConcurrentMod… (#1992)