/*
 * 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.remoting.transport.netty4;

import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.Version;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.ExecutorUtil;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.remoting.Channel;
import org.apache.dubbo.remoting.ChannelHandler;
import org.apache.dubbo.remoting.RemotingException;
import org.apache.dubbo.remoting.transport.netty4.ssl.SslClientTlsHandler;
import org.apache.dubbo.remoting.api.WireProtocol;
import org.apache.dubbo.remoting.api.connection.AbstractConnectionClient;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoop;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.ssl.SslContext;
import io.netty.util.AttributeKey;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.GlobalEventExecutor;
import io.netty.util.concurrent.Promise;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_CLIENT_THREADPOOL;
import static org.apache.dubbo.common.constants.CommonConstants.SSL_ENABLED_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.THREADPOOL_KEY;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_CLIENT_CONNECT_TIMEOUT;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_CONNECT_PROVIDER;
import static org.apache.dubbo.remoting.transport.netty4.NettyEventLoopFactory.socketChannelClass;

public class NettyConnectionClient extends AbstractConnectionClient {

    private static final ErrorTypeAwareLogger LOGGER = LoggerFactory.getErrorTypeAwareLogger(NettyConnectionClient.class);

    private AtomicReference<Promise<Object>> connectingPromise;

    private Promise<Void> closePromise;

    private AtomicReference<io.netty.channel.Channel> channel;

    private ConnectionListener connectionListener;

    private Bootstrap bootstrap;

    public static final AttributeKey<AbstractConnectionClient> CONNECTION = AttributeKey.valueOf("connection");


    public NettyConnectionClient(URL url, ChannelHandler handler) throws RemotingException {
        super(url, handler);
    }

    @Override
    protected void initConnectionClient() {
        URL url = ExecutorUtil.setThreadName(getUrl(), "DubboClientHandler");
        url = url.addParameterIfAbsent(THREADPOOL_KEY, DEFAULT_CLIENT_THREADPOOL);
        setUrl(url);
        this.protocol = url.getOrDefaultFrameworkModel().getExtensionLoader(WireProtocol.class).getExtension(url.getProtocol());
        this.remote = getConnectAddress();
        this.connectingPromise = new AtomicReference<>();
        this.connectionListener = new ConnectionListener();
        this.channel = new AtomicReference<>();
        this.closePromise = new DefaultPromise<>(GlobalEventExecutor.INSTANCE);
        this.init = new AtomicBoolean(false);
    }

    @Override
    protected void doOpen() throws Throwable {
        initConnectionClient();
        initBootstrap();
    }

    private void initBootstrap() {
        final Bootstrap nettyBootstrap = new Bootstrap();
        nettyBootstrap.group(NettyEventLoopFactory.NIO_EVENT_LOOP_GROUP.get())
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .remoteAddress(getConnectAddress())
                .channel(socketChannelClass());

        final NettyConnectionHandler connectionHandler = new NettyConnectionHandler(this);
        nettyBootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getConnectTimeout());
        nettyBootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) {
                NettyChannel nettyChannel = NettyChannel.getOrAddChannel(ch, getUrl(), getChannelHandler());
                final ChannelPipeline pipeline = ch.pipeline();
                SslContext sslContext = null;
                if (getUrl().getParameter(SSL_ENABLED_KEY, false)) {
                    pipeline.addLast("negotiation", new SslClientTlsHandler(getUrl()));
                }

//                pipeline.addLast("logging", new LoggingHandler(LogLevel.INFO)); //for debug
                // TODO support IDLE
//                int heartbeatInterval = UrlUtils.getHeartbeat(getUrl());
                pipeline.addLast("connectionHandler", connectionHandler);

                NettyConfigOperator operator = new NettyConfigOperator(nettyChannel, getChannelHandler());
                protocol.configClientPipeline(getUrl(), operator, sslContext);
                // TODO support Socks5
            }
        });
        this.bootstrap = nettyBootstrap;
    }

    @Override
    protected void doClose() {
        // AbstractPeer close can set closed true.
        if (isClosed()) {
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug(String.format("Connection:%s freed ", this));
            }
            final io.netty.channel.Channel current = getNettyChannel();
            if (current != null) {
                current.close();
            }
            this.channel.set(null);
            closePromise.setSuccess(null);
        }
    }

    @Override
    protected void doConnect() throws RemotingException {
        if (isClosed()) {
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug(String.format("%s aborted to reconnect cause connection closed. ",
                        NettyConnectionClient.this));
            }
        }
        init.compareAndSet(false, true);
        long start = System.currentTimeMillis();

        createConnectingPromise();
        final ChannelFuture promise = bootstrap.connect();

        promise.addListener(this.connectionListener);

        boolean ret = connectingPromise.get().awaitUninterruptibly(getConnectTimeout(), TimeUnit.MILLISECONDS);
        // destroy connectingPromise after used
        synchronized (this) {
            connectingPromise.set(null);
        }
        if (promise.cause() != null) {
            Throwable cause = promise.cause();

            // 6-1 Failed to connect to provider server by other reason.
            RemotingException remotingException = new RemotingException(this, "client(url: " + getUrl() + ") failed to connect to server "
                    + getConnectAddress() + ", error message is:" + cause.getMessage(), cause);

            LOGGER.error(TRANSPORT_FAILED_CONNECT_PROVIDER, "network disconnected", "",
                    "Failed to connect to provider server by other reason.", cause);

            throw remotingException;
        } else if (!ret || !promise.isSuccess()) {
            // 6-2 Client-side timeout
            RemotingException remotingException = new RemotingException(this, "client(url: " + getUrl() + ") failed to connect to server "
                    + getConnectAddress() + " client-side timeout "
                    + getConnectTimeout() + "ms (elapsed: " + (System.currentTimeMillis() - start) + "ms) from netty client "
                    + NetUtils.getLocalHost() + " using dubbo version " + Version.getVersion());

            LOGGER.error(TRANSPORT_CLIENT_CONNECT_TIMEOUT, "provider crash", "",
                    "Client-side timeout.", remotingException);

            throw remotingException;
        }
    }

    @Override
    protected void doDisConnect() {
        NettyChannel.removeChannelIfDisconnected(getNettyChannel());
    }

    @Override
    public void onConnected(Object channel) {
        if (!(channel instanceof io.netty.channel.Channel)) {
            return;
        }
        io.netty.channel.Channel nettyChannel = ((io.netty.channel.Channel) channel);
        if (isClosed()) {
            nettyChannel.close();
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug(String.format("%s is closed, ignoring connected event", this));
            }
            return;
        }
        this.channel.set(nettyChannel);
        // This indicates that the connection is available.
        if (this.connectingPromise.get() != null) {
            this.connectingPromise.get().trySuccess(CONNECTED_OBJECT);
        }
        nettyChannel.attr(CONNECTION).set(this);
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug(String.format("%s connected ", this));
        }
    }

    @Override
    public void onGoaway(Object channel) {
        if (!(channel instanceof io.netty.channel.Channel)) {
            return;
        }
        io.netty.channel.Channel nettyChannel = (io.netty.channel.Channel) channel;
        if (this.channel.compareAndSet(nettyChannel, null)) {
            NettyChannel.removeChannelIfDisconnected(nettyChannel);
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug(String.format("%s goaway", this));
            }
        }
    }

    @Override
    protected Channel getChannel() {
        io.netty.channel.Channel c = getNettyChannel();
        if (c == null) {
            return null;
        }
        return NettyChannel.getOrAddChannel(c, getUrl(), this);
    }

    io.netty.channel.Channel getNettyChannel() {
        return this.channel.get();
    }

    @Override
    public Object getChannel(Boolean generalizable) {
        return Boolean.TRUE.equals(generalizable) ? getNettyChannel() : getChannel();
    }

    @Override
    public boolean isAvailable() {
        if (isClosed()) {
            return false;
        }
        io.netty.channel.Channel nettyChannel = getNettyChannel();
        if (nettyChannel != null && nettyChannel.isActive()) {
            return true;
        }

        if (init.compareAndSet(false, true)) {
            try {
                doConnect();
            } catch (RemotingException e) {
                LOGGER.error("Failed to connect to server: " + getConnectAddress());
            }
        }

        createConnectingPromise();
        connectingPromise.get().awaitUninterruptibly(getConnectTimeout(), TimeUnit.MILLISECONDS);
        // destroy connectingPromise after used
        synchronized (this) {
            connectingPromise.set(null);
        }

        nettyChannel = getNettyChannel();
        return nettyChannel != null && nettyChannel.isActive();

    }

    @Override
    public void createConnectingPromise() {
        connectingPromise.compareAndSet(null, new DefaultPromise<>(GlobalEventExecutor.INSTANCE));
    }

    public Promise<Void> getClosePromise() {
        return closePromise;
    }

    public static AbstractConnectionClient getConnectionClientFromChannel(io.netty.channel.Channel channel) {
        return channel.attr(CONNECTION).get();
    }

    public ChannelFuture write(Object request) throws RemotingException {
        if (!isAvailable()) {
            throw new RemotingException(null, null,
                    "Failed to send request " + request + ", cause: The channel to " + remote
                            + " is closed!");
        }
        return ((io.netty.channel.Channel) getChannel()).writeAndFlush(request);
    }

    @Override
    public void addCloseListener(Runnable func) {
        getClosePromise().addListener(future -> func.run());
    }

    @Override
    public void destroy() {
        close();
    }

    @Override
    public String toString() {
        return super.toString() + " (Ref=" + this.getCounter() + ",local=" +
                (getChannel() == null ? null : getChannel().getLocalAddress()) + ",remote=" + getRemoteAddress();
    }

    class ConnectionListener implements ChannelFutureListener {

        @Override
        public void operationComplete(ChannelFuture future) {
            if (future.isSuccess()) {
                return;
            }
            final NettyConnectionClient connectionClient = NettyConnectionClient.this;
            if (connectionClient.isClosed() || connectionClient.getCounter() == 0) {
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug(String.format("%s aborted to reconnect. %s", connectionClient,
                            future.cause().getMessage()));
                }
                return;
            }
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug(String.format("%s is reconnecting, attempt=%d cause=%s", connectionClient, 0,
                        future.cause().getMessage()));
            }
            final EventLoop loop = future.channel().eventLoop();
            loop.schedule(() -> {
                try {
                    connectionClient.doConnect();
                } catch (RemotingException e) {
                    LOGGER.error("Failed to connect to server: " + getConnectAddress());
                }
            }, 1L, TimeUnit.SECONDS);
        }
    }
}
