<commit_before>package com.orientechnologies.orient.setup;

import com.google.common.io.ByteStreams;
import com.orientechnologies.orient.core.db.OrientDB;
import com.orientechnologies.orient.core.db.OrientDBConfig;
import io.kubernetes.client.PortForward;
import io.kubernetes.client.openapi.ApiClient;
import io.kubernetes.client.openapi.ApiException;
import io.kubernetes.client.openapi.Configuration;
import io.kubernetes.client.util.Config;
import org.joda.time.DateTime;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;

import static com.orientechnologies.orient.setup.TestSetupUtil.log;

public class PortForwarder {
//  private Server pf;
  private KubectlPortForwarder pf;
  private String namespace, podName;
  private int targetPort;
  private String pfId;

  public PortForwarder(String namespace, String podName, int targetPort)
      throws IOException, ApiException {
    this.namespace = namespace;
    this.podName = podName;
    this.targetPort = targetPort;
    pfId = String.format("%s/%s:%d", namespace, podName, targetPort);
    ApiClient client = Config.defaultClient();
    Configuration.setDefaultApiClient(client);
//    PortForward forward = new PortForward();
//    List<Integer> ports = new ArrayList<>();
//    ports.add(targetPort);
//    final PortForward.PortForwardResult result = forward.forward(namespace, podName, ports);
//    pf = new Server(result.getInputStream(targetPort), result.getOutboundStream(targetPort));
    pf = new KubectlPortForwarder(namespace, podName, targetPort);
  }

  public int start() throws IOException {
    int localPort = pf.start();
    log("  Forwarding port %s -> localhost:%d...", pfId, localPort);
    return localPort;
  }

  public void stop() {
    pf.stop();
  }

  private class KubectlPortForwarder {
    private String namespace, podName;
    private int targetPort;
    private Process process;

    public KubectlPortForwarder(String namespace, String podName, int targetPort) {
      this.namespace = namespace;
      this.podName = podName;
      this.targetPort = targetPort;
    }

    public int start() throws IOException {
      for (int i = 0; i < 5; i++) {
        int port = getFreePort();
        String cmd =
            String.format(
                "kubectl -n %s port-forward %s %d:%d", namespace, podName, port, targetPort);
        log("  Running command: " + cmd);
        process = Runtime.getRuntime().exec(cmd);
        boolean stdout = false, stderr = false;
//        System.out.printf("Waiting for command to run '%s' ...\n", cmd);
        while (!stdout && !stderr) {
          try {
            Thread.sleep(1000);
          } catch (InterruptedException e) {
          }
          stdout = process.getInputStream().available() > 0;
          stderr = process.getErrorStream().available() > 0;
        }
        if (stderr) {
          BufferedReader reader =
              new BufferedReader(new InputStreamReader(process.getErrorStream()));
          log("  Error setting up kubectl port-forward: " + reader.readLine());
        } else if (stdout) {
          BufferedReader reader =
              new BufferedReader(new InputStreamReader(process.getInputStream()));
          log("  Output of kubectl port-forward: " + reader.readLine());
          if (targetPort != 2424) {
            return port;
          }
          // check if working
          OrientDB remote = new OrientDB(String.format("remote:localhost:%d", port), "root", "test", OrientDBConfig.defaultConfig());
          try {
            remote.list();
            return port;
          } catch (Exception e) {
          } finally {
            remote.close();
          }
        }
        process.destroy();
        log("  Trying kubectl again!");
        try {
          Thread.sleep(5000);
        } catch (InterruptedException e) {
        }
      }
      throw new TestSetupException("Cannot setup kubectl port-forward.");
    }

    public void stop() {
      process.destroy();
    }

    private int getFreePort() {
      try (ServerSocket ss = new ServerSocket(0)) {
        ss.setReuseAddress(true);
        return ss.getLocalPort();
      } catch (IOException e) {
        throw new TestSetupException("Error getting a free port.", e);
      }
    }
  }

  private class Server {
    private Thread inputCopier;
    private Thread outputCopier;
    private Thread server;
    private ServerSocket ss;
    private InputStream targetInputStream;
    private OutputStream targetOutputStream;
    private volatile boolean active;

    public Server(InputStream targetInputStream, OutputStream targetOutputStream)
        throws IOException {
      ss = new ServerSocket(0);
      this.targetInputStream = targetInputStream;
      this.targetOutputStream = targetOutputStream;
    }

    public int start() {
      active = true;
      server =
          new Thread(
              () -> {
                try {
                  runCopiers();
                } catch (IOException e) {
                  throw new TestSetupException(e);
                }
              });
      server.start();
      return ss.getLocalPort();
    }

