package com.wavefront.agent;

import com.google.common.base.Function;

import com.wavefront.agent.api.ForceQueueEnabledAgentAPI;
import com.wavefront.ingester.Decoder;
import com.yammer.metrics.Metrics;
import com.yammer.metrics.core.Counter;
import com.yammer.metrics.core.MetricName;

import org.apache.commons.lang.StringUtils;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.UUID;
import java.util.regex.Pattern;

import javax.annotation.Nullable;

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import sunnylabs.report.ReportPoint;

/**
 * Parses points from a channel using the given decoder and send it off to the AgentAPI interface.
 *
 * @author Clement Pang (clement@wavefront.com).
 */
@ChannelHandler.Sharable
public class ChannelStringHandler extends SimpleChannelInboundHandler<String> {

  private final Decoder decoder;
  private final List<ReportPoint> validatedPoints = new ArrayList<>();
  private final String prefix;
  /**
   * Transformer to transform each line.
   */
  @Nullable
  private final Function<String, String> transformer;
  private final PointHandler pointHandler;

  @Nullable
  private final Pattern pointLineWhiteList;
  @Nullable
  private final Pattern pointLineBlackList;

  private final Counter regexRejects;

  private int blockedPointsPerBatch;

  public ChannelStringHandler(Decoder decoder, final ForceQueueEnabledAgentAPI agentAPI,
                              final UUID daemonId,
                              final int port,
                              final String prefix,
                              final String logLevel,
                              final String validationLevel,
                              final long millisecondsPerBatch,
                              final int blockedPointsPerBatch,
                              @Nullable final Function<String, String> transformer,
                              @Nullable final String pointLineWhiteListRegex,
                              @Nullable final String pointLineBlackListRegex) {
    this.decoder = decoder;
    this.pointHandler = new PointHandler(agentAPI, daemonId, port, logLevel, validationLevel,
        millisecondsPerBatch, blockedPointsPerBatch);

    this.prefix = prefix;
    this.blockedPointsPerBatch = blockedPointsPerBatch;
    this.transformer = transformer;

    this.pointLineWhiteList = StringUtils.isBlank(pointLineWhiteListRegex) ?
        null : Pattern.compile(pointLineWhiteListRegex);

    this.pointLineBlackList = StringUtils.isBlank(pointLineBlackListRegex) ?
        null : Pattern.compile(pointLineBlackListRegex);

    this.regexRejects = Metrics.newCounter(
        new MetricName("validationRegex." + String.valueOf(port), "", "points-rejected"));
  }

  public static final String PUSH_DATA_DELIMETER = "\n";

  public static List<String> unjoinPushData(String pushData) {
    return Arrays.asList(StringUtils.split(pushData, PUSH_DATA_DELIMETER));
  }

  public static String joinPushData(List<String> pushData) {
    return StringUtils.join(pushData, PUSH_DATA_DELIMETER);
  }

  protected boolean passesWhiteAndBlackLists(String pointLine) {
    if (pointLineWhiteList != null && !pointLineWhiteList.matcher(pointLine).matches() ||
        pointLineBlackList != null && pointLineBlackList.matcher(pointLine).matches()) {
      regexRejects.inc();
      return false;
    }
    return true;
  }

  @Override
  protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
    // ignore empty lines.
    if (msg == null || msg.trim().length() == 0) return;
    if (transformer != null) {
      msg = transformer.apply(msg);
    }
    String pointLine = msg;

    // apply white/black lists after formatting, but before prefixing
    if (!passesWhiteAndBlackLists(pointLine)) {
      handleBlockedPoint(pointLine);
      return;
    }
    if (prefix != null) {
      pointLine = prefix + "." + msg;
    }
    if (validatedPoints.size() != 0) {
      validatedPoints.clear();
    }
    try {
      decoder.decodeReportPoints(pointLine, validatedPoints, "dummy");
    } catch (Exception e) {
      handleBlockedPoint(pointLine);
    }
    if (!validatedPoints.isEmpty()) {
      ReportPoint point = validatedPoints.get(0);
      pointHandler.reportPoint(point, pointLine);
    }
  }

  private void handleBlockedPoint(String pointLine) {
    if (pointHandler.sendDataTask.getBlockedSampleSize() < this.blockedPointsPerBatch) {
      pointHandler.sendDataTask.addBlockedSample(pointLine);
    }
    pointHandler.sendDataTask.incrementBlockedPoints();
  }

  @Override
  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    // ignore.
  }
}

@@ -23,16 +23,60 @@
 agent;%0A%0A
+import com.beust.jcommander.internal.Lists;%0A
 import c
@@ -107,17 +107,16 @@
 nction;%0A
-%0A
 import c
@@ -326,396 +326,430 @@
 me;%0A
-%0Aimport org.apache.commons.lang.StringUtils;%0A%0Aimport java.util.ArrayList;%0Aimport java.util.Arrays;%0Aimport java.util.List;%0Aimport java.util.UUID;%0Aimport java.util.regex.Pattern;%0A%0Aimport javax.annotation.Nullable;%0A%0Aimport io.netty.channel.ChannelHandler;%0Aimport io.netty.channel.ChannelHandlerContext;%0Aimport io.netty.channel.SimpleChannelInboundHandler;%0Aimport sunnylabs.report.ReportPoint
+import io.netty.channel.ChannelHandler;%0Aimport io.netty.channel.ChannelHandlerContext;%0Aimport io.netty.channel.SimpleChannelInboundHandler;%0Aimport org.apache.commons.lang.StringUtils;%0Aimport sunnylabs.report.ReportPoint;%0A%0Aimport javax.annotation.Nullable;%0Aimport java.util.Arrays;%0Aimport java.util.List;%0Aimport java.util.UUID;%0Aimport java.util.logging.Level;%0Aimport java.util.logging.Logger;%0Aimport java.util.regex.Pattern
 ;%0A%0A/
@@ -1026,100 +1026,134 @@
 ate 
-final Decoder decoder;%0A  private final List%3CReportPoint%3E validatedPoints = new ArrayList%3C%3E()
+static final Logger logger = Logger.getLogger(ChannelStringHandler.class.getCanonicalName());%0A%0A  private final Decoder decoder
 ;%0A  
@@ -4143,79 +4143,56 @@
     
-if (validatedPoints.size() != 0) %7B%0A      validatedPoints.clear();%0A    %7D
+List%3CReportPoint%3E points = Lists.newArrayList();
 %0A   
@@ -4242,26 +4242,17 @@
 ntLine, 
-validatedP
+p
 oints, %22
@@ -4284,24 +4284,123 @@
 eption e) %7B%0A
+      logger.log(Level.WARNING, %22Cannot parse input line with decoder: %5C%22%22 + pointLine + %22%5C%22%22, e);%0A
       handle
@@ -4439,26 +4439,17 @@
    if (!
-validatedP
+p
 oints.is
@@ -4489,18 +4489,9 @@
 t = 
-validatedP
+p
 oint

Fix issue with thread safety regarding the use of a shared ArrayList