<commit_before>package com.senseidb.search.node;

import java.io.File;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.HashSet;
import java.util.List;

import javax.management.StandardMBean;

import org.apache.log4j.Logger;
import org.mortbay.jetty.Server;

import proj.zoie.api.DataProvider;
import scala.actors.threadpool.Arrays;

import com.linkedin.norbert.javacompat.cluster.ClusterClient;
import com.linkedin.norbert.javacompat.cluster.Node;
import com.linkedin.norbert.javacompat.network.NetworkServer;
import com.linkedin.norbert.network.NetworkingException;
import com.senseidb.conf.SenseiServerBuilder;
import com.senseidb.jmx.JmxUtil;
import com.senseidb.plugin.SenseiPluginRegistry;
import com.senseidb.search.req.AbstractSenseiRequest;
import com.senseidb.search.req.AbstractSenseiResult;
import com.senseidb.svc.impl.AbstractSenseiCoreService;
import com.senseidb.svc.impl.CoreSenseiServiceImpl;
import com.senseidb.svc.impl.SenseiCoreServiceMessageHandler;
import com.senseidb.svc.impl.SysSenseiCoreServiceImpl;


public class SenseiServer {
  private static final Logger logger = Logger.getLogger(SenseiServer.class);

    private static final String AVAILABLE = "available";
    private static final String UNAVAILABLE = "unavailable";
    private static final String DUMMY_OUT_IP = "172.16.31.10";

    private int _id;
    private int _port;
    private int[] _partitions;
    private NetworkServer _networkServer;
    private ClusterClient _clusterClient;
    private final SenseiCore _core;
    protected volatile Node _serverNode;
    private final CoreSenseiServiceImpl _innerSvc;
    private final List<AbstractSenseiCoreService<AbstractSenseiRequest, AbstractSenseiResult>> _externalSvc;

    //private Server _adminServer;

    protected volatile boolean _available = false;

    private final SenseiPluginRegistry pluginRegistry;

    public SenseiServer(int id, int port, int[] partitions,
                        NetworkServer networkServer,
                        ClusterClient clusterClient,
                        SenseiZoieFactory<?> zoieSystemFactory,
                        SenseiIndexingManager indexingManager,
                        SenseiQueryBuilderFactory queryBuilderFactory,
                        List<AbstractSenseiCoreService<AbstractSenseiRequest, AbstractSenseiResult>> externalSvc, SenseiPluginRegistry pluginRegistry)
    {
       this(port,networkServer,clusterClient,new SenseiCore(id, partitions,zoieSystemFactory, indexingManager, queryBuilderFactory),externalSvc, pluginRegistry);
    }

    public SenseiServer(int port,
            NetworkServer networkServer,
            ClusterClient clusterClient,
            SenseiCore senseiCore,
            List<AbstractSenseiCoreService<AbstractSenseiRequest, AbstractSenseiResult>> externalSvc, SenseiPluginRegistry pluginRegistry)
   {
        _core = senseiCore;
        this.pluginRegistry = pluginRegistry;
        _id = senseiCore.getNodeId();
        _port = port;
        _partitions = senseiCore.getPartitions();

        _networkServer = networkServer;
        _clusterClient = clusterClient;

        _innerSvc = new CoreSenseiServiceImpl(senseiCore);
        _externalSvc = externalSvc;
   }

  private static String help(){
    StringBuffer buffer = new StringBuffer();
    buffer.append("Usage: <conf.dir> [availability]\n");
    buffer.append("====================================\n");
    buffer.append("conf.dir - server configuration directory, required\n");
    buffer.append("availability - \"available\" or \"unavailable\", optional default is \"available\"\n");
    buffer.append("====================================\n");
    return buffer.toString();
  }

  /*
  public Collection<Zoie<BoboIndexReader,?,?>> getZoieSystems(){
    return _core.zoieSystems;
  }
  */

  public DataProvider getDataProvider()
  {
    return _core.getDataProvider();
  }

  public SenseiCore getSenseiCore()
  {
    return _core;
  }

