/*
 * Copyright 1999-2020 Alibaba Group Holding Ltd.
 *
 * Licensed 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 com.alibaba.nacos.common.remote.client;

import com.alibaba.nacos.api.ability.constant.AbilityKey;
import com.alibaba.nacos.api.ability.constant.AbilityStatus;
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.remote.RequestCallBack;
import com.alibaba.nacos.api.remote.request.ClientDetectionRequest;
import com.alibaba.nacos.api.remote.request.ConnectResetRequest;
import com.alibaba.nacos.api.remote.request.HealthCheckRequest;
import com.alibaba.nacos.api.remote.request.Request;
import com.alibaba.nacos.api.remote.response.ClientDetectionResponse;
import com.alibaba.nacos.api.remote.response.ConnectResetResponse;
import com.alibaba.nacos.api.remote.response.ErrorResponse;
import com.alibaba.nacos.api.remote.response.HealthCheckResponse;
import com.alibaba.nacos.api.remote.response.Response;
import com.alibaba.nacos.common.remote.ConnectionType;
import com.alibaba.nacos.common.remote.client.grpc.DefaultGrpcClientConfig;
import com.alibaba.nacos.common.remote.client.grpc.GrpcConnection;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.stubbing.Answer;

import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

@ExtendWith(MockitoExtension.class)
class RpcClientTest {
    
    RpcClient rpcClient;
    
    Field serverListFactoryField;
    
    Field reconnectionSignalField;
    
    Method resolveServerInfoMethod;
    
    Method healthCheck;
    
    Answer<?> runAsSync;
    
    Answer<?> notInvoke;
    
    @Mock
    ServerListFactory serverListFactory;
    
    @Mock
    Connection connection;
    
    RpcClientConfig rpcClientConfig;
    
    @BeforeEach
    void setUp() throws NoSuchFieldException, IllegalAccessException, NoSuchMethodException, InvocationTargetException {
        rpcClientConfig = spy(new RpcClientConfig() {
            @Override
            public String name() {
                return "test";
            }
            
            @Override
            public int retryTimes() {
                return 1;
            }
            
            @Override
            public long timeOutMills() {
                return 3000L;
            }
            
            @Override
            public long connectionKeepAlive() {
                return 5000L;
            }
            
            @Override
            public int healthCheckRetryTimes() {
                return 1;
            }
            
            @Override
            public long healthCheckTimeOut() {
                return 3000L;
            }
            
            @Override
            public Map<String, String> labels() {
                return new HashMap<>();
            }
        });
        rpcClient = spy(new RpcClient(rpcClientConfig) {
            @Override
            public ConnectionType getConnectionType() {
                return null;
            }
            
            @Override
            public int rpcPortOffset() {
                return 0;
            }
            
            @Override
            public Connection connectToServer(ServerInfo serverInfo) {
                return null;
            }
        });
        
        serverListFactoryField = RpcClient.class.getDeclaredField("serverListFactory");
        serverListFactoryField.setAccessible(true);
        
        reconnectionSignalField = RpcClient.class.getDeclaredField("reconnectionSignal");
        reconnectionSignalField.setAccessible(true);
        
        Method getDeclaredFields0 = Class.class.getDeclaredMethod("getDeclaredFields0", boolean.class);
        getDeclaredFields0.setAccessible(true);
        Field[] fields = (Field[]) getDeclaredFields0.invoke(Field.class, false);
        Field modifiersField1 = null;
        for (Field each : fields) {
            if ("modifiers".equals(each.getName())) {
                modifiersField1 = each;
            }
        }
        if (modifiersField1 != null) {
            modifiersField1.setAccessible(true);
            modifiersField1.setInt(reconnectionSignalField, reconnectionSignalField.getModifiers() & ~Modifier.FINAL);
        }
        
        resolveServerInfoMethod = RpcClient.class.getDeclaredMethod("resolveServerInfo", String.class);
        resolveServerInfoMethod.setAccessible(true);
        
        healthCheck = RpcClient.class.getDeclaredMethod("healthCheck");
        healthCheck.setAccessible(true);
        
        runAsSync = invocationOnMock -> {
            Runnable runnable = (Runnable) invocationOnMock.getArguments()[0];
            runnable.run();
            return null;
        };
        
        notInvoke = invocationOnMock -> null;
    }
    
    @AfterEach
    void tearDown() throws IllegalAccessException, NacosException {
        rpcClientConfig.labels().clear();
        rpcClient.rpcClientStatus.set(RpcClientStatus.WAIT_INIT);
        serverListFactoryField.set(rpcClient, null);
        ((Queue<?>) reconnectionSignalField.get(rpcClient)).clear();
        rpcClient.currentConnection = null;
        System.clearProperty("nacos.server.port");
        rpcClient.eventLinkedBlockingQueue.clear();
        rpcClient.shutdown();
    }
    
    @Test
    void testInitServerListFactory() {
        rpcClient.rpcClientStatus.set(RpcClientStatus.WAIT_INIT);
        rpcClient.serverListFactory(serverListFactory);
        assertEquals(RpcClientStatus.INITIALIZED, rpcClient.rpcClientStatus.get());
        
        rpcClient.rpcClientStatus.set(RpcClientStatus.INITIALIZED);
        rpcClient.serverListFactory(serverListFactory);
        assertEquals(RpcClientStatus.INITIALIZED, rpcClient.rpcClientStatus.get());
        
        RpcClient client1 = new RpcClient(new RpcClientConfig() {
            @Override
            public String name() {
                return "test";
            }
            
            @Override
            public int retryTimes() {
                return 3;
            }
            
            @Override
            public long timeOutMills() {
                return 3000L;
            }
            
            @Override
            public long connectionKeepAlive() {
                return 5000L;
            }
            
            @Override
            public int healthCheckRetryTimes() {
                return 1;
            }
            
            @Override
            public long healthCheckTimeOut() {
                return 3000L;
            }
            
            @Override
            public Map<String, String> labels() {
                return new HashMap<>();
            }
        }, serverListFactory) {
            @Override
            public ConnectionType getConnectionType() {
                return null;
            }
            
            @Override
            public int rpcPortOffset() {
                return 0;
            }
            
            @Override
            public Connection connectToServer(ServerInfo serverInfo) {
                return null;
            }
        };
        assertEquals(RpcClientStatus.INITIALIZED, client1.rpcClientStatus.get());
        
        RpcClient client2 = new RpcClient(rpcClientConfig, serverListFactory) {
            @Override
            public ConnectionType getConnectionType() {
                return null;
            }
            
            @Override
            public int rpcPortOffset() {
                return 0;
            }
            
            @Override
            public Connection connectToServer(ServerInfo serverInfo) {
                return null;
            }
        };
        assertEquals(RpcClientStatus.INITIALIZED, client2.rpcClientStatus.get());
    }
    
    @Test
    void testLabels() {
        when(rpcClientConfig.labels()).thenReturn(Collections.singletonMap("labelKey1", "labelValue1"));
        Map.Entry<String, String> element = rpcClient.getLabels().entrySet().iterator().next();
        assertEquals("labelKey1", element.getKey());
        assertEquals("labelValue1", element.getValue());
        
        // accumulate labels
        Map<String, String> map = new HashMap<>();
        map.put("labelKey2", "labelValue2");
        when(rpcClientConfig.labels()).thenReturn(map);
        assertEquals(1, rpcClient.getLabels().size());
        assertEquals("test", rpcClient.getName());
    }
    
    @Test
    void testOnServerListChangeWhenCurrentConnectionIsNullThenDoNothing() throws IllegalAccessException {
        int beforeSize = ((Queue<?>) reconnectionSignalField.get(rpcClient)).size();
        rpcClient.serverListFactory(serverListFactory);
        
        rpcClient.onServerListChange();
        
        int afterSize = ((Queue<?>) reconnectionSignalField.get(rpcClient)).size();
        assertEquals(beforeSize, afterSize);
    }
    
    @Test
    void testOnServerListChangeWhenServiceInfoIsNullThenDoNothing() throws IllegalAccessException {
        int beforeSize = ((Queue<?>) reconnectionSignalField.get(rpcClient)).size();
        rpcClient.currentConnection = mock(Connection.class);
        
        rpcClient.onServerListChange();
        
        int afterSize = ((Queue<?>) reconnectionSignalField.get(rpcClient)).size();
        assertEquals(beforeSize, afterSize);
    }
    
    @Test
    void testOnServerListChangeWhenCurrentConnectionIsNotInServerListThenSwitchServerAsync() throws IllegalAccessException {
        final int beforeSize = ((Queue<?>) reconnectionSignalField.get(rpcClient)).size();
        rpcClient.serverListFactory(serverListFactory);
        rpcClient.currentConnection = new GrpcConnection(new RpcClient.ServerInfo("10.10.10.10", 8848), null);
        doReturn(Collections.singletonList("")).when(serverListFactory).getServerList();
        when(serverListFactory.getServerList()).thenReturn(Collections.singletonList("127.0.0.1"));
        rpcClient.onServerListChange();
        
        int afterSize = ((Queue<?>) reconnectionSignalField.get(rpcClient)).size();
        assertEquals(beforeSize + 1, afterSize);
    }
    
    @Test
    void testOnServerListChangeWhenCurrentConnectionIsInServerListThenDoNothing() throws IllegalAccessException {
        final int beforeSize = ((Queue<?>) reconnectionSignalField.get(rpcClient)).size();
        rpcClient.serverListFactory(serverListFactory);
        rpcClient.currentConnection = new GrpcConnection(new RpcClient.ServerInfo("10.10.10.10", 8848), null);
        doReturn(Collections.singletonList("http://10.10.10.10:8848")).when(serverListFactory).getServerList();
        rpcClient.onServerListChange();
        
        int afterSize = ((Queue<?>) reconnectionSignalField.get(rpcClient)).size();
        assertEquals(beforeSize, afterSize);
    }
    
    @Test
    void testResolveServerInfo1() throws InvocationTargetException, IllegalAccessException {
        assertEquals("10.10.10.10:8848",
                ((RpcClient.ServerInfo) resolveServerInfoMethod.invoke(rpcClient, "10.10.10.10::8848")).getAddress());
        assertEquals("10.10.10.10:8848",
                ((RpcClient.ServerInfo) resolveServerInfoMethod.invoke(rpcClient, "10.10.10.10:8848")).getAddress());
        assertEquals("10.10.10.10:8848",
                ((RpcClient.ServerInfo) resolveServerInfoMethod.invoke(rpcClient, "http://10.10.10.10:8848")).getAddress());
        assertEquals("10.10.10.10:8848",
                ((RpcClient.ServerInfo) resolveServerInfoMethod.invoke(rpcClient, "http://10.10.10.10::8848")).getAddress());
        assertEquals("10.10.10.10:8848",
                ((RpcClient.ServerInfo) resolveServerInfoMethod.invoke(rpcClient, "http://10.10.10.10")).getAddress());
        assertEquals("10.10.10.10:8848",
                ((RpcClient.ServerInfo) resolveServerInfoMethod.invoke(rpcClient, "https://10.10.10.10::8848")).getAddress());
    }
    
    @Test
    void testResolveServerInfo2() throws InvocationTargetException, IllegalAccessException {
        System.setProperty("nacos.server.port", "4424");
        assertEquals("10.10.10.10:4424",
                ((RpcClient.ServerInfo) resolveServerInfoMethod.invoke(rpcClient, "http://10.10.10.10")).getAddress());
    }
    
    @Test
    void testRequestSuccess() throws NacosException, NoSuchFieldException, IllegalAccessException {
        rpcClient.currentConnection = connection;
        rpcClient.rpcClientStatus.set(RpcClientStatus.RUNNING);
        when(connection.request(any(), anyLong())).thenReturn(new HealthCheckResponse());
        Field lastActiveTimeStampField = RpcClient.class.getDeclaredField("lastActiveTimeStamp");
        lastActiveTimeStampField.setAccessible(true);
        final long lastActiveTimeStamp = (long) lastActiveTimeStampField.get(rpcClient);
        Response response = rpcClient.request(new HealthCheckRequest());
        assertTrue(response instanceof HealthCheckResponse);
        assertTrue(lastActiveTimeStamp <= (long) lastActiveTimeStampField.get(rpcClient));
    }
    
    @Test
    void testRequestWithoutAnyTry() throws NacosException {
        assertThrows(NacosException.class, () -> {
            when(rpcClientConfig.retryTimes()).thenReturn(-1);
            rpcClient.request(null);
        });
    }
    
    @Test
    void testRequestWhenClientAlreadyShutDownThenThrowException() throws NacosException {
        assertThrows(NacosException.class, () -> {
            rpcClient.rpcClientStatus.set(RpcClientStatus.SHUTDOWN);
            rpcClient.currentConnection = connection;
            rpcClient.request(null);
        });
    }
    
    @Test
    void testRequestWhenTimeoutThenThrowException() throws NacosException {
        assertThrows(NacosException.class, () -> {
            rpcClient.rpcClientStatus.set(RpcClientStatus.RUNNING);
            rpcClient.currentConnection = connection;
            doReturn(null).when(connection).request(any(), anyLong());
            rpcClient.request(null, 10000);
        });
    }
    
    @Test
    void testRequestWhenResponseErrorThenThrowException() throws NacosException {
        assertThrows(NacosException.class, () -> {
            rpcClient.rpcClientStatus.set(RpcClientStatus.RUNNING);
            rpcClient.currentConnection = connection;
            doReturn(new ErrorResponse()).when(connection).request(any(), anyLong());
            rpcClient.request(null, 10000);
        });
    }
    
    @Test
    void testRequestWhenResponseUnregisterThenSwitchServer() throws NacosException {
        rpcClient.rpcClientStatus.set(RpcClientStatus.RUNNING);
        rpcClient.currentConnection = connection;
        ErrorResponse errorResponse = new ErrorResponse();
        errorResponse.setErrorCode(NacosException.UN_REGISTER);
        doReturn(errorResponse).when(connection).request(any(), anyLong());
        Exception exception = null;
        
        try {
            rpcClient.request(mock(Request.class), 10000);
        } catch (Exception e) {
            exception = e;
        }
        
        assertEquals(RpcClientStatus.UNHEALTHY, rpcClient.rpcClientStatus.get());
        verify(rpcClient).switchServerAsync();
        assertNotNull(exception);
    }
    
    @Test
    void testAsyncRequestSuccess() throws NacosException {
        rpcClient.currentConnection = connection;
        rpcClient.rpcClientStatus.set(RpcClientStatus.RUNNING);
        RequestCallBack<?> requestCallBack = mock(RequestCallBack.class);
        when(requestCallBack.getTimeout()).thenReturn(1000L);
        rpcClient.asyncRequest(null, requestCallBack);
        verify(connection).asyncRequest(any(), any());
    }
    
    @Test
    void testAsyncRequestWithoutAnyTry() throws NacosException {
        assertThrows(NacosException.class, () -> {
            when(rpcClientConfig.retryTimes()).thenReturn(-1);
            rpcClient.asyncRequest(null, null);
        });
    }
    
    @Test
    void testAsyncRequestWhenClientAlreadyShutDownThenThrowException() throws NacosException {
        assertThrows(NacosException.class, () -> {
            rpcClient.rpcClientStatus.set(RpcClientStatus.SHUTDOWN);
            rpcClient.currentConnection = connection;
            RequestCallBack<?> requestCallBack = mock(RequestCallBack.class);
            doReturn(10000L).when(requestCallBack).getTimeout();
            rpcClient.asyncRequest(null, requestCallBack);
        });
    }
    
    @Test
    void testAsyncRequestWhenSendRequestFailedMannyTimesThenSwitchServer() throws NacosException {
        rpcClient.rpcClientStatus.set(RpcClientStatus.RUNNING);
        rpcClient.currentConnection = connection;
        doThrow(new NacosException()).when(connection).asyncRequest(any(), any());
        RequestCallBack<?> requestCallBack = mock(RequestCallBack.class);
        doReturn(10000L).when(requestCallBack).getTimeout();
        Exception exception = null;
        
        try {
            rpcClient.asyncRequest(null, requestCallBack);
        } catch (NacosException e) {
            exception = e;
        }
        
        verify(connection, atLeastOnce()).asyncRequest(any(), any());
        verify(rpcClient).switchServerAsyncOnRequestFail();
        assertNotNull(exception);
        assertEquals(RpcClientStatus.UNHEALTHY, rpcClient.rpcClientStatus.get());
    }
    
    @Test
    void testRequestFutureWithoutAnyTry() throws NacosException {
        assertThrows(NacosException.class, () -> {
            when(rpcClientConfig.retryTimes()).thenReturn(-1);
            rpcClient.requestFuture(null);
        });
    }
    
    @Test
    void testRequestFutureWhenClientAlreadyShutDownThenThrowException() throws NacosException {
        assertThrows(NacosException.class, () -> {
            rpcClient.rpcClientStatus.set(RpcClientStatus.SHUTDOWN);
            rpcClient.currentConnection = connection;
            rpcClient.requestFuture(null);
        });
    }
    
    @Test
    void testRequestFutureWhenRetryReachMaxRetryTimesThenSwitchServer() throws NacosException {
        when(rpcClientConfig.timeOutMills()).thenReturn(5000L);
        when(rpcClientConfig.retryTimes()).thenReturn(3);
        rpcClient.rpcClientStatus.set(RpcClientStatus.RUNNING);
        rpcClient.currentConnection = connection;
        doThrow(NacosException.class).when(connection).requestFuture(any());
        Exception exception = null;
        
        try {
            rpcClient.requestFuture(null);
        } catch (NacosException e) {
            exception = e;
        }
        
        verify(connection, times(4)).requestFuture(any());
        verify(rpcClient).switchServerAsyncOnRequestFail();
        assertNotNull(exception);
        assertEquals(RpcClientStatus.UNHEALTHY, rpcClient.rpcClientStatus.get());
    }
    
    @Test
    void testRpcClientShutdownWhenClientDidntStart() throws NacosException {
        RpcClient rpcClient = new RpcClient(new RpcClientConfig() {
            @Override
            public String name() {
                return "test-client";
            }
            
            @Override
            public int retryTimes() {
                return 3;
            }
            
            @Override
            public long timeOutMills() {
                return 3000L;
            }
            
            @Override
            public long connectionKeepAlive() {
                return 5000L;
            }
            
            @Override
            public int healthCheckRetryTimes() {
                return 1;
            }
            
            @Override
            public long healthCheckTimeOut() {
                return 3000L;
            }
            
            @Override
            public Map<String, String> labels() {
                return new HashMap<>();
            }
        }) {
            @Override
            public ConnectionType getConnectionType() {
                return null;
            }
            
            @Override
            public int rpcPortOffset() {
                return 0;
            }
            
            @Override
            public Connection connectToServer(ServerInfo serverInfo) {
                return null;
            }
        };
        
        rpcClient.shutdown();
        assertTrue(rpcClient.isShutdown());
    }
    
    @Test
    void testHealthCheck() throws IllegalAccessException, NacosException {
        Random random = new Random();
        int retry = random.nextInt(10);
        when(rpcClientConfig.healthCheckRetryTimes()).thenReturn(retry);
        rpcClient.rpcClientStatus.set(RpcClientStatus.RUNNING);
        rpcClient.currentConnection = connection;
        doThrow(new NacosException()).when(connection).request(any(), anyLong());
        try {
            healthCheck.invoke(rpcClient);
        } catch (InvocationTargetException e) {
            e.printStackTrace();
        }
        verify(connection, times(retry + 1)).request(any(), anyLong());
    }
    
    @Test
    void testNextRpcServerForIpv4WithPort() {
        RpcClient rpcClient = buildTestNextRpcServerClient();
        rpcClient.serverListFactory(serverListFactory);
        when(serverListFactory.genNextServer()).thenReturn("127.0.0.1:7777");
        RpcClient.ServerInfo actual = rpcClient.nextRpcServer();
        assertEquals("127.0.0.1:7777", actual.getAddress());
        assertEquals("127.0.0.1", actual.getServerIp());
        assertEquals(7777, actual.getServerPort());
    }
    
    @Test
    void testNextRpcServerForIpv4WithoutPort() {
        RpcClient rpcClient = buildTestNextRpcServerClient();
        rpcClient.serverListFactory(serverListFactory);
        when(serverListFactory.genNextServer()).thenReturn("127.0.0.1");
        RpcClient.ServerInfo actual = rpcClient.nextRpcServer();
        assertEquals("127.0.0.1:8848", actual.getAddress());
        assertEquals("127.0.0.1", actual.getServerIp());
        assertEquals(8848, actual.getServerPort());
    }
    
    @Test
    void testNextRpcServerForIpv6WithPort() {
        RpcClient rpcClient = buildTestNextRpcServerClient();
        rpcClient.serverListFactory(serverListFactory);
        when(serverListFactory.genNextServer()).thenReturn("[fe80::35ba:6827:c5ff:d161%11]:7777");
        RpcClient.ServerInfo actual = rpcClient.nextRpcServer();
        assertEquals("[fe80::35ba:6827:c5ff:d161%11]:7777", actual.getAddress());
        assertEquals("[fe80::35ba:6827:c5ff:d161%11]", actual.getServerIp());
        assertEquals(7777, actual.getServerPort());
    }
    
    @Test
    void testNextRpcServerForIpv6WithoutPort() {
        RpcClient rpcClient = buildTestNextRpcServerClient();
        rpcClient.serverListFactory(serverListFactory);
        when(serverListFactory.genNextServer()).thenReturn("[fe80::35ba:6827:c5ff:d161%11]");
        RpcClient.ServerInfo actual = rpcClient.nextRpcServer();
        assertEquals("[fe80::35ba:6827:c5ff:d161%11]:8848", actual.getAddress());
        assertEquals("[fe80::35ba:6827:c5ff:d161%11]", actual.getServerIp());
        assertEquals(8848, actual.getServerPort());
    }
    
    @Test
    void testNextRpcServerForDomainWithPort() {
        RpcClient rpcClient = buildTestNextRpcServerClient();
        rpcClient.serverListFactory(serverListFactory);
        when(serverListFactory.genNextServer()).thenReturn("nacos.io:7777");
        RpcClient.ServerInfo actual = rpcClient.nextRpcServer();
        assertEquals("nacos.io:7777", actual.getAddress());
        assertEquals("nacos.io", actual.getServerIp());
        assertEquals(7777, actual.getServerPort());
    }
    
    @Test
    void testNextRpcServerForDomainWithoutPort() {
        RpcClient rpcClient = buildTestNextRpcServerClient();
        rpcClient.serverListFactory(serverListFactory);
        when(serverListFactory.genNextServer()).thenReturn("nacos.io");
        RpcClient.ServerInfo actual = rpcClient.nextRpcServer();
        assertEquals("nacos.io:8848", actual.getAddress());
        assertEquals("nacos.io", actual.getServerIp());
        assertEquals(8848, actual.getServerPort());
    }
    
    @Test
    void testNextRpcServerForLocalhostWithPort() {
        RpcClient rpcClient = buildTestNextRpcServerClient();
        rpcClient.serverListFactory(serverListFactory);
        when(serverListFactory.genNextServer()).thenReturn("localhost:7777");
        RpcClient.ServerInfo actual = rpcClient.nextRpcServer();
        assertEquals("localhost:7777", actual.getAddress());
        assertEquals("localhost", actual.getServerIp());
        assertEquals(7777, actual.getServerPort());
    }
    
    @Test
    void testNextRpcServerForLocalhostWithoutPort() {
        RpcClient rpcClient = buildTestNextRpcServerClient();
        rpcClient.serverListFactory(serverListFactory);
        when(serverListFactory.genNextServer()).thenReturn("localhost");
        RpcClient.ServerInfo actual = rpcClient.nextRpcServer();
        assertEquals("localhost:8848", actual.getAddress());
        assertEquals("localhost", actual.getServerIp());
        assertEquals(8848, actual.getServerPort());
    }
    
    @Test
    void testNextRpcServerForEmpty() {
        assertThrows(IllegalArgumentException.class, () -> {
            RpcClient rpcClient = buildTestNextRpcServerClient();
            rpcClient.serverListFactory(serverListFactory);
            when(serverListFactory.genNextServer()).thenReturn("");
            rpcClient.nextRpcServer();
        });
    }
    
    private RpcClient buildTestNextRpcServerClient() {
        return new RpcClient(DefaultGrpcClientConfig.newBuilder().build()) {
            @Override
            public ConnectionType getConnectionType() {
                return null;
            }
            
            @Override
            public int rpcPortOffset() {
                return 0;
            }
            
            @Override
            public Connection connectToServer(ServerInfo serverInfo) {
                return null;
            }
            
            @Override
            public ServerInfo nextRpcServer() {
                return super.nextRpcServer();
            }
        };
    }
    
    @Test
    void testHandleServerRequestWhenExceptionThenThrowException() throws RuntimeException {
        assertThrows(RuntimeException.class, () -> {
            RpcClient rpcClient = buildTestNextRpcServerClient();
            Request request = new Request() {
                @Override
                public String getModule() {
                    return null;
                }
            };
            rpcClient.serverRequestHandlers.add((req, conn) -> {
                throw new RuntimeException();
            });
            rpcClient.handleServerRequest(request);
        });
    }
    
    @Test
    void testNotifyDisConnectedForEmpty() {
        rpcClient.notifyDisConnected(null);
        verify(rpcClientConfig, never()).name();
    }
    
    @Test
    void testNotifyDisConnected() {
        ConnectionEventListener listener = mock(ConnectionEventListener.class);
        rpcClient.registerConnectionListener(listener);
        rpcClient.notifyDisConnected(null);
        verify(listener).onDisConnect(null);
        verify(rpcClientConfig, times(2)).name();
    }
    
    @Test
    void testNotifyDisConnectedException() {
        ConnectionEventListener listener = mock(ConnectionEventListener.class);
        rpcClient.registerConnectionListener(listener);
        doThrow(new RuntimeException("test")).when(listener).onDisConnect(null);
        rpcClient.notifyDisConnected(null);
        verify(rpcClientConfig, times(3)).name();
    }
    
    @Test
    void testNotifyConnectedForEmpty() {
        rpcClient.notifyConnected(null);
        verify(rpcClientConfig, never()).name();
    }
    
    @Test
    void testNotifyConnected() {
        ConnectionEventListener listener = mock(ConnectionEventListener.class);
        rpcClient.registerConnectionListener(listener);
        rpcClient.notifyConnected(null);
        verify(listener).onConnected(null);
        verify(rpcClientConfig, times(2)).name();
    }
    
    @Test
    void testNotifyConnectedException() {
        ConnectionEventListener listener = mock(ConnectionEventListener.class);
        rpcClient.registerConnectionListener(listener);
        doThrow(new RuntimeException("test")).when(listener).onConnected(null);
        rpcClient.notifyConnected(null);
        verify(rpcClientConfig, times(3)).name();
    }
    
    @Test
    void testStartClient() throws NacosException {
        when(serverListFactory.genNextServer()).thenReturn("127.0.0.1:8848");
        connection.serverInfo = new RpcClient.ServerInfo("127.0.0.1", 8848);
        RpcClient rpcClient = buildTestStartClient(new Function<RpcClient.ServerInfo, Connection>() {
            
            private int count;
            
            @Override
            public Connection apply(RpcClient.ServerInfo serverInfo) {
                if (count == 0) {
                    count++;
                    throw new RuntimeException("test");
                }
                return connection;
            }
        });
        try {
            rpcClient.start();
            assertTrue(rpcClient.isRunning());
        } finally {
            rpcClient.shutdown();
        }
    }
    
    @Test
    void testStartClientWithFailed() throws NacosException, InterruptedException {
        RpcClient rpcClient = buildTestStartClient(serverInfo -> null);
        try {
            rpcClient.start();
            TimeUnit.MILLISECONDS.sleep(1000);
            assertFalse(rpcClient.isRunning());
        } finally {
            rpcClient.shutdown();
        }
    }
    
    @Test
    void testStartClientAfterShutdown() throws NacosException {
        RpcClient rpcClient = buildTestStartClient(serverInfo -> null);
        rpcClient.shutdown();
        rpcClient.start();
        assertTrue(rpcClient.isShutdown());
    }
    
    @Test
    void testDisConnectionEventAfterStart() throws NacosException, InterruptedException {
        when(serverListFactory.genNextServer()).thenReturn("127.0.0.1:8848");
        connection.serverInfo = new RpcClient.ServerInfo("127.0.0.1", 8848);
        RpcClient rpcClient = buildTestStartClient(serverInfo -> connection);
        ConnectionEventListener listener = mock(ConnectionEventListener.class);
        rpcClient.registerConnectionListener(listener);
        try {
            rpcClient.start();
            TimeUnit.MILLISECONDS.sleep(100);
            rpcClient.eventLinkedBlockingQueue.put(new RpcClient.ConnectionEvent(0, connection));
            TimeUnit.MILLISECONDS.sleep(100);
            verify(listener).onDisConnect(connection);
        } finally {
            rpcClient.shutdown();
        }
    }
    
    @Test
    void testReconnectContextAfterStartWithNullConnection() throws NacosException, InterruptedException {
        RpcClient rpcClient = buildTestStartClient(serverInfo -> null);
        try {
            when(rpcClientConfig.connectionKeepAlive()).thenReturn(-1L);
            rpcClient.start();
            TimeUnit.MILLISECONDS.sleep(100);
            verify(rpcClientConfig, never()).healthCheckRetryTimes();
        } finally {
            rpcClient.shutdown();
        }
    }
    
    @Test
    void testReconnectContextAfterStartWithConnectionHealthCheckFail() throws NacosException, InterruptedException {
        when(serverListFactory.genNextServer()).thenReturn("127.0.0.1:8848");
        connection.serverInfo = new RpcClient.ServerInfo("127.0.0.1", 8848);
        RpcClient rpcClient = buildTestStartClient(new Function<RpcClient.ServerInfo, Connection>() {
            
            private int count;
            
            @Override
            public Connection apply(RpcClient.ServerInfo serverInfo) {
                if (count == 0) {
                    count++;
                    return connection;
                }
                return null;
            }
        });
        try {
            when(rpcClientConfig.connectionKeepAlive()).thenReturn(10L);
            rpcClient.start();
            TimeUnit.MILLISECONDS.sleep(500);
            assertEquals(RpcClientStatus.UNHEALTHY, rpcClient.rpcClientStatus.get());
        } finally {
            rpcClient.shutdown();
        }
    }
    
    @Test
    void testReconnectContextAfterStartWithConnectionHealthCheckSuccess()
            throws NacosException, InterruptedException, NoSuchFieldException, IllegalAccessException {
        when(serverListFactory.genNextServer()).thenReturn("127.0.0.1:8848");
        connection.serverInfo = new RpcClient.ServerInfo("127.0.0.1", 8848);
        RpcClient rpcClient = buildTestStartClient(serverInfo -> connection);
        when(connection.request(any(Request.class), anyLong())).thenReturn(new HealthCheckResponse());
        try {
            Field lastActiveTimeStampField = RpcClient.class.getDeclaredField("lastActiveTimeStamp");
            lastActiveTimeStampField.setAccessible(true);
            final long lastActiveTimeStamp = (long) lastActiveTimeStampField.get(rpcClient);
            when(rpcClientConfig.connectionKeepAlive()).thenReturn(10L);
            rpcClient.start();
            TimeUnit.MILLISECONDS.sleep(100);
            assertEquals(RpcClientStatus.RUNNING, rpcClient.rpcClientStatus.get());
            long newLastActiveTimeStamp = (long) lastActiveTimeStampField.get(rpcClient);
            assertTrue(newLastActiveTimeStamp > lastActiveTimeStamp);
        } finally {
            rpcClient.shutdown();
        }
    }
    
    @Test
    void testReconnectContextAfterStartWithActiveTimeIsNew()
            throws NacosException, InterruptedException, NoSuchFieldException, IllegalAccessException {
        when(serverListFactory.genNextServer()).thenReturn("127.0.0.1:8848");
        connection.serverInfo = new RpcClient.ServerInfo("127.0.0.1", 8848);
        RpcClient rpcClient = buildTestStartClient(serverInfo -> connection);
        try {
            Field lastActiveTimeStampField = RpcClient.class.getDeclaredField("lastActiveTimeStamp");
            lastActiveTimeStampField.setAccessible(true);
            long setTime = System.currentTimeMillis() + 10000;
            lastActiveTimeStampField.set(rpcClient, setTime);
            when(rpcClientConfig.connectionKeepAlive()).thenReturn(10L);
            rpcClient.start();
            TimeUnit.MILLISECONDS.sleep(100);
            assertEquals(RpcClientStatus.RUNNING, rpcClient.rpcClientStatus.get());
            long newLastActiveTimeStamp = (long) lastActiveTimeStampField.get(rpcClient);
            assertEquals(setTime, newLastActiveTimeStamp);
        } finally {
            rpcClient.shutdown();
        }
    }
    
    @Test
    void testReconnectContextAfterStartWithOldServiceInfo() throws NacosException, InterruptedException, IllegalAccessException {
        when(serverListFactory.genNextServer()).thenReturn("127.0.0.1:8848");
        when(serverListFactory.getServerList()).thenReturn(Collections.singletonList("127.0.0.1:8848"));
        connection.serverInfo = new RpcClient.ServerInfo("127.0.0.1", 8848);
        RpcClient rpcClient = buildTestStartClient(serverInfo -> connection);
        try {
            rpcClient.start();
            RpcClient.ReconnectContext reconnectContext = new RpcClient.ReconnectContext(new RpcClient.ServerInfo("127.0.0.1", 0),
                    false);
            ((BlockingQueue<RpcClient.ReconnectContext>) reconnectionSignalField.get(rpcClient)).put(reconnectContext);
            TimeUnit.MILLISECONDS.sleep(100);
            assertEquals(RpcClientStatus.RUNNING, rpcClient.rpcClientStatus.get());
            assertEquals(8848, reconnectContext.serverInfo.serverPort);
        } finally {
            rpcClient.shutdown();
        }
    }
    
    @Test
    void testReconnectContextAfterStartWithNewServiceInfo() throws NacosException, InterruptedException, IllegalAccessException {
        when(serverListFactory.genNextServer()).thenReturn("127.0.0.1:8848");
        when(serverListFactory.getServerList()).thenReturn(Collections.singletonList("1.1.1.1:8848"));
        connection.serverInfo = new RpcClient.ServerInfo("127.0.0.1", 8848);
        RpcClient rpcClient = buildTestStartClient(serverInfo -> connection);
        try {
            rpcClient.start();
            RpcClient.ReconnectContext reconnectContext = new RpcClient.ReconnectContext(new RpcClient.ServerInfo("127.0.0.1", 0),
                    false);
            ((BlockingQueue<RpcClient.ReconnectContext>) reconnectionSignalField.get(rpcClient)).put(reconnectContext);
            TimeUnit.MILLISECONDS.sleep(100);
            assertEquals(RpcClientStatus.RUNNING, rpcClient.rpcClientStatus.get());
            assertNull(reconnectContext.serverInfo);
        } finally {
            rpcClient.shutdown();
        }
    }
    
    @Test
    void testHandleConnectionResetRequestWithoutServer() throws NacosException, InterruptedException {
        when(serverListFactory.genNextServer()).thenReturn("127.0.0.1:8848", "1.1.1.1:8848");
        connection.serverInfo = new RpcClient.ServerInfo("127.0.0.1", 8848);
        RpcClient rpcClient = buildTestStartClient(serverInfo -> {
            connection.serverInfo = serverInfo;
            return connection;
        });
        try {
            rpcClient.start();
            Response response = rpcClient.handleServerRequest(new ConnectResetRequest());
            assertTrue(response instanceof ConnectResetResponse);
            TimeUnit.MILLISECONDS.sleep(500);
            assertEquals("1.1.1.1", connection.serverInfo.getServerIp());
        } finally {
            rpcClient.shutdown();
        }
    }
    
    @Test
    void testHandleConnectionResetRequestWithServer() throws NacosException, InterruptedException {
        when(serverListFactory.genNextServer()).thenReturn("127.0.0.1:8848", "1.1.1.1:8848");
        List<String> serverList = new LinkedList<>();
        serverList.add("127.0.0.1:8848");
        serverList.add("1.1.1.1:8848");
        serverList.add("2.2.2.2:8848");
        when(serverListFactory.getServerList()).thenReturn(serverList);
        connection.serverInfo = new RpcClient.ServerInfo("127.0.0.1", 8848);
        RpcClient rpcClient = buildTestStartClient(serverInfo -> {
            connection.serverInfo = serverInfo;
            return connection;
        });
        try {
            rpcClient.start();
            ConnectResetRequest connectResetRequest = new ConnectResetRequest();
            connectResetRequest.setServerIp("2.2.2.2");
            connectResetRequest.setServerPort("8848");
            Response response = rpcClient.handleServerRequest(connectResetRequest);
            assertTrue(response instanceof ConnectResetResponse);
            TimeUnit.MILLISECONDS.sleep(500);
            assertEquals("2.2.2.2", connection.serverInfo.getServerIp());
        } finally {
            rpcClient.shutdown();
        }
    }
    
    @Test
    void testHandleConnectionResetRequestWithException() throws NacosException, InterruptedException {
        when(serverListFactory.genNextServer()).thenReturn("127.0.0.1:8848", "1.1.1.1:8848");
        connection.serverInfo = new RpcClient.ServerInfo("127.0.0.1", 8848);
        RpcClient rpcClient = buildTestStartClient(serverInfo -> {
            connection.serverInfo = serverInfo;
            return connection;
        });
        try {
            rpcClient.start();
            System.setProperty("nacos.server.port", "2.2.2.2");
            ConnectResetRequest connectResetRequest = new ConnectResetRequest();
            connectResetRequest.setServerIp("2.2.2.2");
            Response response = rpcClient.handleServerRequest(connectResetRequest);
            assertTrue(response instanceof ConnectResetResponse);
            TimeUnit.MILLISECONDS.sleep(500);
            assertEquals("127.0.0.1", connection.serverInfo.getServerIp());
        } finally {
            rpcClient.shutdown();
        }
    }
    
    @Test
    void testHandleClientDetectionRequest() throws NacosException {
        RpcClient rpcClient = buildTestStartClient(serverInfo -> null);
        try {
            rpcClient.start();
            Response response = rpcClient.handleServerRequest(new ClientDetectionRequest());
            assertTrue(response instanceof ClientDetectionResponse);
        } finally {
            rpcClient.shutdown();
        }
    }
    
    @Test
    void testHandleOtherRequest() throws NacosException {
        RpcClient rpcClient = buildTestStartClient(serverInfo -> null);
        try {
            rpcClient.start();
            Response response = rpcClient.handleServerRequest(new HealthCheckRequest());
            assertNull(response);
        } finally {
            rpcClient.shutdown();
        }
    }
    
    @Test
    void testReconnectForRequestFailButHealthCheckOK() throws NacosException {
        RpcClient rpcClient = buildTestStartClient(serverInfo -> null);
        rpcClient.currentConnection = connection;
        connection.serverInfo = new RpcClient.ServerInfo("127.0.0.1", 8848);
        when(connection.request(any(Request.class), anyLong())).thenReturn(new HealthCheckResponse());
        rpcClient.reconnect(null, true);
        assertTrue(rpcClient.isRunning());
    }
    
    @Test
    void testReconnectFailTimes() throws NacosException {
        when(serverListFactory.genNextServer()).thenReturn("127.0.0.1:8848");
        when(serverListFactory.getServerList()).thenReturn(Collections.singletonList("127.0.0.1:8848"));
        final AtomicInteger count = new AtomicInteger(0);
        RpcClient rpcClient = buildTestStartClient(serverInfo -> {
            int actual = count.incrementAndGet();
            return actual > 3 ? connection : null;
        });
        rpcClient.currentConnection = connection;
        connection.serverInfo = new RpcClient.ServerInfo("127.0.0.1", 8848);
        long start = System.currentTimeMillis();
        rpcClient.reconnect(null, false);
        assertTrue(rpcClient.isRunning());
        assertTrue(System.currentTimeMillis() - start > 400);
    }
    
    @Test
    void testGetCurrentServer() {
        assertNull(rpcClient.getCurrentServer());
        rpcClient.currentConnection = connection;
        rpcClient.serverListFactory(serverListFactory);
        connection.serverInfo = new RpcClient.ServerInfo("127.0.0.1", 8848);
        assertNotNull(rpcClient.getCurrentServer());
    }
    
    @Test
    void testCurrentRpcServer() throws IllegalAccessException {
        when(serverListFactory.getCurrentServer()).thenReturn("127.0.0.1:8848");
        serverListFactoryField.set(rpcClient, serverListFactory);
        RpcClient.ServerInfo serverInfo = rpcClient.currentRpcServer();
        assertEquals("127.0.0.1", serverInfo.getServerIp());
        assertEquals(8848, serverInfo.getServerPort());
        assertEquals("127.0.0.1:8848", serverInfo.getAddress());
    }
    
    private RpcClient buildTestStartClient(Function<RpcClient.ServerInfo, Connection> function) {
        return new RpcClient(rpcClientConfig, serverListFactory) {
            
            @Override
            public ConnectionType getConnectionType() {
                return ConnectionType.GRPC;
            }
            
            @Override
            public int rpcPortOffset() {
                return 0;
            }
            
            @Override
            public Connection connectToServer(ServerInfo serverInfo) {
                return function.apply(serverInfo);
            }
        };
    }
    
    @Test
    void testServerInfoSet() {
        RpcClient.ServerInfo serverInfo = new RpcClient.ServerInfo();
        String ip = "127.0.0.1";
        int port = 80;
        serverInfo.setServerIp(ip);
        serverInfo.setServerPort(port);
        assertEquals("127.0.0.1:80", serverInfo.getAddress());
        assertEquals(port, serverInfo.getServerPort());
        assertEquals(ip, serverInfo.getServerIp());
        String expected = "{serverIp = '127.0.0.1', server main port = 80}";
        assertEquals(expected, serverInfo.toString());
    }
    
    @Test
    void testSetTenant() {
        String tenant = "testTenant";
        assertNull(rpcClient.getTenant());
        rpcClient.setTenant(tenant);
        assertEquals(tenant, rpcClient.getTenant());
    }
    
    @Test
    void testGetConnectionAbilityWithNullConnection() {
        AbilityStatus abilityStatus = rpcClient.getConnectionAbility(AbilityKey.SERVER_FUZZY_WATCH);
        assertNull(abilityStatus);
    }
    
    @Test
    void testGetConnectionAbilityWithReadyConnection() {
        when(connection.getConnectionAbility(AbilityKey.SERVER_FUZZY_WATCH)).thenReturn(AbilityStatus.SUPPORTED);
        rpcClient.currentConnection = connection;
        AbilityStatus abilityStatus = rpcClient.getConnectionAbility(AbilityKey.SERVER_FUZZY_WATCH);
        assertNotNull(abilityStatus);
        assertEquals(AbilityStatus.SUPPORTED, abilityStatus);
    }
}
