<commit_before>/*
 * Copyright 2012 Metamarkets Group 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.metamx.emitter.core;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Charsets;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.metamx.common.ISE;
import com.metamx.common.lifecycle.LifecycleStart;
import com.metamx.common.lifecycle.LifecycleStop;
import com.metamx.common.logger.Logger;
import com.metamx.http.client.HttpClient;
import com.metamx.http.client.Request;
import com.metamx.http.client.response.StatusResponseHandler;
import com.metamx.http.client.response.StatusResponseHolder;
import org.jboss.netty.handler.codec.http.HttpMethod;

import java.io.ByteArrayOutputStream;
import java.io.Closeable;
import java.io.Flushable;
import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

public class HttpPostEmitter implements Flushable, Closeable, Emitter
{
  private static final int MAX_EVENT_SIZE = 1023 * 1024; // Set max size slightly less than 1M to allow for metadata
  private static final long BUFFER_FULL_WARNING_THROTTLE = 30000;

  private static final Logger log = new Logger(HttpPostEmitter.class);
  private static final AtomicInteger instanceCounter = new AtomicInteger();

  private final HttpEmitterConfig config;
  private final HttpClient client;
  private final ObjectMapper jsonMapper;
  private final URL url;

  private final AtomicReference<List<byte[]>> eventsList =
      new AtomicReference<List<byte[]>>(Lists.<byte[]>newLinkedList());
  private final AtomicInteger count = new AtomicInteger(0);
  private final AtomicLong bufferedSize = new AtomicLong(0);
  private final ScheduledExecutorService exec = Executors.newSingleThreadScheduledExecutor(
      new ThreadFactoryBuilder()
          .setDaemon(true)
          .setNameFormat(String.format("HttpPostEmitter-%s-%%s", instanceCounter.incrementAndGet()))
          .build()
  );
  private final AtomicLong version = new AtomicLong(0);
  private final AtomicBoolean started = new AtomicBoolean(false);

  // Trackers for buffer-full warnings. Only use under synchronized(eventsList).
  private long lastBufferFullWarning = 0;
  private long messagesDroppedSinceLastBufferFullWarning = 0;

  public HttpPostEmitter(
      HttpEmitterConfig config,
      HttpClient client
  )
  {
    this(config, client, new ObjectMapper());
  }

  public HttpPostEmitter(
      HttpEmitterConfig config,
      HttpClient client,
      ObjectMapper jsonMapper
  )
  {
    final int batchOverhead = config.getBatchingStrategy().batchStart().length +
                              config.getBatchingStrategy().batchEnd().length;
    Preconditions.checkArgument(
        config.getMaxBatchSize() >= MAX_EVENT_SIZE + batchOverhead,
        String.format(
            "maxBatchSize must be greater than MAX_EVENT_SIZE[%,d] + overhead[%,d].",
            MAX_EVENT_SIZE,
            batchOverhead
        )
    );
    Preconditions.checkArgument(
        config.getMaxBufferSize() >= MAX_EVENT_SIZE,
        String.format(
            "maxBufferSize must be greater than MAX_EVENT_SIZE[%,d].",
            MAX_EVENT_SIZE
        )
    );
    this.config = config;
    this.client = client;
    this.jsonMapper = jsonMapper;
    try {
      this.url = new URL(config.getRecipientBaseUrl());
    }
    catch (MalformedURLException e) {
      throw new ISE(e, "Bad URL: %s", config.getRecipientBaseUrl());
    }
  }

  @Override
  @LifecycleStart
  public void start()
  {
    synchronized (started) {
      if (!started.getAndSet(true)) {
        exec.schedule(
            new EmittingRunnable(version.get()),
            config.getFlushMillis(),
            TimeUnit.MILLISECONDS
        );
      }
    }
  }

  @Override
  public void emit(Event event)
  {
    synchronized (started) {
      if (!started.get()) {
        throw new RejectedExecutionException("Service is closed.");
      }
    }

    final byte[] eventBytes;
    try {
      eventBytes = jsonMapper.writeValueAsBytes(event);
    }
    catch (IOException e) {
      throw Throwables.propagate(e);
    }

    if (eventBytes.length > MAX_EVENT_SIZE) {
      log.error(
          "Event too large to emit (%,d > %,d): %s ...",
          eventBytes.length,
          MAX_EVENT_SIZE,
          new String(eventBytes, 0, 1024)
      );
      return;
    }

    synchronized (eventsList) {
      if (bufferedSize.get() + eventBytes.length <= config.getMaxBufferSize()) {
        eventsList.get().add(eventBytes);
        bufferedSize.addAndGet(eventBytes.length);
        if (!event.isSafeToBuffer() || count.incrementAndGet() >= config.getFlushCount()) {
          exec.execute(new EmittingRunnable(version.get()));
        }
      } else {
        messagesDroppedSinceLastBufferFullWarning++;
      }

      final long now = System.currentTimeMillis();
      if (messagesDroppedSinceLastBufferFullWarning > 0 && lastBufferFullWarning + BUFFER_FULL_WARNING_THROTTLE < now) {
        log.error("Buffer full: dropped %,d events!", messagesDroppedSinceLastBufferFullWarning);
        lastBufferFullWarning = now;<commit_msg>Change error handling to use utf8<commit_after>/*
 * Copyright 2012 Metamarkets Group 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.metamx.emitter.core;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Charsets;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.metamx.common.ISE;
import com.metamx.common.StringUtils;
import com.metamx.common.lifecycle.LifecycleStart;
import com.metamx.common.lifecycle.LifecycleStop;
import com.metamx.common.logger.Logger;
import com.metamx.http.client.HttpClient;
import com.metamx.http.client.Request;
import com.metamx.http.client.response.StatusResponseHandler;
import com.metamx.http.client.response.StatusResponseHolder;
import org.jboss.netty.handler.codec.http.HttpMethod;

import java.io.ByteArrayOutputStream;
import java.io.Closeable;
import java.io.Flushable;
import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

public class HttpPostEmitter implements Flushable, Closeable, Emitter
{
  private static final int MAX_EVENT_SIZE = 1023 * 1024; // Set max size slightly less than 1M to allow for metadata
  private static final long BUFFER_FULL_WARNING_THROTTLE = 30000;

  private static final Logger log = new Logger(HttpPostEmitter.class);
  private static final AtomicInteger instanceCounter = new AtomicInteger();

  private final HttpEmitterConfig config;
  private final HttpClient client;
  private final ObjectMapper jsonMapper;
  private final URL url;

  private final AtomicReference<List<byte[]>> eventsList =
      new AtomicReference<List<byte[]>>(Lists.<byte[]>newLinkedList());
  private final AtomicInteger count = new AtomicInteger(0);
  private final AtomicLong bufferedSize = new AtomicLong(0);
  private final ScheduledExecutorService exec = Executors.newSingleThreadScheduledExecutor(
      new ThreadFactoryBuilder()
          .setDaemon(true)
          .setNameFormat(String.format("HttpPostEmitter-%s-%%s", instanceCounter.incrementAndGet()))
          .build()
  );
  private final AtomicLong version = new AtomicLong(0);
  private final AtomicBoolean started = new AtomicBoolean(false);

  // Trackers for buffer-full warnings. Only use under synchronized(eventsList).
  private long lastBufferFullWarning = 0;
  private long messagesDroppedSinceLastBufferFullWarning = 0;

  public HttpPostEmitter(
      HttpEmitterConfig config,
      HttpClient client
  )
  {
    this(config, client, new ObjectMapper());
  }

  public HttpPostEmitter(
      HttpEmitterConfig config,
      HttpClient client,
      ObjectMapper jsonMapper
  )
  {
    final int batchOverhead = config.getBatchingStrategy().batchStart().length +
                              config.getBatchingStrategy().batchEnd().length;
    Preconditions.checkArgument(
        config.getMaxBatchSize() >= MAX_EVENT_SIZE + batchOverhead,
        String.format(
            "maxBatchSize must be greater than MAX_EVENT_SIZE[%,d] + overhead[%,d].",
            MAX_EVENT_SIZE,
            batchOverhead
        )
    );
    Preconditions.checkArgument(
        config.getMaxBufferSize() >= MAX_EVENT_SIZE,
        String.format(
            "maxBufferSize must be greater than MAX_EVENT_SIZE[%,d].",
            MAX_EVENT_SIZE
        )
    );
    this.config = config;
    this.client = client;
    this.jsonMapper = jsonMapper;
    try {
      this.url = new URL(config.getRecipientBaseUrl());
    }
    catch (MalformedURLException e) {
      throw new ISE(e, "Bad URL: %s", config.getRecipientBaseUrl());
    }
  }

  @Override
  @LifecycleStart
  public void start()
  {
    synchronized (started) {
      if (!started.getAndSet(true)) {
        exec.schedule(
            new EmittingRunnable(version.get()),
            config.getFlushMillis(),
            TimeUnit.MILLISECONDS
        );
      }
    }
  }

  @Override
  public void emit(Event event)
  {
    synchronized (started) {
      if (!started.get()) {
        throw new RejectedExecutionException("Service is closed.");
      }
    }

    final byte[] eventBytes;
    try {
      eventBytes = jsonMapper.writeValueAsBytes(event);
    }
    catch (IOException e) {
      throw Throwables.propagate(e);
    }

    if (eventBytes.length > MAX_EVENT_SIZE) {
      log.error(
          "Event too large to emit (%,d > %,d): %s ...",
          eventBytes.length,
          MAX_EVENT_SIZE,
          StringUtils.fromUtf8(ByteBuffer.wrap(eventBytes), 1024)
      );
      return;
    }

    synchronized (eventsList) {
      if (bufferedSize.get() + eventBytes.length <= config.getMaxBufferSize()) {
        eventsList.get().add(eventBytes);
        bufferedSize.addAndGet(eventBytes.length);
        if (!event.isSafeToBuffer() || count.incrementAndGet() >= config.getFlushCount()) {
          exec.execute(new EmittingRunnable(version.get()));
        }
      } else {
        messagesDroppedSinceLastBufferFullWarning++;
      }

      final long now = System.currentTimeMillis();
      if (messagesDroppedSinceLastBufferFullWarning > 0 && lastBufferFullWarning + BUFFER_FULL_WARNING_THROTTLE < now) {
        log.error("Buffer full: dropped %,d events!", messagesDroppedSinceLastBufferFullWarning);
        lastBufferFullWarning = now;
{'fraction_non_alphanumeric': 0.09649817778482016, 'fraction_numerical': 0.005308192045634606, 'mean_word_length': 3.522751701898961, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 2}, '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': '2650277', 'n_tokens_mistral': 3652, 'n_tokens_neox': 3539, 'n_words': 858}