/*
    * 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.shardingsphere.proxy.frontend;
  
  import io.netty.bootstrap.ServerBootstrap;
  import io.netty.buffer.PooledByteBufAllocator;
  import io.netty.channel.ChannelFuture;
  import io.netty.channel.ChannelOption;
  import io.netty.channel.EventLoopGroup;
  import io.netty.channel.WriteBufferWaterMark;
  import io.netty.channel.epoll.Epoll;
  import io.netty.channel.epoll.EpollEventLoopGroup;
  import io.netty.channel.epoll.EpollServerSocketChannel;
  import io.netty.channel.nio.NioEventLoopGroup;
  import io.netty.channel.socket.nio.NioServerSocketChannel;
  import io.netty.handler.logging.LogLevel;
  import io.netty.handler.logging.LoggingHandler;
  import lombok.RequiredArgsConstructor;
  import org.apache.shardingsphere.proxy.frontend.netty.CDCServerHandlerInitializer;
  
  import java.util.ArrayList;
  import java.util.List;
  
  /**
   * CDC server.
   */
  @RequiredArgsConstructor
  public final class CDCServer extends Thread {
      
      private final List<String> addressed;
      
      private final int port;
      
      private EventLoopGroup bossGroup;
      
      private EventLoopGroup workerGroup;
      
      @Override
      public void run() {
          try {
              List<ChannelFuture> futures = startInternal(addressed, port);
              for (ChannelFuture each : futures) {
                  each.channel().closeFuture().sync();
              }
          } catch (final InterruptedException ignored) {
              Thread.currentThread().interrupt();
          } finally {
              close();
          }
      }
      
      private List<ChannelFuture> startInternal(final List<String> addresses, final int port) throws InterruptedException {
          createEventLoopGroup();
          ServerBootstrap bootstrap = new ServerBootstrap();
          bootstrap.channel(Epoll.isAvailable() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                  .group(bossGroup, workerGroup)
                  .option(ChannelOption.SO_REUSEADDR, true)
                  .option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(8 * 1024 * 1024, 16 * 1024 * 1024))
                  .childOption(ChannelOption.SO_KEEPALIVE, true)
                  .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                  .childOption(ChannelOption.TCP_NODELAY, true)
                  .handler(new LoggingHandler(LogLevel.INFO))
                  .childHandler(new CDCServerHandlerInitializer());
          List<ChannelFuture> result = new ArrayList<>();
          for (String each : addresses) {
              result.add(bootstrap.bind(each, port).sync());
          }
          return result;
      }
      
      private void createEventLoopGroup() {
          bossGroup = Epoll.isAvailable() ? new EpollEventLoopGroup(1) : new NioEventLoopGroup(1);
          workerGroup = Epoll.isAvailable() ? new EpollEventLoopGroup() : new NioEventLoopGroup();
      }
      
      private void close() {
          if (null != bossGroup) {
              bossGroup.shutdownGracefully();
          }
          if (null != workerGroup) {
              workerGroup.shutdownGracefully();
          }
      }
  }