/*
 * 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.asterix.replication.management;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.asterix.common.api.INcApplicationContext;
import org.apache.asterix.common.config.ReplicationProperties;
import org.apache.asterix.common.replication.IReplicationChannel;
import org.apache.asterix.replication.api.IReplicaTask;
import org.apache.asterix.replication.api.IReplicationWorker;
import org.apache.asterix.replication.logging.RemoteLogsProcessor;
import org.apache.asterix.replication.messaging.ReplicationProtocol;
import org.apache.asterix.replication.messaging.ReplicationProtocol.ReplicationRequestType;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.network.INetworkSecurityManager;
import org.apache.hyracks.api.network.ISocketChannel;
import org.apache.hyracks.api.network.ISocketChannelFactory;
import org.apache.hyracks.util.NetworkUtil;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

/**
 * This class is used to receive and process replication requests from remote replicas or replica events from CC
 */
public class ReplicationChannel extends Thread implements IReplicationChannel {

    private static final Logger LOGGER = LogManager.getLogger();
    private ServerSocketChannel serverSocketChannel = null;
    private final INcApplicationContext appCtx;
    private final RemoteLogsProcessor logsProcessor;
    private final AtomicInteger replicationWorkerCounter = new AtomicInteger(0);

    public ReplicationChannel(INcApplicationContext appCtx) {
        this.appCtx = appCtx;
        logsProcessor = new RemoteLogsProcessor(appCtx);
    }

    @Override
    public void run() {
        final String nodeId = appCtx.getServiceContext().getNodeId();
        Thread.currentThread().setName(nodeId + " Replication Channel Thread");
        final ReplicationProperties replicationProperties = appCtx.getReplicationProperties();
        final String nodeIP = replicationProperties.getReplicationAddress();
        final int dataPort = replicationProperties.getReplicationPort();
        try {
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(true);
            InetSocketAddress replicationChannelAddress =
                    new InetSocketAddress(InetAddress.getByName(nodeIP), dataPort);
            serverSocketChannel.socket().bind(replicationChannelAddress);
            LOGGER.log(Level.INFO, "opened Replication Channel @ IP Address: " + nodeIP + ":" + dataPort);
            while (serverSocketChannel.isOpen()) {
                SocketChannel socketChannel = serverSocketChannel.accept();
                connectionAccepted(socketChannel);
            }
        } catch (AsynchronousCloseException e) {
            LOGGER.debug("Replication channel closed", e);
        } catch (IOException e) {
            throw new IllegalStateException("Failed to bind replication channel @ " + nodeIP + ":" + dataPort, e);
        }
    }

    public RemoteLogsProcessor getRemoteLogsProcessor() {
        return logsProcessor;
    }

    @Override
    public void close() throws IOException {
        if (serverSocketChannel != null) {
            serverSocketChannel.close();
            LOGGER.info("Replication channel closed.");
        }
    }

    private void connectionAccepted(SocketChannel socketChannel) {
        try {
            NetworkUtil.configure(socketChannel);
            socketChannel.configureBlocking(false);
            final INetworkSecurityManager networkSecurityManager =
                    appCtx.getServiceContext().getControllerService().getNetworkSecurityManager();
            final ISocketChannelFactory socketChannelFactory = networkSecurityManager.getSocketChannelFactory();
            final ISocketChannel serverChannel = socketChannelFactory.createServerChannel(socketChannel);
            //start a new thread to handle the request
            appCtx.getThreadExecutor().execute(new ReplicationWorker(serverChannel));
        } catch (Exception e) {
            LOGGER.error("failed to process accepted connection", e);
        }
    }

    private class ReplicationWorker implements IReplicationWorker {
        private final ISocketChannel socketChannel;
        private final ByteBuffer inBuffer;
        private final ByteBuffer outBuffer;

        public ReplicationWorker(ISocketChannel socketChannel) {
            this.socketChannel = socketChannel;
            inBuffer = ByteBuffer.allocate(ReplicationProtocol.INITIAL_BUFFER_SIZE);
            outBuffer = ByteBuffer.allocate(ReplicationProtocol.INITIAL_BUFFER_SIZE);
        }

        @Override
        public void run() {
            Thread.currentThread().setName("Replication Worker-" + replicationWorkerCounter.incrementAndGet() + "("
                    + getRemoteAddress() + ")");
            try {
                if (socketChannel.requiresHandshake() && !socketChannel.handshake()) {
                    LOGGER.warn("failed to complete handshake with {}", this::getRemoteAddress);
                    return;
                }
                socketChannel.getSocketChannel().configureBlocking(true);
                LOGGER.trace("reading replication worker initial request");
                ReplicationRequestType requestType = ReplicationProtocol.getRequestType(socketChannel, inBuffer);
                LOGGER.trace("got request type: {}", requestType);
                while (requestType != ReplicationRequestType.GOODBYE) {
                    handle(requestType);
                    LOGGER.trace("handled request type: {}", requestType);
                    requestType = ReplicationProtocol.getRequestType(socketChannel, inBuffer);
                    LOGGER.trace("got request type: {}", requestType);
                }
            } catch (Exception e) {
                LOGGER.warn("unexpected error during replication.", e);
            } finally {
                NetworkUtil.closeQuietly(socketChannel);
            }
        }

        @Override
        public ISocketChannel getChannel() {
            return socketChannel;
        }

        @Override
        public ByteBuffer getReusableBuffer() {
            return outBuffer;
        }

        @Override
        public String getRemoteAddress() {
            try {
                return socketChannel.getSocketChannel().getRemoteAddress().toString();
            } catch (Exception e) {
                return "unknown";
            }
        }

        private void handle(ReplicationRequestType requestType) throws HyracksDataException {
            final IReplicaTask task =
                    (IReplicaTask) ReplicationProtocol.readMessage(requestType, socketChannel, inBuffer);
            task.perform(appCtx, this);
        }
    }
}
