/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF 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
 *
 *     http://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 org.apache.dubbo.rpc.protocol.dubbo;

import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.utils.DubboAppender;
import org.apache.dubbo.common.utils.LogUtil;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.common.utils.ReflectUtils;
import org.apache.dubbo.remoting.exchange.ExchangeClient;
import org.apache.dubbo.rpc.Exporter;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.ProxyFactory;
import org.apache.dubbo.rpc.protocol.AsyncToSyncInvoker;
import org.apache.dubbo.rpc.protocol.dubbo.support.ProtocolUtils;

import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Objects;

import static org.apache.dubbo.remoting.Constants.CONNECTIONS_KEY;
import static org.apache.dubbo.rpc.protocol.dubbo.Constants.LAZY_REQUEST_WITH_WARNING_KEY;
import static org.apache.dubbo.rpc.protocol.dubbo.Constants.SHARE_CONNECTIONS_KEY;


public class ReferenceCountExchangeClientTest {

    public static ProxyFactory proxy = ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();
    private static DubboProtocol protocol = DubboProtocol.getDubboProtocol();
    Exporter<?> demoExporter;
    Exporter<?> helloExporter;
    Invoker<IDemoService> demoServiceInvoker;
    Invoker<IHelloService> helloServiceInvoker;
    IDemoService demoService;
    IHelloService helloService;
    ExchangeClient demoClient;
    ExchangeClient helloClient;
    String errorMsg = "safe guard client , should not be called ,must have a bug";

    @BeforeAll
    public static void setUpBeforeClass() throws Exception {
    }

    @AfterAll
    public static void tearDownAfterClass() {
        ProtocolUtils.closeAll();
    }

    public static Invoker<?> referInvoker(Class<?> type, URL url) {
        return (Invoker<?>) protocol.refer(type, url);
    }

    public static <T> Exporter<T> export(T instance, Class<T> type, String url) {
        return export(instance, type, URL.valueOf(url));
    }

    public static <T> Exporter<T> export(T instance, Class<T> type, URL url) {
        return protocol.export(proxy.getInvoker(instance, type, url));
    }

    @BeforeEach
    public void setUp() throws Exception {
    }

    /**
     * test connection sharing
     */
    @Test
    public void test_share_connect() {
        init(0, 1);
        Assertions.assertEquals(demoClient.getLocalAddress(), helloClient.getLocalAddress());
        Assertions.assertEquals(demoClient, helloClient);
        destoy();
    }

    /**
     * test connection not sharing
     */
    @Test
    public void test_not_share_connect() {
        init(1, 1);
        Assertions.assertNotSame(demoClient.getLocalAddress(), helloClient.getLocalAddress());
        Assertions.assertNotSame(demoClient, helloClient);
        destoy();
    }

    /**
     * test using multiple shared connections
     */
    @Test
    public void test_mult_share_connect() {
        // here a three shared connection is established between a consumer process and a provider process.
        final int shareConnectionNum = 3;

        init(0, shareConnectionNum);

        List<ReferenceCountExchangeClient> helloReferenceClientList = getReferenceClientList(helloServiceInvoker);
        Assertions.assertEquals(shareConnectionNum, helloReferenceClientList.size());

        List<ReferenceCountExchangeClient> demoReferenceClientList = getReferenceClientList(demoServiceInvoker);
        Assertions.assertEquals(shareConnectionNum, demoReferenceClientList.size());

        // because helloServiceInvoker and demoServiceInvoker use share connect， so client list must be equal
        Assertions.assertEquals(helloReferenceClientList, demoReferenceClientList);

        Assertions.assertEquals(demoClient.getLocalAddress(), helloClient.getLocalAddress());
        Assertions.assertEquals(demoClient, helloClient);

        destoy();
    }

    /**
     * test counter won't count down incorrectly when invoker is destroyed for multiple times
     */
    @Test
    public void test_multi_destory() {
        init(0, 1);
        DubboAppender.doStart();
        DubboAppender.clear();
        demoServiceInvoker.destroy();
        demoServiceInvoker.destroy();
        Assertions.assertEquals("hello", helloService.hello());
        Assertions.assertEquals(0, LogUtil.findMessage(errorMsg), "should not  warning message");
        LogUtil.checkNoError();
        DubboAppender.doStop();
        destoy();
    }

    /**
     * Test against invocation still succeed even if counter has error
     */
    @Test
    public void test_counter_error() {
        init(0, 1);
        DubboAppender.doStart();
        DubboAppender.clear();

        // because the two interfaces are initialized, the ReferenceCountExchangeClient reference counter is 2
        ReferenceCountExchangeClient client = getReferenceClient(helloServiceInvoker);

        // close once, counter counts down from 2 to 1, no warning occurs
        client.close();
        Assertions.assertEquals("hello", helloService.hello());
        Assertions.assertEquals(0, LogUtil.findMessage(errorMsg), "should not warning message");

        // close twice, counter counts down from 1 to 0, no warning occurs
        client.close();

        // wait close done.
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            Assertions.fail();
        }

        // client has been replaced with lazy client, close status is false because a new lazy client's exchange client is null.
        Assertions.assertFalse(client.isClosed(), "client status close");
        // invoker status is available because the default value of associated lazy client's initial state is true.
        Assertions.assertTrue(helloServiceInvoker.isAvailable(), "invoker status unavailable");

        // due to the effect of LazyConnectExchangeClient, the client will be "revived" whenever there is a call.
        Assertions.assertEquals("hello", helloService.hello());
        Assertions.assertEquals(1, LogUtil.findMessage(errorMsg), "should warning message");

