<commit_before>package com.continuuity.gateway.v2.handlers.v2.stream;

import com.continuuity.api.common.Bytes;
import com.continuuity.api.data.OperationException;
import com.continuuity.api.data.stream.StreamSpecification;
import com.continuuity.api.flow.flowlet.StreamEvent;
import com.continuuity.app.verification.VerifyResult;
import com.continuuity.common.conf.Constants;
import com.continuuity.common.http.core.AbstractHttpHandler;
import com.continuuity.common.http.core.HandlerContext;
import com.continuuity.common.http.core.HttpResponder;
import com.continuuity.common.metrics.CMetrics;
import com.continuuity.common.metrics.MetricsHelper;
import com.continuuity.common.queue.QueueName;
import com.continuuity.data2.queue.DequeueResult;
import com.continuuity.data2.queue.QueueClientFactory;
import com.continuuity.data2.transaction.TransactionSystemClient;
import com.continuuity.gateway.GatewayMetrics;
import com.continuuity.gateway.GatewayMetricsHelperWrapper;
import com.continuuity.gateway.auth.GatewayAuthenticator;
import com.continuuity.gateway.util.StreamCache;
import com.continuuity.internal.app.verification.StreamVerification;
import com.continuuity.metadata.MetadataService;
import com.continuuity.metadata.Stream;
import com.continuuity.streamevent.DefaultStreamEvent;
import com.continuuity.streamevent.StreamEventCodec;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.cache.RemovalListener;
import com.google.common.cache.RemovalNotification;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableMultimap;
import com.google.common.collect.Maps;
import com.google.common.hash.HashCode;
import com.google.common.hash.Hashing;
import com.google.common.util.concurrent.FutureCallback;
import com.google.inject.Inject;
import org.jboss.netty.handler.codec.http.HttpRequest;
import org.jboss.netty.handler.codec.http.HttpResponseStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.PUT;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import static com.continuuity.common.metrics.MetricsHelper.Status.BadRequest;
import static com.continuuity.common.metrics.MetricsHelper.Status.Error;
import static com.continuuity.common.metrics.MetricsHelper.Status.NoData;
import static com.continuuity.common.metrics.MetricsHelper.Status.NotFound;
import static com.continuuity.common.metrics.MetricsHelper.Status.Success;

/**
 * Handler that handles stream event requests. This supports
 * POST requests to send an event to a stream, and GET requests to inspect
 * or retrieve events from the stream.
 */
@Path("/v2")
public class StreamHandler extends AbstractHttpHandler {
  private static final Logger LOG = LoggerFactory.getLogger(StreamHandler.class);
  private static final String NAME = Constants.Gateway.STREAM_HANDLER_NAME;

  private final StreamCache streamCache;
  private final MetadataService metadataService;
  private final CMetrics cMetrics;
  private final GatewayMetrics gatewayMetrics;
  private final CachedStreamEventCollector streamEventCollector;
  private final GatewayAuthenticator authenticator;

  private final LoadingCache<ConsumerKey, ConsumerHolder> queueConsumerCache;

  @Inject
  public StreamHandler(final TransactionSystemClient txClient, StreamCache streamCache,
                       MetadataService metadataService, CMetrics cMetrics, GatewayMetrics gatewayMetrics,
                       final QueueClientFactory queueClientFactory, GatewayAuthenticator authenticator,
                       CachedStreamEventCollector cachedStreamEventCollector) {
    this.streamCache = streamCache;
    this.metadataService = metadataService;
    this.cMetrics = cMetrics;
    this.gatewayMetrics = gatewayMetrics;
    this.authenticator = authenticator;

    this.streamEventCollector = cachedStreamEventCollector;

    this.queueConsumerCache = CacheBuilder.newBuilder()
      .expireAfterAccess(1, TimeUnit.HOURS)
      .removalListener(
        new RemovalListener<ConsumerKey, ConsumerHolder>() {
          @Override
          public void onRemoval(RemovalNotification<ConsumerKey, ConsumerHolder> notification) {
            ConsumerHolder holder = notification.getValue();
            if (holder == null) {
              return;
            }

            try {
              holder.close();
            } catch (IOException e) {
              LOG.error("Exception while closing consumer {}", holder, e);
            }
          }
        }
      )
      .build(
        new CacheLoader<ConsumerKey, ConsumerHolder>() {
          @Override
          public ConsumerHolder load(ConsumerKey key) throws Exception {
            return new ConsumerHolder(key, txClient, queueClientFactory);
          }
        });
  }

