<commit_before>package com.outbrain.ob1k.server.netty;

import com.outbrain.ob1k.common.marshalling.RequestMarshallerRegistry;
import com.outbrain.ob1k.common.metrics.NettyQueuesGaugeBuilder;
import com.outbrain.ob1k.server.Server;
import com.outbrain.ob1k.server.StaticPathResolver;
import com.outbrain.ob1k.server.registry.ServiceRegistry;
import com.outbrain.ob1k.server.util.SyncRequestQueueObserver;
import com.outbrain.swinfra.metrics.api.MetricFactory;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpContentCompressor;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpRequestDecoder;
import io.netty.handler.codec.http.HttpResponseEncoder;
import io.netty.handler.stream.ChunkedWriteHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * User: aronen
 * Date: 6/23/13
 * Time: 10:27 AM
 */
public class NettyServer implements Server {
  private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);

  private final int port;
  private final String contextPath;
  private final RequestMarshallerRegistry marshallerRegistry;
  private final SyncRequestQueueObserver queueObserver;
  private final ChannelGroup activeChannels;
  private final long requestTimeoutMs;
  private volatile Channel channel;
  private final StaticPathResolver staticResolver;
  private final ServiceDispatcher dispatcher;
  private final EventLoopGroup nioGroup;
  private final String applicationName;
  private final boolean acceptKeepAlive;
  private final boolean supportZip;
  private final MetricFactory metricFactory;
  private final int maxContentLength;
  private final CopyOnWriteArrayList<Listener> listeners = new CopyOnWriteArrayList<>();

  public NettyServer(final int port, final ServiceRegistry registry, final RequestMarshallerRegistry marshallerRegistry,
                     final StaticPathResolver staticResolver, final SyncRequestQueueObserver queueObserver,
                     final ChannelGroup activeChannels, final String contextPath, final String applicationName,
                     final boolean acceptKeepAlive, final boolean supportZip, final MetricFactory metricFactory,
                     final int maxContentLength, final long requestTimeoutMs) {
    System.setProperty("com.outbrain.web.context.path", contextPath);
    this.port = port;
    this.staticResolver = staticResolver;
    this.queueObserver = queueObserver;
    this.activeChannels = activeChannels;
    this.contextPath = contextPath;
    this.applicationName = applicationName;
    this.marshallerRegistry = marshallerRegistry;
    this.dispatcher = new ServiceDispatcher(registry, marshallerRegistry);
    this.nioGroup = new NioEventLoopGroup();
    this.acceptKeepAlive = acceptKeepAlive;
    this.supportZip = supportZip;
    this.metricFactory = metricFactory;
    this.maxContentLength = maxContentLength;
    this.requestTimeoutMs = requestTimeoutMs;
    registry.logRegisteredEndpoints();
  }

  @Override
  public InetSocketAddress start() {
    logger.info("################## Starting OB1K server for module '{}' ##################", applicationName);
    try {
      final ServerBootstrap b = new ServerBootstrap();
      b.option(ChannelOption.SO_BACKLOG, 1024);
      b.option(ChannelOption.SO_RCVBUF, 64 * 1024);
      b.option(ChannelOption.SO_SNDBUF, 64 * 1024);
      // it means that the max static file can be 1024*ResourceRegion.BUFFER_SIZE = 64Mb
      b.childOption(ChannelOption.WRITE_SPIN_COUNT, 1024);
      b.childOption(ChannelOption.TCP_NODELAY, true);
      b.group(nioGroup)
          .channel(NioServerSocketChannel.class)
          .childHandler(new RPCServerInitializer(maxContentLength));

      channel = b.bind(port).sync().channel();
      queueObserver.setServerChannel(channel);
      NettyQueuesGaugeBuilder.registerQueueGauges(metricFactory, nioGroup, applicationName);

      final InetSocketAddress address = (InetSocketAddress) channel.localAddress();
      onStarted();
      logger.info("server is up and bounded on address: {}{}", address, getOpeningText());
      return address;
    } catch (final Exception e) {
      logger.error("failed to start server", e);
      return null;
    }
  }

  private static String getOpeningText() {
    final InputStream inputStream = NettyServer.class.getResourceAsStream("/litany-against-fear.txt");
    try (BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream))) {
      String line;
      final StringBuilder buffer = new StringBuilder();
      while ((line = reader.readLine()) != null) {
        buffer.append(line).append("\n");
      }

      return buffer.toString();
    } catch (final IOException e) {
      logger.warn("can't read opening text resource");
      return "";
    }
  }

  @Override
  public String getContextPath() {
    return contextPath;
  }

  @Override
  public void stop() {
    logger.info("################## Stopping OB1K server for module '{}' ##################", applicationName);
    queueObserver.setServerChannel(null);
    channel.close().awaitUninterruptibly(200);
    nioGroup.shutdownGracefully();
  }

  @Override
  public void addListener(final Listener listener) {
    listeners.add(listener);
  }<commit_msg>Added a shutdown hook to close the server<commit_after>package com.outbrain.ob1k.server.netty;