        // output one error every 5000 invocations.
        Assertions.assertEquals("hello", helloService.hello());
        Assertions.assertEquals(1, LogUtil.findMessage(errorMsg), "should warning message");

        DubboAppender.doStop();

        /**
         * This is the third time to close the same client. Under normal circumstances,
         * a client value should be closed once (that is, the shutdown operation is irreversible).
         * After closing, the value of the reference counter of the client has become -1.
         *
         * But this is a bit special, because after the client is closed twice, there are several calls to helloService,
         * that is, the client inside the ReferenceCountExchangeClient is actually active, so the third shutdown here is still effective,
         * let the resurrection After the client is really closed.
         */
        client.close();

        // close status is false because the lazy client's exchange client is null again after close().
        Assertions.assertFalse(client.isClosed(), "client status close");
        // invoker status is available because the default value of associated lazy client's initial state is true.
        Assertions.assertTrue(helloServiceInvoker.isAvailable(), "invoker status unavailable");

        // revive: initial the lazy client's exchange client again.  
        Assertions.assertEquals("hello", helloService.hello());

        destoy();
    }

    @SuppressWarnings("unchecked")
    private void init(int connections, int shareConnections) {
        Assertions.assertTrue(connections >= 0);
        Assertions.assertTrue(shareConnections >= 1);

        int port = NetUtils.getAvailablePort();
        String params = CONNECTIONS_KEY + "=" + connections
                + "&" + SHARE_CONNECTIONS_KEY + "=" + shareConnections
                + "&" + LAZY_REQUEST_WITH_WARNING_KEY + "=" + "true";
        URL demoUrl = URL.valueOf("dubbo://127.0.0.1:" + port + "/demo?" + params);
        URL helloUrl = URL.valueOf("dubbo://127.0.0.1:" + port + "/hello?" + params);

        demoExporter = export(new DemoServiceImpl(), IDemoService.class, demoUrl);
        helloExporter = export(new HelloServiceImpl(), IHelloService.class, helloUrl);

        demoServiceInvoker = (Invoker<IDemoService>) referInvoker(IDemoService.class, demoUrl);
        demoService = proxy.getProxy(demoServiceInvoker);
        Assertions.assertEquals("demo", demoService.demo());

        helloServiceInvoker = (Invoker<IHelloService>) referInvoker(IHelloService.class, helloUrl);
        helloService = proxy.getProxy(helloServiceInvoker);
        Assertions.assertEquals("hello", helloService.hello());

        demoClient = getClient(demoServiceInvoker);
        helloClient = getClient(helloServiceInvoker);
    }

    private void destoy() {
        demoServiceInvoker.destroy();
        helloServiceInvoker.destroy();
        demoExporter.getInvoker().destroy();
        helloExporter.getInvoker().destroy();
    }

    private ExchangeClient getClient(Invoker<?> invoker) {
        if (invoker.getUrl().getParameter(CONNECTIONS_KEY, 1) == 1) {
            return getInvokerClient(invoker);
        } else {
            ReferenceCountExchangeClient client = getReferenceClient(invoker);
            try {
                Field clientField = ReferenceCountExchangeClient.class.getDeclaredField("client");
                ReflectUtils.makeAccessible(clientField);
                return (ExchangeClient) clientField.get(client);
            } catch (Exception e) {
                e.printStackTrace();
                Assertions.fail(e.getMessage());
                throw new RuntimeException(e);
            }
        }
    }

    private ReferenceCountExchangeClient getReferenceClient(Invoker<?> invoker) {
        return getReferenceClientList(invoker).get(0);
    }

    private List<ReferenceCountExchangeClient> getReferenceClientList(Invoker<?> invoker) {
        List<ExchangeClient> invokerClientList = getInvokerClientList(invoker);

        List<ReferenceCountExchangeClient> referenceCountExchangeClientList = new ArrayList<>(invokerClientList.size());
        for (ExchangeClient exchangeClient : invokerClientList) {
            Assertions.assertTrue(exchangeClient instanceof ReferenceCountExchangeClient);
            referenceCountExchangeClientList.add((ReferenceCountExchangeClient) exchangeClient);
        }

        return referenceCountExchangeClientList;
    }

    private ExchangeClient getInvokerClient(Invoker<?> invoker) {
        return getInvokerClientList(invoker).get(0);
    }

    private List<ExchangeClient> getInvokerClientList(Invoker<?> invoker) {
        @SuppressWarnings("rawtypes") DubboInvoker dInvoker = (DubboInvoker) ((AsyncToSyncInvoker) invoker).getInvoker();
        try {
            Field clientField = DubboInvoker.class.getDeclaredField("clients");
            ReflectUtils.makeAccessible(clientField);
            ExchangeClient[] clients = (ExchangeClient[]) clientField.get(dInvoker);

            List<ExchangeClient> clientList = new ArrayList<ExchangeClient>(clients.length);
            for (ExchangeClient client : clients) {
                clientList.add(client);
            }

            // sorting makes it easy to compare between lists
            Collections.sort(clientList, Comparator.comparing(c -> Integer.valueOf(Objects.hashCode(c))));

            return clientList;

        } catch (Exception e) {
            e.printStackTrace();
            Assertions.fail(e.getMessage());
            throw new RuntimeException(e);
        }
    }

    public interface IDemoService {
        String demo();
    }

    public interface IHelloService {
        String hello();
    }

    public class DemoServiceImpl implements IDemoService {
        public String demo() {
            return "demo";
        }
    }

    public class HelloServiceImpl implements IHelloService {
        public String hello() {
            return "hello";
        }
    }
}