  @Override
  public void init(HandlerContext context) {
    LOG.info("Starting StreamHandler.");
    streamEventCollector.startAndWait();
  }

  @Override
  public void destroy(HandlerContext context) {
    LOG.info("Stopping StreamHandler.");
    streamEventCollector.stopAndWait();
  }

  @PUT
  @Path("/streams/{stream-id}")
  public void create(HttpRequest request, HttpResponder responder, @PathParam("stream-id") String destination) {
    GatewayMetricsHelperWrapper helper = new GatewayMetricsHelperWrapper(
      new MetricsHelper(this.getClass(), cMetrics, Constants.Gateway.GATEWAY_PREFIX + NAME), gatewayMetrics);
    helper.setMethod("create");
    helper.setScope(destination);

    String accountId = authenticate(request, responder, helper);
    if (accountId == null) {
      return;
    }

    if (!isId(destination)) {
      LOG.trace("Stream id '{}' is not a printable ascii character string", destination);
      responder.sendStatus(HttpResponseStatus.BAD_REQUEST);
      helper.finish(BadRequest);
      return;
    }

    Stream stream = new Stream(destination);
    stream.setName(destination);

    // Check if a stream with the same id exists
    try {
      Stream existingStream = metadataService.getStream(accountId, stream.getId());
      if (existingStream != null) {
        metadataService.createStream(accountId, stream);
      }
    } catch (Exception e) {
      LOG.trace("Error during creation of stream id '{}'", destination, e);
      responder.sendStatus(HttpResponseStatus.INTERNAL_SERVER_ERROR);
      return;
    }

    responder.sendStatus(HttpResponseStatus.OK);
    helper.finish(Success);
  }
<commit_msg>fix bug in gateway stream create<commit_after>package com.continuuity.gateway.v2.handlers.v2.stream;

import com.continuuity.api.common.Bytes;
import com.continuuity.api.data.OperationException;
import com.continuuity.api.data.stream.StreamSpecification;
import com.continuuity.api.flow.flowlet.StreamEvent;
import com.continuuity.app.verification.VerifyResult;
import com.continuuity.common.conf.Constants;
import com.continuuity.common.http.core.AbstractHttpHandler;
import com.continuuity.common.http.core.HandlerContext;
import com.continuuity.common.http.core.HttpResponder;
import com.continuuity.common.metrics.CMetrics;
import com.continuuity.common.metrics.MetricsHelper;
import com.continuuity.common.queue.QueueName;
import com.continuuity.data2.queue.DequeueResult;
import com.continuuity.data2.queue.QueueClientFactory;
import com.continuuity.data2.transaction.TransactionSystemClient;
import com.continuuity.gateway.GatewayMetrics;
import com.continuuity.gateway.GatewayMetricsHelperWrapper;
import com.continuuity.gateway.auth.GatewayAuthenticator;
import com.continuuity.gateway.util.StreamCache;
import com.continuuity.internal.app.verification.StreamVerification;
import com.continuuity.metadata.MetadataService;
import com.continuuity.metadata.Stream;
import com.continuuity.streamevent.DefaultStreamEvent;
import com.continuuity.streamevent.StreamEventCodec;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.cache.RemovalListener;
import com.google.common.cache.RemovalNotification;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableMultimap;
import com.google.common.collect.Maps;
import com.google.common.hash.HashCode;
import com.google.common.hash.Hashing;
import com.google.common.util.concurrent.FutureCallback;
import com.google.inject.Inject;
import org.jboss.netty.handler.codec.http.HttpRequest;
import org.jboss.netty.handler.codec.http.HttpResponseStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.PUT;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import static com.continuuity.common.metrics.MetricsHelper.Status.BadRequest;
import static com.continuuity.common.metrics.MetricsHelper.Status.Error;
import static com.continuuity.common.metrics.MetricsHelper.Status.NoData;
import static com.continuuity.common.metrics.MetricsHelper.Status.NotFound;
import static com.continuuity.common.metrics.MetricsHelper.Status.Success;

/**
 * Handler that handles stream event requests. This supports
 * POST requests to send an event to a stream, and GET requests to inspect
 * or retrieve events from the stream.
 */
@Path("/v2")
public class StreamHandler extends AbstractHttpHandler {
  private static final Logger LOG = LoggerFactory.getLogger(StreamHandler.class);
  private static final String NAME = Constants.Gateway.STREAM_HANDLER_NAME;

