/*
<<<<<<< HEAD:src/main/java/com/actiontech/dble/net/AIOAcceptor.java
 * Copyright (C) 2016-2022 ActionTech.
 * based on code by MyCATCopyrightHolder Copyright (c) 2013, OpenCloudDB/MyCAT.
 * License: http://www.gnu.org/licenses/gpl.html GPL version 2 or higher.
 */
package com.actiontech.dble.net.impl.aio;

import com.actiontech.dble.DbleServer;
import com.actiontech.dble.net.IOProcessor;
import com.actiontech.dble.net.SocketAcceptor;
import com.actiontech.dble.net.connection.FrontendConnection;
import com.actiontech.dble.net.factory.FrontendConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.StandardSocketOptions;
import java.nio.channels.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author mycat
 */
public final class AIOAcceptor implements SocketAcceptor,
        CompletionHandler<AsynchronousSocketChannel, Long> {
    private static final Logger LOGGER = LoggerFactory.getLogger(AIOAcceptor.class);
    private static final AcceptIdGenerator ID_GENERATOR = new AcceptIdGenerator();

    private final int port;
    private final AsynchronousServerSocketChannel serverChannel;
    private final FrontendConnectionFactory factory;

    private final String name;

    public AIOAcceptor(String name, String ip, int port, int backlog,
                       FrontendConnectionFactory factory, AsynchronousChannelGroup group)
            throws IOException {
        this.name = name;
        this.port = port;
        this.factory = factory;
        serverChannel = AsynchronousServerSocketChannel.open(group);
        /** set TCP option */
        serverChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
        serverChannel.setOption(StandardSocketOptions.SO_RCVBUF, 1024 * 16 * 2);
        // backlog=100
        serverChannel.bind(new InetSocketAddress(ip, port), backlog);
    }

    public String getName() {
        return name;
    }

    public void start() {
        this.pendingAccept();
    }

    public int getPort() {
        return port;
    }

    private void accept(NetworkChannel channel, Long id) {
        try {
            AIOSocketWR socketWR = new AIOSocketWR();
            FrontendConnection c = factory.make(channel, socketWR);
            socketWR.initFromConnection(c);
            c.setId(id);
            IOProcessor processor = DbleServer.getInstance().nextFrontProcessor();
            c.setProcessor(processor);
            c.register();
        } catch (Exception e) {
            LOGGER.info("AioAcceptorError", e);
            closeChannel(channel);
        }
    }

    private void pendingAccept() {
        if (serverChannel.isOpen()) {
            serverChannel.accept(ID_GENERATOR.getId(), this);
        } else {
            throw new IllegalStateException(
                    "Server Channel has been closed");
        }

    }

    @Override
    public void completed(AsynchronousSocketChannel result, Long id) {
        accept(result, id);
        // next pending waiting
        pendingAccept();

    }

    @Override
    public void failed(Throwable exc, Long id) {
        LOGGER.info("acception connect failed:" + exc);
        // next pending waiting
        pendingAccept();

    }

    private static void closeChannel(NetworkChannel channel) {
        if (channel == null) {
            return;
        }
        try {
            channel.close();
        } catch (IOException e) {
            LOGGER.info("AioAcceptorError", e);
        }
    }

    /**
     * @author mycat
     */
    private static class AcceptIdGenerator {

        private static final long MAX_VALUE = 0xffffffffL;

        private AtomicLong acceptId = new AtomicLong();
        private final Object lock = new Object();

        private long getId() {
            long newValue = acceptId.getAndIncrement();
            if (newValue >= MAX_VALUE) {
                synchronized (lock) {
                    newValue = acceptId.getAndIncrement();
                    if (newValue >= MAX_VALUE) {
                        acceptId.set(0);
                    }
                }
                return acceptId.getAndDecrement();
            } else {
                return newValue;
            }
        }
    }
}
