/*
 * 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.plc4x.java.spi.connection;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.SocketAddress;

/**
 * Adapter with sensible defaults for a Netty Based Channel Factory.
 * <p>
 * By Default Nettys {@link NioEventLoopGroup} is used.
 * Transports which have to use a different EventLoopGroup have to override {@link #getEventLoopGroup()}.
 */
public abstract class NettyChannelFactory implements ChannelFactory {

    private static final Logger logger = LoggerFactory.getLogger(NettyChannelFactory.class);

    /**
     * TODO should be removed together with the Constructor.
     */
    private SocketAddress address;

    protected NettyChannelFactory(SocketAddress address) {
        this.address = address;
    }

    /**
     * Channel to Use, e.g. NiO, EiO
     */
    public abstract Class<? extends Channel> getChannel();

    /**
     * We need to be able to override this in the TestChanelFactory
     *
     * @return the Bootstrap instance we will be using to initialize the channel.
     */
    protected Bootstrap createBootstrap() {
        return new Bootstrap();
    }

    /**
     * This Method is used to modify the Bootstrap Element of Netty, if one wishes to do so.
     * E.g. for Protocol Specific extension.
     * For TCP e.g.
     * <code>
     * bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
     * bootstrap.option(ChannelOption.TCP_NODELAY, true);
     * </code>
     */
    public abstract void configureBootstrap(Bootstrap bootstrap);

    /**
     * Event Loop Group to use.
     * Has to be in accordance with {@link #getChannel()}
     * otherwise a Runtime Exception will be produced by Netty
     * <p>
     * By Default Nettys {@link NioEventLoopGroup} is used.
     * Transports which have to use a different EventLoopGroup have to override {@link #getEventLoopGroup()}.
     */
    public EventLoopGroup getEventLoopGroup() {
        return new NioEventLoopGroup();
    }

    @Override
    public Channel createChannel(ChannelHandler channelHandler) throws PlcConnectionException {
        try {
            Bootstrap bootstrap = createBootstrap();

            final EventLoopGroup workerGroup = getEventLoopGroup();
            if (workerGroup != null) {
                bootstrap.group(workerGroup);
            }

            bootstrap.channel(getChannel());
            // Callback to allow subclasses to modify the Bootstrap

            configureBootstrap(bootstrap);
            bootstrap.handler(channelHandler);
            // Start the client.
            final ChannelFuture f = bootstrap.connect(address);
            f.addListener(future -> {
                if (!future.isSuccess()) {
                    logger.info("Unable to connect, shutting down worker thread.");
                    workerGroup.shutdownGracefully();
                }
            });

            final Channel channel = f.channel();

            // It seems the embedded channel operates differently.
            // Intentionally using the class name as we don't want to require a
            // hard dependency on the test-channel.
            if (!"Plc4xEmbeddedChannel".equals(channel.getClass().getSimpleName())) {
                // Wait for sync
                f.sync();
                // Wait till the session is finished initializing.
                f.awaitUninterruptibly(); // jf: unsure if we need that
            }

            return channel;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new PlcConnectionException("Error creating channel.", e);
        } catch (Exception e) {
            throw new PlcConnectionException("Error creating channel.", e);
        }
    }

}
