<commit_before>package com.metamx.druid.realtime.firehose;

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.metamx.common.logger.Logger;
import com.metamx.druid.indexer.data.StringInputRowParser;
import com.metamx.druid.input.InputRow;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.Properties;

/**
 * A FirehoseFactory for RabbitMQ.
 * <p/>
 * It will receive it's configuration through the realtime.spec file and expects to find a
 * consumerProps element in the firehose definition with values for a number of configuration options.
 * Below is a complete example for a RabbitMQ firehose configuration with some explanation. Options
 * that have defaults can be skipped but options with no defaults must be specified with the exception
 * of the URI property. If the URI property is set, it will override any other property that was also
 * set.
 * <p/>
 * File: <em>realtime.spec</em>
 * <pre>
 *   "firehose" : {
 *     "type" : "rabbitmq",
 *     "consumerProps" : {
 *       "host": "localhost",                 # The hostname of the RabbitMQ broker to connect to. Default: 'localhost'
 *       "port": "5672",                      # The port number to connect to on the RabbitMQ broker. Default: '5672'
 *       "username": "test-dude",             # The username to use to connect to RabbitMQ. Default: 'guest'
 *       "password": "<PASSWORD>",             # The password to use to connect to RabbitMQ. Default: 'guest'
 *       "virtualHost": "test-vhost",         # The virtual host to connect to. Default: '/'
 *       "uri": "amqp://mqserver:1234/vhost", # The URI string to use to connect to RabbitMQ. No default and not needed
 *       "exchange": "test-exchange",         # The exchange to connect to. No default
 *       "queue" : "druidtest",               # The queue to connect to or create. No default
 *       "routingKey": "#",                   # The routing key to use to bind the queue to the exchange. No default
 *       "durable": "true",                   # Whether the queue should be durable. Default: 'false'
 *       "exclusive": "false",                # Whether the queue should be exclusive. Default: 'false'
 *       "autoDelete": "false"                # Whether the queue should auto-delete on disconnect. Default: 'false'
 *     },
 *     "parser" : {
 *       "timestampSpec" : { "column" : "utcdt", "format" : "iso" },
 *       "data" : { "format" : "json" },
 *       "dimensionExclusions" : ["wp"]
 *     }
 *   },
 * </pre>
 * <p/>
 * <b>Limitations:</b> This implementation will not attempt to reconnect to the MQ broker if the
 * connection to it is lost. Furthermore it does not support any automatic failover on high availability
 * RabbitMQ clusters. This is not supported by the underlying AMQP client library and while the behavior
 * could be "faked" to some extent we haven't implemented that yet. However, if a policy is defined in
 * the RabbitMQ cluster that sets the "ha-mode" and "ha-sync-mode" properly on the queue that this
 * Firehose connects to, messages should survive an MQ broker node failure and be delivered once a
 * connection to another node is set up.
 * <p/>
 * For more information on RabbitMQ high availability please see:
 * <a href="http://www.rabbitmq.com/ha.html">http://www.rabbitmq.com/ha.html</a>.
 */
public class RabbitMQFirehoseFactory implements FirehoseFactory
{
  private static final Logger log = new Logger(RabbitMQFirehoseFactory.class);

  @JsonProperty
  private final RabbitMQFirehoseConfig config;

  @JsonProperty
  private final StringInputRowParser parser;

  @JsonProperty
  private final ConnectionFactory connectionFactory;