    private void runCopiers() throws IOException {
      System.out.printf("Waiting for requests on port %d.\n", ss.getLocalPort());
      while (active) {
        final Socket s = ss.accept();

        inputCopier =
            new Thread(
                () -> {
                  try {
                    ByteStreams.copy(targetInputStream, s.getOutputStream());
                  } catch (Exception e) {
                    System.err.println("Error while copying target input stream.");
                    e.printStackTrace();
                    throw new TestSetupException(e);
                  }
                  System.out.printf("Exiting target input copier %s->%s.\n", pfId, s.getPort());
                });
        outputCopier =
            new Thread(
                () -> {
                  try {
                    ByteStreams.copy(s.getInputStream(), targetOutputStream);
                  } catch (Exception e) {
                    System.err.println("Error while copying local input stream.");
                    e.printStackTrace();
                    throw new TestSetupException(e);
                  }
                  System.out.printf("Exiting local input copier %s->%s.\n", pfId, s.getPort());
                });

        inputCopier.start();
        outputCopier.start();
        System.out.printf("Started copying threads for %s->%d.\n", pfId, s.getPort());
      }
      System.out.printf("Stopping port forwarder %s...\n", pfId);
    }

    public void stop() throws IOException {
      active = false;
      ss.close();
    }
  }
}<commit_msg>remove old port forwarder<commit_after>package com.orientechnologies.orient.setup;

import com.orientechnologies.orient.core.db.OrientDB;
import com.orientechnologies.orient.core.db.OrientDBConfig;
import io.kubernetes.client.openapi.ApiClient;
import io.kubernetes.client.openapi.ApiException;
import io.kubernetes.client.openapi.Configuration;
import io.kubernetes.client.util.Config;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.ServerSocket;

import static com.orientechnologies.orient.setup.TestSetupUtil.log;

public class PortForwarder {
  private KubectlPortForwarder pf;
  private String pfId;

  public PortForwarder(String namespace, String podName, int targetPort)
      throws IOException, ApiException {
    pfId = String.format("%s/%s:%d", namespace, podName, targetPort);
    ApiClient client = Config.defaultClient();
    Configuration.setDefaultApiClient(client);
    pf = new KubectlPortForwarder(namespace, podName, targetPort);
  }

  public int start() throws IOException {
    int localPort = pf.start();
    log("  Forwarding port %s -> localhost:%d...", pfId, localPort);
    return localPort;
  }

  public void stop() {
    pf.stop();
  }

  private class KubectlPortForwarder {
    private String namespace, podName;
    private int targetPort;
    private Process process;

    public KubectlPortForwarder(String namespace, String podName, int targetPort) {
      this.namespace = namespace;
      this.podName = podName;
      this.targetPort = targetPort;
    }

    public int start() throws IOException {
      for (int i = 0; i < 5; i++) {
        int port = getFreePort();
        String cmd =
            String.format(
                "kubectl -n %s port-forward %s %d:%d", namespace, podName, port, targetPort);
        log("  Running command: " + cmd);
        process = Runtime.getRuntime().exec(cmd);
        boolean stdout = false, stderr = false;
        // todo: have a timeout
        while (!stdout && !stderr) {
          try {
            Thread.sleep(1000);
          } catch (InterruptedException e) {
          }
          stdout = process.getInputStream().available() > 0;
          stderr = process.getErrorStream().available() > 0;
        }
        if (stderr) {
          BufferedReader reader =
              new BufferedReader(new InputStreamReader(process.getErrorStream()));
          log("  Error setting up kubectl port-forward: " + reader.readLine());
        } else if (stdout) {
          BufferedReader reader =
              new BufferedReader(new InputStreamReader(process.getInputStream()));
          log("  Output of kubectl port-forward: " + reader.readLine());
          if (targetPort != 2424) {
            return port;
          }
          // check if working
          OrientDB remote =
              new OrientDB(
                  String.format("remote:localhost:%d", port),
                  "root",
                  "test",
                  OrientDBConfig.defaultConfig());
          try {
            remote.list();
            return port;
          } catch (Exception e) {
          } finally {
            remote.close();
          }
        }
        process.destroy();
        log("  Trying kubectl again!");
        try {
          Thread.sleep(5000);
        } catch (InterruptedException e) {
        }
      }
      throw new TestSetupException("Cannot setup kubectl port-forward.");
    }

    public void stop() {
      process.destroy();
    }

    private int getFreePort() {
      try (ServerSocket ss = new ServerSocket(0)) {
        ss.setReuseAddress(true);
        return ss.getLocalPort();
      } catch (IOException e) {
        throw new TestSetupException("Error getting a free port.", e);
      }
    }
  }
}
{'fraction_non_alphanumeric': 0.10955401950572863, 'fraction_numerical': 0.0033140801060505634, 'mean_word_length': 2.2161997563946407, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 17, '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': '3406050', 'n_tokens_mistral': 2919, 'n_tokens_neox': 2808, 'n_words': 715}