  /*
  public void setAdminServer(Server server)
  {
    _adminServer = server;
  }

  public SenseiNodeInfo getSenseiNodeInfo()
  {
    StringBuffer adminLink = new StringBuffer();
    if (_adminServer != null && _adminServer.getConnectors() != null && _adminServer.getConnectors().length != 0)
    {
      adminLink.append("http://").append(_adminServer.getConnectors()[0].getHost()).append(":")
               .append(_adminServer.getConnectors()[0].getPort());
    }

    return new SenseiNodeInfo(_id, _partitions, _serverNode.getUrl(), adminLink.toString());
  }
  */

  public void shutdown(){
    try {
    	logger.info("shutting down node...");
        try
        {
          _core.shutdown();
          pluginRegistry.stop();
          _clusterClient.removeNode(_id);
          _clusterClient.shutdown();
          _serverNode = null;
        } catch (Exception e)
        {
          logger.warn(e.getMessage());
        } finally
        {
          if (_networkServer != null)
          {
        	  _networkServer.shutdown();
          }
        }
    } catch (Exception e) {
      logger.error(e.getMessage(),e);
    }
  }

  public void start(boolean available) throws Exception{
        _core.start();
//        ClusterClient clusterClient = ClusterClientFactory.newInstance().newZookeeperClient();
      String clusterName = _clusterClient.getServiceName();

      logger.info("Cluster Name: " + clusterName);
      logger.info("Cluster info: " + _clusterClient.toString());

      AbstractSenseiCoreService coreSenseiService = new CoreSenseiServiceImpl(_core);
      AbstractSenseiCoreService sysSenseiCoreService = new SysSenseiCoreServiceImpl(_core);
    // create the zookeeper cluster client
//    SenseiClusterClientImpl senseiClusterClient = new SenseiClusterClientImpl(clusterName, zookeeperURL, zookeeperTimeout, false);
      SenseiCoreServiceMessageHandler senseiMsgHandler =  new SenseiCoreServiceMessageHandler(coreSenseiService);
      SenseiCoreServiceMessageHandler senseiSysMsgHandler =  new SenseiCoreServiceMessageHandler(sysSenseiCoreService);

      _networkServer.registerHandler(senseiMsgHandler, coreSenseiService.getSerializer());
      _networkServer.registerHandler(senseiSysMsgHandler, sysSenseiCoreService.getSerializer());

      if (_externalSvc!=null){
    	for (AbstractSenseiCoreService svc : _externalSvc){
    		_networkServer.registerHandler(new SenseiCoreServiceMessageHandler(svc), svc.getSerializer());
    	}
    }
    HashSet<Integer> partition = new HashSet<Integer>();
    for (int partId : _partitions){
    	partition.add(partId);
    }

    boolean nodeExists = false;
    try
    {
      logger.info("waiting to connect to cluster...");
      _clusterClient.awaitConnectionUninterruptibly();
      _serverNode = _clusterClient.getNodeWithId(_id);
      nodeExists = (_serverNode != null);
      if (!nodeExists)
      {
        DatagramSocket ds = new DatagramSocket();
        ds.connect(InetAddress.getByName(DUMMY_OUT_IP), 80);
        String ipAddr = (new InetSocketAddress(ds.getLocalAddress(), _port)).toString().replaceAll("/", "");

        logger.info("Node id : " + _id + " IP address : " + ipAddr);

        _serverNode = _clusterClient.addNode(_id, ipAddr, partition);

        logger.info("added node id: " + _id);
      } else
      {
        // node exists

      }
    } catch (Exception e)
    {
      logger.error(e.getMessage(), e);
      throw e;
    }

    try
    {
      logger.info("binding server ...");
      _networkServer.bind(_id, available);

      // exponential backoff
      Thread.sleep(1000);
<commit_msg>Fix unit tests on Windows platform<commit_after>package com.senseidb.search.node;

import java.io.File;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.HashSet;
import java.util.List;

import javax.management.StandardMBean;

import org.apache.log4j.Logger;
import org.mortbay.jetty.Server;

import proj.zoie.api.DataProvider;
import scala.actors.threadpool.Arrays;

import com.linkedin.norbert.javacompat.cluster.ClusterClient;
import com.linkedin.norbert.javacompat.cluster.Node;
import com.linkedin.norbert.javacompat.network.NetworkServer;
import com.linkedin.norbert.network.NetworkingException;
import com.senseidb.conf.SenseiServerBuilder;
import com.senseidb.jmx.JmxUtil;
import com.senseidb.plugin.SenseiPluginRegistry;
import com.senseidb.search.req.AbstractSenseiRequest;
import com.senseidb.search.req.AbstractSenseiResult;
import com.senseidb.svc.impl.AbstractSenseiCoreService;
import com.senseidb.svc.impl.CoreSenseiServiceImpl;
import com.senseidb.svc.impl.SenseiCoreServiceMessageHandler;
import com.senseidb.svc.impl.SysSenseiCoreServiceImpl;


public class SenseiServer {
  private static final Logger logger = Logger.getLogger(SenseiServer.class);

    private static final String AVAILABLE = "available";
    private static final String UNAVAILABLE = "unavailable";
    private static final String DUMMY_OUT_IP = "172.16.31.10";

    private int _id;
    private int _port;
    private int[] _partitions;
    private NetworkServer _networkServer;
    private ClusterClient _clusterClient;
    private final SenseiCore _core;
    protected volatile Node _serverNode;
    private final CoreSenseiServiceImpl _innerSvc;
    private final List<AbstractSenseiCoreService<AbstractSenseiRequest, AbstractSenseiResult>> _externalSvc;

    //private Server _adminServer;

    protected volatile boolean _available = false;

    private final SenseiPluginRegistry pluginRegistry;

    public SenseiServer(int id, int port, int[] partitions,
                        NetworkServer networkServer,
                        ClusterClient clusterClient,
                        SenseiZoieFactory<?> zoieSystemFactory,
                        SenseiIndexingManager indexingManager,
                        SenseiQueryBuilderFactory queryBuilderFactory,
                        List<AbstractSenseiCoreService<AbstractSenseiRequest, AbstractSenseiResult>> externalSvc, SenseiPluginRegistry pluginRegistry)
    {
       this(port,networkServer,clusterClient,new SenseiCore(id, partitions,zoieSystemFactory, indexingManager, queryBuilderFactory),externalSvc, pluginRegistry);
    }

    public SenseiServer(int port,
            NetworkServer networkServer,
            ClusterClient clusterClient,
            SenseiCore senseiCore,
            List<AbstractSenseiCoreService<AbstractSenseiRequest, AbstractSenseiResult>> externalSvc, SenseiPluginRegistry pluginRegistry)
   {
        _core = senseiCore;
        this.pluginRegistry = pluginRegistry;
        _id = senseiCore.getNodeId();
        _port = port;
        _partitions = senseiCore.getPartitions();

        _networkServer = networkServer;
        _clusterClient = clusterClient;

        _innerSvc = new CoreSenseiServiceImpl(senseiCore);
        _externalSvc = externalSvc;
   }

  private static String help(){
    StringBuffer buffer = new StringBuffer();
    buffer.append("Usage: <conf.dir> [availability]\n");
    buffer.append("====================================\n");
    buffer.append("conf.dir - server configuration directory, required\n");
    buffer.append("availability - \"available\" or \"unavailable\", optional default is \"available\"\n");
    buffer.append("====================================\n");
    return buffer.toString();
  }

  /*
  public Collection<Zoie<BoboIndexReader,?,?>> getZoieSystems(){
    return _core.zoieSystems;
  }
  */

  public DataProvider getDataProvider()
  {
    return _core.getDataProvider();
  }

  public SenseiCore getSenseiCore()
  {
    return _core;
  }

  /*
  public void setAdminServer(Server server)
  {
    _adminServer = server;
  }

  public SenseiNodeInfo getSenseiNodeInfo()
  {
    StringBuffer adminLink = new StringBuffer();
    if (_adminServer != null && _adminServer.getConnectors() != null && _adminServer.getConnectors().length != 0)
    {
      adminLink.append("http://").append(_adminServer.getConnectors()[0].getHost()).append(":")
               .append(_adminServer.getConnectors()[0].getPort());
    }

    return new SenseiNodeInfo(_id, _partitions, _serverNode.getUrl(), adminLink.toString());
  }
  */

  public void shutdown(){
    try {
    	logger.info("shutting down node...");
        try
        {
          _core.shutdown();
          pluginRegistry.stop();
          _clusterClient.removeNode(_id);
          _clusterClient.shutdown();
          _serverNode = null;
        } catch (Exception e)
        {
          logger.warn(e.getMessage());
        } finally
        {
          if (_networkServer != null)
          {
        	  _networkServer.shutdown();
          }
        }
    } catch (Exception e) {
      logger.error(e.getMessage(),e);
    }
  }

  public void start(boolean available) throws Exception{
        _core.start();
//        ClusterClient clusterClient = ClusterClientFactory.newInstance().newZookeeperClient();
      String clusterName = _clusterClient.getServiceName();

      logger.info("Cluster Name: " + clusterName);
      logger.info("Cluster info: " + _clusterClient.toString());

      AbstractSenseiCoreService coreSenseiService = new CoreSenseiServiceImpl(_core);
      AbstractSenseiCoreService sysSenseiCoreService = new SysSenseiCoreServiceImpl(_core);
    // create the zookeeper cluster client
//    SenseiClusterClientImpl senseiClusterClient = new SenseiClusterClientImpl(clusterName, zookeeperURL, zookeeperTimeout, false);
      SenseiCoreServiceMessageHandler senseiMsgHandler =  new SenseiCoreServiceMessageHandler(coreSenseiService);
      SenseiCoreServiceMessageHandler senseiSysMsgHandler =  new SenseiCoreServiceMessageHandler(sysSenseiCoreService);

      _networkServer.registerHandler(senseiMsgHandler, coreSenseiService.getSerializer());
      _networkServer.registerHandler(senseiSysMsgHandler, sysSenseiCoreService.getSerializer());

      if (_externalSvc!=null){
    	for (AbstractSenseiCoreService svc : _externalSvc){
    		_networkServer.registerHandler(new SenseiCoreServiceMessageHandler(svc), svc.getSerializer());
    	}
    }
    HashSet<Integer> partition = new HashSet<Integer>();
    for (int partId : _partitions){
    	partition.add(partId);
    }

    boolean nodeExists = false;
    try
    {
      logger.info("waiting to connect to cluster...");
      _clusterClient.awaitConnectionUninterruptibly();
      _serverNode = _clusterClient.getNodeWithId(_id);
      nodeExists = (_serverNode != null);
      if (!nodeExists)
      {
        DatagramSocket ds = new DatagramSocket();
        ds.connect(InetAddress.getByName(DUMMY_OUT_IP), 80);
        String inetAddress = new InetSocketAddress(ds.getLocalAddress(), _port).toString();
		if (inetAddress.trim().startsWith("0.0.0.0/")) {
			 inetAddress = new InetSocketAddress(InetAddress.getLocalHost().getHostAddress(), _port).toString();
		}
        String ipAddr = inetAddress.replaceAll("/", "");

        logger.info("Node id : " + _id + " IP address : " + ipAddr);

        _serverNode = _clusterClient.addNode(_id, ipAddr, partition);

        logger.info("added node id: " + _id);
      } else
      {
        // node exists

      }
    } catch (Exception e)
    {
      logger.error(e.getMessage(), e);
      throw e;
    }

    try
    {
      logger.info("binding server ...");
      _networkServer.bind(_id, available);

      // exponential backoff
      Thread.sleep(1000);

{'fraction_non_alphanumeric': 0.10770226537216829, 'fraction_numerical': 0.0027184466019417475, 'mean_word_length': 3.4527377521613833, 'pattern_counts': {'":': 2, '<': 27, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 3, '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': '2654049', 'n_tokens_mistral': 4520, 'n_tokens_neox': 4349, 'n_words': 832}