<commit_before>/**
 * Copyright 2014 Netflix, Inc.
 *
 * Licensed 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 com.netflix.spectator.nflx;

import com.sun.net.httpserver.Headers;
import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpHandler;
import com.sun.net.httpserver.HttpServer;
import io.netty.buffer.ByteBuf;
import io.netty.handler.timeout.ReadTimeoutException;
import io.reactivex.netty.protocol.http.client.HttpClientResponse;
import io.reactivex.netty.protocol.http.client.HttpRedirectException;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
import rx.Observable;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Actions;
import rx.functions.Func1;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.URI;
import java.nio.charset.Charset;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerArray;
import java.util.concurrent.atomic.AtomicReference;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

@RunWith(JUnit4.class)
public class RxHttpTest {

  private static final String client = "test";
  private static final int retries = 2;

  private static HttpServer server;
  private static int port;

  private static AtomicInteger statusCode = new AtomicInteger(200);
  private static AtomicIntegerArray statusCounts = new AtomicIntegerArray(600);

  private static AtomicInteger redirects = new AtomicInteger(0);

  private static void ignore(InputStream input) throws IOException {
    try (InputStream in = input) {
      byte[] buf = new byte[1024];
      while (in.read(buf) > 0);
    }
  }

  private static byte[] gzip(String data) throws IOException {
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    try (GZIPOutputStream out = new GZIPOutputStream(baos)) {
      out.write(data.getBytes("UTF-8"));
    }
    return baos.toByteArray();
  }

  private static String gunzip(InputStream input) throws IOException {
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    try (GZIPInputStream in = new GZIPInputStream(input)) {
      byte[] buf = new byte[1024];
      int length;
      while ((length = in.read(buf)) > 0) {
        baos.write(buf, 0, length);
      }
    }
    return new String(baos.toByteArray(), "UTF-8");
  }

  @BeforeClass
  public static void startServer() throws Exception {
    server = HttpServer.create(new InetSocketAddress(0), 100);
    server.setExecutor(Executors.newFixedThreadPool(10, new ThreadFactory() {
      @Override public Thread newThread(Runnable r) {
        return new Thread(r, "HttpServer");
      }
    }));
    port = server.getAddress().getPort();

    server.createContext("/empty", new HttpHandler() {
      @Override
      public void handle(HttpExchange exchange) throws IOException {
        ignore(exchange.getRequestBody());
        statusCounts.incrementAndGet(statusCode.get());
        exchange.sendResponseHeaders(statusCode.get(), -1L);
        exchange.close();
      }
    });

    server.createContext("/echo", new HttpHandler() {
      @Override
      public void handle(HttpExchange exchange) throws IOException {
        Headers headers = exchange.getRequestHeaders();
        int contentLength = Integer.parseInt(headers.getFirst("Content-Length"));
        String contentEnc = headers.getFirst("Content-Encoding");
        if (contentEnc != null) {
          exchange.getResponseHeaders().add("Content-Encoding", contentEnc);
        }

        int code = statusCode.get();
        if (contentLength > 512 && !"gzip".equals(contentEnc)) {
          code = 400;
        }

        statusCounts.incrementAndGet(code);
        exchange.sendResponseHeaders(code, contentLength);
        try (InputStream input = exchange.getRequestBody();
            OutputStream output = exchange.getResponseBody()) {
          byte[] buf = new byte[1024];
          int length;
          while ((length = input.read(buf)) > 0) {
            output.write(buf, 0, length);
          }
        }
        exchange.close();
      }
    });

    server.createContext("/relativeRedirect", new HttpHandler() {
      @Override
      public void handle(HttpExchange exchange) throws IOException {
        ignore(exchange.getRequestBody());
        if (redirects.get() <= 0) {
          System.out.println("/relativeRedirect " + statusCode.get());
          statusCounts.incrementAndGet(statusCode.get());
          exchange.getResponseHeaders().add("Location", "/empty");
          exchange.sendResponseHeaders(statusCode.get(), -1L);
          exchange.close();
        } else {
          redirects.decrementAndGet();
          statusCounts.incrementAndGet(302);
          exchange.getResponseHeaders().add("Location", "/relativeRedirect");
          exchange.sendResponseHeaders(302, -1L);
          exchange.close();
        }
      }
    });

    server.createContext("/absoluteRedirect", new HttpHandler() {
      @Override
      public void handle(HttpExchange exchange) throws IOException {
        String host = "http://" + exchange.getRequestHeaders().getFirst("Host") + ":" + port;
        ignore(exchange.getRequestBody());<commit_msg>remove debug statement<commit_after>/**
 * Copyright 2014 Netflix, Inc.
 *
 * Licensed 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 com.netflix.spectator.nflx;

import com.sun.net.httpserver.Headers;
import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpHandler;
import com.sun.net.httpserver.HttpServer;
import io.netty.buffer.ByteBuf;
import io.netty.handler.timeout.ReadTimeoutException;
import io.reactivex.netty.protocol.http.client.HttpClientResponse;
import io.reactivex.netty.protocol.http.client.HttpRedirectException;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
import rx.Observable;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Actions;
import rx.functions.Func1;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.URI;
import java.nio.charset.Charset;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerArray;
import java.util.concurrent.atomic.AtomicReference;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

@RunWith(JUnit4.class)
public class RxHttpTest {

  private static final String client = "test";
  private static final int retries = 2;

  private static HttpServer server;
  private static int port;

  private static AtomicInteger statusCode = new AtomicInteger(200);
  private static AtomicIntegerArray statusCounts = new AtomicIntegerArray(600);

  private static AtomicInteger redirects = new AtomicInteger(0);

  private static void ignore(InputStream input) throws IOException {
    try (InputStream in = input) {
      byte[] buf = new byte[1024];
      while (in.read(buf) > 0);
    }
  }

  private static byte[] gzip(String data) throws IOException {
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    try (GZIPOutputStream out = new GZIPOutputStream(baos)) {
      out.write(data.getBytes("UTF-8"));
    }
    return baos.toByteArray();
  }

  private static String gunzip(InputStream input) throws IOException {
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    try (GZIPInputStream in = new GZIPInputStream(input)) {
      byte[] buf = new byte[1024];
      int length;
      while ((length = in.read(buf)) > 0) {
        baos.write(buf, 0, length);
      }
    }
    return new String(baos.toByteArray(), "UTF-8");
  }

  @BeforeClass
  public static void startServer() throws Exception {
    server = HttpServer.create(new InetSocketAddress(0), 100);
    server.setExecutor(Executors.newFixedThreadPool(10, new ThreadFactory() {
      @Override public Thread newThread(Runnable r) {
        return new Thread(r, "HttpServer");
      }
    }));
    port = server.getAddress().getPort();

    server.createContext("/empty", new HttpHandler() {
      @Override
      public void handle(HttpExchange exchange) throws IOException {
        ignore(exchange.getRequestBody());
        statusCounts.incrementAndGet(statusCode.get());
        exchange.sendResponseHeaders(statusCode.get(), -1L);
        exchange.close();
      }
    });

    server.createContext("/echo", new HttpHandler() {
      @Override
      public void handle(HttpExchange exchange) throws IOException {
        Headers headers = exchange.getRequestHeaders();
        int contentLength = Integer.parseInt(headers.getFirst("Content-Length"));
        String contentEnc = headers.getFirst("Content-Encoding");
        if (contentEnc != null) {
          exchange.getResponseHeaders().add("Content-Encoding", contentEnc);
        }

        int code = statusCode.get();
        if (contentLength > 512 && !"gzip".equals(contentEnc)) {
          code = 400;
        }

        statusCounts.incrementAndGet(code);
        exchange.sendResponseHeaders(code, contentLength);
        try (InputStream input = exchange.getRequestBody();
            OutputStream output = exchange.getResponseBody()) {
          byte[] buf = new byte[1024];
          int length;
          while ((length = input.read(buf)) > 0) {
            output.write(buf, 0, length);
          }
        }
        exchange.close();
      }
    });

    server.createContext("/relativeRedirect", new HttpHandler() {
      @Override
      public void handle(HttpExchange exchange) throws IOException {
        ignore(exchange.getRequestBody());
        if (redirects.get() <= 0) {
          statusCounts.incrementAndGet(statusCode.get());
          exchange.getResponseHeaders().add("Location", "/empty");
          exchange.sendResponseHeaders(statusCode.get(), -1L);
          exchange.close();
        } else {
          redirects.decrementAndGet();
          statusCounts.incrementAndGet(302);
          exchange.getResponseHeaders().add("Location", "/relativeRedirect");
          exchange.sendResponseHeaders(302, -1L);
          exchange.close();
        }
      }
    });

    server.createContext("/absoluteRedirect", new HttpHandler() {
      @Override
      public void handle(HttpExchange exchange) throws IOException {
        String host = "http://" + exchange.getRequestHeaders().getFirst("Host") + ":" + port;
        ignore(exchange.getRequestBody());
{'fraction_non_alphanumeric': 0.11122031122031122, 'fraction_numerical': 0.010155610155610156, 'mean_word_length': 3.500921489126428, 'pattern_counts': {'":': 2, '<': 5, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '3439509', 'n_tokens_mistral': 3432, 'n_tokens_neox': 3285, 'n_words': 827}