  @JsonCreator
  public RabbitMQFirehoseFactory(
      @JsonProperty("connection") ConnectionFactory connectionFactory,
      @JsonProperty("config") RabbitMQFirehoseConfig config,
      @JsonProperty("parser") StringInputRowParser parser
  )<commit_msg>Errr, needed to make the ConnectionFactory a JacksonifiedConnectionFactory<commit_after>package com.metamx.druid.realtime.firehose;

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.metamx.common.logger.Logger;
import com.metamx.druid.indexer.data.StringInputRowParser;
import com.metamx.druid.input.InputRow;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.Properties;

/**
 * A FirehoseFactory for RabbitMQ.
 * <p/>
 * It will receive it's configuration through the realtime.spec file and expects to find a
 * consumerProps element in the firehose definition with values for a number of configuration options.
 * Below is a complete example for a RabbitMQ firehose configuration with some explanation. Options
 * that have defaults can be skipped but options with no defaults must be specified with the exception
 * of the URI property. If the URI property is set, it will override any other property that was also
 * set.
 * <p/>
 * File: <em>realtime.spec</em>
 * <pre>
 *   "firehose" : {
 *     "type" : "rabbitmq",
 *     "consumerProps" : {
 *       "host": "localhost",                 # The hostname of the RabbitMQ broker to connect to. Default: 'localhost'
 *       "port": "5672",                      # The port number to connect to on the RabbitMQ broker. Default: '5672'
 *       "username": "test-dude",             # The username to use to connect to RabbitMQ. Default: 'guest'
 *       "password": "<PASSWORD>",             # The password to use to connect to RabbitMQ. Default: 'guest'
 *       "virtualHost": "test-vhost",         # The virtual host to connect to. Default: '/'
 *       "uri": "amqp://mqserver:1234/vhost", # The URI string to use to connect to RabbitMQ. No default and not needed
 *       "exchange": "test-exchange",         # The exchange to connect to. No default
 *       "queue" : "druidtest",               # The queue to connect to or create. No default
 *       "routingKey": "#",                   # The routing key to use to bind the queue to the exchange. No default
 *       "durable": "true",                   # Whether the queue should be durable. Default: 'false'
 *       "exclusive": "false",                # Whether the queue should be exclusive. Default: 'false'
 *       "autoDelete": "false"                # Whether the queue should auto-delete on disconnect. Default: 'false'
 *     },
 *     "parser" : {
 *       "timestampSpec" : { "column" : "utcdt", "format" : "iso" },
 *       "data" : { "format" : "json" },
 *       "dimensionExclusions" : ["wp"]
 *     }
 *   },
 * </pre>
 * <p/>
 * <b>Limitations:</b> This implementation will not attempt to reconnect to the MQ broker if the
 * connection to it is lost. Furthermore it does not support any automatic failover on high availability
 * RabbitMQ clusters. This is not supported by the underlying AMQP client library and while the behavior
 * could be "faked" to some extent we haven't implemented that yet. However, if a policy is defined in
 * the RabbitMQ cluster that sets the "ha-mode" and "ha-sync-mode" properly on the queue that this
 * Firehose connects to, messages should survive an MQ broker node failure and be delivered once a
 * connection to another node is set up.
 * <p/>
 * For more information on RabbitMQ high availability please see:
 * <a href="http://www.rabbitmq.com/ha.html">http://www.rabbitmq.com/ha.html</a>.
 */
public class RabbitMQFirehoseFactory implements FirehoseFactory
{
  private static final Logger log = new Logger(RabbitMQFirehoseFactory.class);

  @JsonProperty
  private final RabbitMQFirehoseConfig config;

  @JsonProperty
  private final StringInputRowParser parser;

  @JsonProperty
  private final ConnectionFactory connectionFactory;

  @JsonCreator
  public RabbitMQFirehoseFactory(
      @JsonProperty("connection") JacksonifiedConnectionFactory connectionFactory,
      @JsonProperty("config") RabbitMQFirehoseConfig config,
      @JsonProperty("parser") StringInputRowParser parser
  )
{'fraction_non_alphanumeric': 0.0983708493968412, 'fraction_numerical': 0.002984703395100112, 'mean_word_length': 3.1972860125260962, 'pattern_counts': {'":': 22, '<': 29, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 4, 'xml': 4}, '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': '986968', 'n_tokens_mistral': 2284, 'n_tokens_neox': 2156, 'n_words': 843}