  private final StreamCache streamCache;
  private final MetadataService metadataService;
  private final CMetrics cMetrics;
  private final GatewayMetrics gatewayMetrics;
  private final CachedStreamEventCollector streamEventCollector;
  private final GatewayAuthenticator authenticator;

  private final LoadingCache<ConsumerKey, ConsumerHolder> queueConsumerCache;

  @Inject
  public StreamHandler(final TransactionSystemClient txClient, StreamCache streamCache,
                       MetadataService metadataService, CMetrics cMetrics, GatewayMetrics gatewayMetrics,
                       final QueueClientFactory queueClientFactory, GatewayAuthenticator authenticator,
                       CachedStreamEventCollector cachedStreamEventCollector) {
    this.streamCache = streamCache;
    this.metadataService = metadataService;
    this.cMetrics = cMetrics;
    this.gatewayMetrics = gatewayMetrics;
    this.authenticator = authenticator;

    this.streamEventCollector = cachedStreamEventCollector;

    this.queueConsumerCache = CacheBuilder.newBuilder()
      .expireAfterAccess(1, TimeUnit.HOURS)
      .removalListener(
        new RemovalListener<ConsumerKey, ConsumerHolder>() {
          @Override
          public void onRemoval(RemovalNotification<ConsumerKey, ConsumerHolder> notification) {
            ConsumerHolder holder = notification.getValue();
            if (holder == null) {
              return;
            }

            try {
              holder.close();
            } catch (IOException e) {
              LOG.error("Exception while closing consumer {}", holder, e);
            }
          }
        }
      )
      .build(
        new CacheLoader<ConsumerKey, ConsumerHolder>() {
          @Override
          public ConsumerHolder load(ConsumerKey key) throws Exception {
            return new ConsumerHolder(key, txClient, queueClientFactory);
          }
        });
  }

  @Override
  public void init(HandlerContext context) {
    LOG.info("Starting StreamHandler.");
    streamEventCollector.startAndWait();
  }

  @Override
  public void destroy(HandlerContext context) {
    LOG.info("Stopping StreamHandler.");
    streamEventCollector.stopAndWait();
  }

  @PUT
  @Path("/streams/{stream-id}")
  public void create(HttpRequest request, HttpResponder responder, @PathParam("stream-id") String destination) {
    GatewayMetricsHelperWrapper helper = new GatewayMetricsHelperWrapper(
      new MetricsHelper(this.getClass(), cMetrics, Constants.Gateway.GATEWAY_PREFIX + NAME), gatewayMetrics);
    helper.setMethod("create");
    helper.setScope(destination);

    String accountId = authenticate(request, responder, helper);
    if (accountId == null) {
      return;
    }

    if (!isId(destination)) {
      LOG.trace("Stream id '{}' is not a printable ascii character string", destination);
      responder.sendStatus(HttpResponseStatus.BAD_REQUEST);
      helper.finish(BadRequest);
      return;
    }

    Stream stream = new Stream(destination);
    stream.setName(destination);

    // Check if a stream with the same id exists
    try {
      Stream existingStream = metadataService.getStream(accountId, stream.getId());
      if (existingStream == null) {
        metadataService.createStream(accountId, stream);
      }
    } catch (Exception e) {
      LOG.trace("Error during creation of stream id '{}'", destination, e);
      responder.sendStatus(HttpResponseStatus.INTERNAL_SERVER_ERROR);
      return;
    }

    responder.sendStatus(HttpResponseStatus.OK);
    helper.finish(Success);
  }

{'fraction_non_alphanumeric': 0.0915013103706477, 'fraction_numerical': 0.001347809809060277, 'mean_word_length': 4.233542319749216, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, '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': '6820457', 'n_tokens_mistral': 3703, 'n_tokens_neox': 3637, 'n_words': 749}