import com.outbrain.ob1k.common.marshalling.RequestMarshallerRegistry;
import com.outbrain.ob1k.common.metrics.NettyQueuesGaugeBuilder;
import com.outbrain.ob1k.server.Server;
import com.outbrain.ob1k.server.StaticPathResolver;
import com.outbrain.ob1k.server.registry.ServiceRegistry;
import com.outbrain.ob1k.server.util.SyncRequestQueueObserver;
import com.outbrain.swinfra.metrics.api.MetricFactory;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpContentCompressor;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpRequestDecoder;
import io.netty.handler.codec.http.HttpResponseEncoder;
import io.netty.handler.stream.ChunkedWriteHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * User: aronen
 * Date: 6/23/13
 * Time: 10:27 AM
 */
public class NettyServer implements Server {
  private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);

  private final int port;
  private final String contextPath;
  private final RequestMarshallerRegistry marshallerRegistry;
  private final SyncRequestQueueObserver queueObserver;
  private final ChannelGroup activeChannels;
  private final long requestTimeoutMs;
  private volatile Channel channel;
  private final StaticPathResolver staticResolver;
  private final ServiceDispatcher dispatcher;
  private final EventLoopGroup nioGroup;
  private final String applicationName;
  private final boolean acceptKeepAlive;
  private final boolean supportZip;
  private final MetricFactory metricFactory;
  private final int maxContentLength;
  private final CopyOnWriteArrayList<Listener> listeners = new CopyOnWriteArrayList<>();

  public NettyServer(final int port, final ServiceRegistry registry, final RequestMarshallerRegistry marshallerRegistry,
                     final StaticPathResolver staticResolver, final SyncRequestQueueObserver queueObserver,
                     final ChannelGroup activeChannels, final String contextPath, final String applicationName,
                     final boolean acceptKeepAlive, final boolean supportZip, final MetricFactory metricFactory,
                     final int maxContentLength, final long requestTimeoutMs) {
    System.setProperty("com.outbrain.web.context.path", contextPath);
    this.port = port;
    this.staticResolver = staticResolver;
    this.queueObserver = queueObserver;
    this.activeChannels = activeChannels;
    this.contextPath = contextPath;
    this.applicationName = applicationName;
    this.marshallerRegistry = marshallerRegistry;
    this.dispatcher = new ServiceDispatcher(registry, marshallerRegistry);
    this.nioGroup = new NioEventLoopGroup();
    this.acceptKeepAlive = acceptKeepAlive;
    this.supportZip = supportZip;
    this.metricFactory = metricFactory;
    this.maxContentLength = maxContentLength;
    this.requestTimeoutMs = requestTimeoutMs;
    registry.logRegisteredEndpoints();
  }

  @Override
  public InetSocketAddress start() {
    logger.info("################## Starting OB1K server for module '{}' ##################", applicationName);
    try {
      final ServerBootstrap b = new ServerBootstrap();
      b.option(ChannelOption.SO_BACKLOG, 1024);
      b.option(ChannelOption.SO_RCVBUF, 64 * 1024);
      b.option(ChannelOption.SO_SNDBUF, 64 * 1024);
      // it means that the max static file can be 1024*ResourceRegion.BUFFER_SIZE = 64Mb
      b.childOption(ChannelOption.WRITE_SPIN_COUNT, 1024);
      b.childOption(ChannelOption.TCP_NODELAY, true);
      b.group(nioGroup)
          .channel(NioServerSocketChannel.class)
          .childHandler(new RPCServerInitializer(maxContentLength));

      channel = b.bind(port).sync().channel();
      addShutdownhook();
      queueObserver.setServerChannel(channel);
      NettyQueuesGaugeBuilder.registerQueueGauges(metricFactory, nioGroup, applicationName);

      final InetSocketAddress address = (InetSocketAddress) channel.localAddress();
      onStarted();
      logger.info("server is up and bounded on address: {}{}", address, getOpeningText());
      return address;
    } catch (final Exception e) {
      logger.error("failed to start server", e);
      return null;
    }
  }

  private void addShutdownhook() {
    Runtime.getRuntime().addShutdownHook(new Thread() {
      @Override
      public void run() {
        NettyServer.this.stop();
      }
    });
  }

  private static String getOpeningText() {
    final InputStream inputStream = NettyServer.class.getResourceAsStream("/litany-against-fear.txt");
    try (BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream))) {
      String line;
      final StringBuilder buffer = new StringBuilder();
      while ((line = reader.readLine()) != null) {
        buffer.append(line).append("\n");
      }

      return buffer.toString();
    } catch (final IOException e) {
      logger.warn("can't read opening text resource");
      return "";
    }
  }

  @Override
  public String getContextPath() {
    return contextPath;
  }

  @Override
  public void stop() {
    logger.info("################## Stopping OB1K server for module '{}' ##################", applicationName);
    queueObserver.setServerChannel(null);
    channel.closeFuture().addListener(new ChannelFutureListener() {
      @Override
      public void operationComplete(ChannelFuture future) throws Exception {
        nioGroup.shutdownGracefully();
      }
    });

    channel.close();
  }

  @Override
  public void addListener(final Listener listener) {
    listeners.add(listener);
  }
{'fraction_non_alphanumeric': 0.10096071367301424, 'fraction_numerical': 0.008148910619317207, 'mean_word_length': 4.311617312072893, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5108486', 'n_tokens_mistral': 3289, 'n_tokens_neox': 3169, 'n_words': 729}