public class KryoNetParameters implements Serializable {
  private static final long serialVersionUID = -5348116647457735026L;
  /** Configuration attribute base name */
  public static final String CONFIG_BASE_NAME = "kryonet";
  /** Configuration attribute name for {@link #getConnectionTimeout()} */
  public static final String CONNECTION_TIMEOUT = "connection.timeout.in.ms";
  /** Default value for {@value #CONNECTION_TIMEOUT} */
  public static final int CONNECTION_TIMEOUT_DEFAULT = 10000;
  /** Configuration attribute name for {@link #getRetrySleepDelay()} */
  public static final String RETRY_SLEEP_DELAY = "retry.sleep.delay.in.ms";
  /** Default value for {@value #RETRY_SLEEP_DELAY} */
  public static final int RETRY_SLEEP_DELAY_DEFAULT = 50;
  /** Configuration attribute name for {@link #getReconnectionDelay()} */
  public static final String RECONNECTION_DELAY = "reconnection.delay.in.ms";
  /** Default value for {@value #RECONNECTION_DELAY} */
  public static final int RECONNECTION_DELAY_DEFAULT = 2000;
  /** Configuration attribute name for {@link #getTerminationTimeout()} */
  public static final String TERMINATION_TO = "termination.timeout.in.ms";
  /** Default value for {@value #TERMINATION_TO} */
  public static final int TERMINATION_TO_DEFAULT = 10000;
  /** Configuration attribute name for {@link #getMaxRetries()} */
  public static final String MAX_RETRIES = "max.retries";
  /** Default value for {@value #MAX_RETRIES} */
  public static final int MAX_RETRIES_DEFAULT = 3;

  protected int connectionTimeout;
  protected int retrySleepDelay;
  protected int reconnectionDelay;
  protected int terminationTimeout;
  protected int maxRetries;

  /**
   * Builds a new {@link KryoNetParameters} based on a Configuration
   * 
   * @param config
   *          The configuration to use
   * @return The newly created {@link KryoNetParameters}
   * @throws F2SConfigurationException
   *           If the configuration specified is invalid
   */
  public static KryoNetParameters from(final Configuration config) throws F2SConfigurationException {
    KryoNetParameters result = new KryoNetParameters();
    try {
      result.setConnectionTimeout(config.getInt(CONNECTION_TIMEOUT, CONNECTION_TIMEOUT_DEFAULT));
    } catch (Exception e) {
      throw F2SConfigurationException.with(config, CONNECTION_TIMEOUT, e);
    }
    try {
      result.setTerminationTimeout(config.getInt(TERMINATION_TO, TERMINATION_TO_DEFAULT));
    } catch (Exception e) {
      throw F2SConfigurationException.with(config, TERMINATION_TO, e);
    }
    try {
      result.setReconnectionDelay(config.getInt(RECONNECTION_DELAY, RECONNECTION_DELAY_DEFAULT));
    } catch (Exception e) {
      throw F2SConfigurationException.with(config, RECONNECTION_DELAY, e);
    }
    try {
      result.setRetrySleepDelay(config.getInt(RETRY_SLEEP_DELAY, RETRY_SLEEP_DELAY_DEFAULT));
    } catch (Exception e) {
      throw F2SConfigurationException.with(config, RETRY_SLEEP_DELAY, e);
    }
    try {
      result.setMaxRetries(config.getInt(MAX_RETRIES, MAX_RETRIES_DEFAULT));
    } catch (Exception e) {
      throw F2SConfigurationException.with(config, MAX_RETRIES, e);
    }
    return result;
  }

  /**
   * Empty constructor - initializes with default values when available
   */
  public KryoNetParameters() {
    connectionTimeout = CONNECTION_TIMEOUT_DEFAULT;
    retrySleepDelay = RETRY_SLEEP_DELAY_DEFAULT;
    reconnectionDelay = RECONNECTION_DELAY_DEFAULT;
    terminationTimeout = TERMINATION_TO_DEFAULT;
    maxRetries = MAX_RETRIES_DEFAULT;
  }

  /**
   * @return The timeout for a connection attempt, in milliseconds
   */
  public int getConnectionTimeout() {
    return connectionTimeout;
  }

  /**
   * @param connectionTo
   *          See #getConnectionTimeout()
   * @return This object
   */
  public KryoNetParameters setConnectionTimeout(int connectionTo) {
    Preconditions.checkArgument(connectionTo > 0, "KryoNet connection timeout must be strictly positive");
    connectionTimeout = connectionTo;
    return this;
  }

  /**
   * @return The delay allowed for the client to terminate, in milliseconds. It
   *         should be greater than the connection timeout
   */
  public int getTerminationTimeout() {
    return terminationTimeout;
  }

  /**
   * @param terminationTo
   *          See #getTerminationTimeout()
   * @return This object
   */
  public KryoNetParameters setTerminationTimeout(int terminationTo) {
    Preconditions.checkArgument(terminationTo > 0, "KryoNet termination timeout must be strictly positive");
    terminationTimeout = terminationTo;
    return this;
  }

  /**
   * @return The time in milliseconds to wait before the client (i.e.
   *         EventReceptor) retries to connect to the server (i.e. EventSender)
   */
  public int getReconnectionDelay() {
    return reconnectionDelay;
  }

  /**
   * @param reconnectionDelay
   *          See #getReconnectionDelay()
   * @return This object
   */
  public KryoNetParameters setReconnectionDelay(int reconnectionDelay) {
    Preconditions.checkArgument(reconnectionDelay > 0, "KryoNet reconnection delay must be strictly positive");
    this.reconnectionDelay = reconnectionDelay;
    return this;
  }

  /**
   * @return The time in milliseconds to wait before the server (i.e.
   *         EventSender) retries sending an event to the same client (i.e.
   *         EventReceptor)
   */
  public int getRetrySleepDelay() {
    return retrySleepDelay;
  }

  /**
   * @param retrySleepDelay
   *          See #getRetrySleepDelay()
   * @return This object
   */
  public KryoNetParameters setRetrySleepDelay(int retrySleepDelay) {
    Preconditions.checkArgument(retrySleepDelay > 0, "KryoNet retry sleep delay must be strictly positive");
    this.retrySleepDelay = retrySleepDelay;
    return this;
  }

  /**
   * @return The maximum number of times an event is attempted to be sent before
   *         considering it failed
   */
  public int getMaxRetries() {
    return maxRetries;
  }

  /**
   * @param maxRetries
   *          See #getMaxRetries()
   * @return This object
   */
  public KryoNetParameters setMaxRetries(int maxRetries) {
    Preconditions.checkArgument(maxRetries > 0, "KryoNet maximum number of retries must be strictly positive");
    this.maxRetries = maxRetries;
    return this;
  }

  /**
   * @see java.lang.Object#hashCode()
   */
  @Override
  public int hashCode() {
    return new HashCodeBuilder().append(connectionTimeout).append(retrySleepDelay).append(reconnectionDelay)
        .append(terminationTimeout).append(maxRetries).hashCode();
  }

  /**
   * @see java.lang.Object#equals(java.lang.Object)
   */
  @Override
  public boolean equals(Object obj) {
    if (this == obj)
      return true;
    if (obj == null)
      return false;
    if (getClass() != obj.getClass())
      return false;
    KryoNetParameters other = (KryoNetParameters) obj;
    return new EqualsBuilder().append(this.connectionTimeout, other.connectionTimeout)
        .append(this.retrySleepDelay, other.retrySleepDelay).append(this.reconnectionDelay, other.reconnectionDelay)
        .append(this.terminationTimeout, other.terminationTimeout).append(this.maxRetries, other.maxRetries).isEquals();
  }

  /**
   * @see java.lang.Object#toString()
   */
  @Override
  public String toString() {
    return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE).append("connectionTimeout", connectionTimeout)
        .append("retrySleepDelay", retrySleepDelay).append("reconnectionDelay", reconnectionDelay)
        .append("terminationTimeout", terminationTimeout).append("maxRetries", maxRetries).toString();
  }
} public interface ConnectionParameters {
  /**
   * @return A unique identifier of the associated {@link EventSender} /
   *         {@link ServiceProvider} (hostname:port for instance)
   * 
   */
  String getId();
} public class KryoNetConnectionParameters implements ConnectionParameters, Serializable {
  private static final long serialVersionUID = -5178954514610562054L;
  /** Configuration attribute name for {@link #getAddress()} */
  public static final String ADDRESS = "address";
  /** Default value for {@value #ADDRESS} */
  public static final String ADDRESS_DEFAULT = "localhost";
  /** Configuration attribute name for {@link #getPort()} */
  public static final String PORT = "port";
  /** Default value for {@value #PORT} */
  public static final int PORT_DEFAULT = 7000;
  /** Configuration attribute name for {@link #getObjectBufferSize()} */
  public static final String OBJECT_BUFFER_SZ = "object.buffer.size";
  /** Default value for {@value #OBJECT_BUFFER_SZ} */
  public static final int OBJECT_BUFFER_SIZE_DEFAULT = 3024;
  /** Configuration attribute name for {@link #getWriteBufferSize()} */
  public static final String WRITE_BUFFER_SZ = "write.buffer.size";
  /** Default value for {@value #WRITE_BUFFER_SZ} */
  public static final int WRITE_BUFFER_SIZE_DEFAULT = OBJECT_BUFFER_SIZE_DEFAULT * 1000;

  protected String address;
  protected int port;
  protected int objectBufferSize;
  protected int writeBufferSize;

  /**
   * @param config
   *          The configuration to use
   * @return The newly constructed {@link KryoNetConnectionParameters} based on
   *         the {@link Configuration} specified
   * @throws F2SConfigurationException
   *           If the configuration specified is invalid
   */
  public static KryoNetConnectionParameters from(Configuration config) throws F2SConfigurationException {
    KryoNetConnectionParameters result = new KryoNetConnectionParameters();
    try {
      String hostname = config.getString(ADDRESS);
      if (hostname == null) {
        try {
          hostname = InetAddress.getLocalHost().getHostAddress();
        } catch (final UnknownHostException e) {
          hostname = ADDRESS_DEFAULT;
        }
      }
      assert hostname != null;
      result.setAddress(hostname);
    } catch (Exception e) {
      throw F2SConfigurationException.with(config, ADDRESS, e);
    }
    try {
      result.setServerPort(config.getInt(PORT, PORT_DEFAULT));
    } catch (Exception e) {
      throw F2SConfigurationException.with(config, PORT, e);
    }
    try {
      result.setObjectBufferSize(config.getInt(OBJECT_BUFFER_SZ, OBJECT_BUFFER_SIZE_DEFAULT));
    } catch (Exception e) {
      throw F2SConfigurationException.with(config, OBJECT_BUFFER_SZ, e);
    }
    try {
      result.setWriteBufferSize(config.getInt(WRITE_BUFFER_SZ, WRITE_BUFFER_SIZE_DEFAULT));
    } catch (Exception e) {
      throw F2SConfigurationException.with(config, WRITE_BUFFER_SZ, e);
    }
    return result;
  }

  /**
   * Constructor that initializes the connection parameters using the defaults
   * settings
   */
  public KryoNetConnectionParameters() {
    address = ADDRESS_DEFAULT;
    port = PORT_DEFAULT;
    objectBufferSize = OBJECT_BUFFER_SIZE_DEFAULT;
    writeBufferSize = WRITE_BUFFER_SIZE_DEFAULT;
  }

  /**
   * Copy constructor
   * 
   * @param copy
   *          Another configuration object for KryoNet connection parameters
   */
  public KryoNetConnectionParameters(final KryoNetConnectionParameters copy) {
    address = copy.address;
    port = copy.port;
    objectBufferSize = copy.objectBufferSize;
    writeBufferSize = copy.writeBufferSize;
  }

  /**
   * @see com.comcast.viper.flume2storm.connection.parameters.ConnectionParameters#getId()
   */
  @Override
  public String getId() {
    return new StringBuilder().append(address).append(":").append(port).toString();
  }

  /**
   * @return Address (or host name) to use for the KryoNet event sender.
   *         Defaults to {@value #ADDRESS_DEFAULT}
   */
  public String getAddress() {
    return address;
  }

  /**
   * @param address
   *          See {@link #getAddress()}
   * @return This object
   */
  public KryoNetConnectionParameters setAddress(final String address) {
    Preconditions.checkArgument(address.length() > 0, "KryoNet server address is missing");
    this.address = address;
    return this;
  }

  /**
   * @return Port to use for the KryoNet event sender. Defaults to
   *         {@value #PORT_DEFAULT}
   */
  public int getPort() {
    return port;
  }

  /**
   * @param port
   *          See {@link #getPort()}
   * @return This object
   */
  public KryoNetConnectionParameters setServerPort(final int port) {
    Preconditions.checkArgument(port > 0 && port <= 65535, "KryoNet server port is invalid");
    this.port = port;
    return this;
  }

  /**
   * @return The connection string (for identification purpose)
   */
  public String getConnectionStr() {
    return address + ":" + port;
  }

  /**
   * @return KryoNet's object buffer size in bytes, which MUST be large enough
   *         to serialize any object, plus some margin for KryoNet's internal
   *         stuff
   */
  public int getObjectBufferSize() {
    return objectBufferSize;
  }

  /**
   * @param objBufSz
   *          See {@link #getObjectBufferSize()}
   * @return This object
   */
  public KryoNetConnectionParameters setObjectBufferSize(final int objBufSz) {
    Preconditions.checkArgument(objBufSz > 0, "KryoNet object size needs to be strictly positive");
    objectBufferSize = objBufSz;
    return this;
  }

  /**
   * @return KryoNet's write buffer size in bytes, size it well...
   */
  public int getWriteBufferSize() {
    return writeBufferSize;
  }

  /**
   * @param writeBufferSz
   *          See {@link #getWriteBufferSize()}
   * @return This object
   */
  public KryoNetConnectionParameters setWriteBufferSize(final int writeBufferSz) {
    Preconditions.checkArgument(writeBufferSz > 0, "KryoNet write buffer size needs to be strictly positive");
    writeBufferSize = writeBufferSz;
    return this;
  }

  /**
   * @see java.lang.Object#hashCode()
   */
  @Override
  public int hashCode() {
    return new HashCodeBuilder().append(address).append(port).append(objectBufferSize).append(writeBufferSize)
        .hashCode();
  }

  /**
   * @see java.lang.Object#equals(java.lang.Object)
   */
  @Override
  public boolean equals(Object obj) {
    if (this == obj)
      return true;
    if (obj == null)
      return false;
    if (getClass() != obj.getClass())
      return false;
    KryoNetConnectionParameters other = (KryoNetConnectionParameters) obj;
    return new EqualsBuilder().append(this.address, other.address).append(this.port, other.port)
        .append(this.objectBufferSize, other.objectBufferSize).append(this.writeBufferSize, other.writeBufferSize)
        .isEquals();
  }

  /**
   * @see java.lang.Object#toString()
   */
  @Override
  public String toString() {
    return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE).append("address", address).append("port", port)
        .append("objectBufferSize", objectBufferSize).append("writeBufferSize", writeBufferSize).toString();
  }
} public class KryoNetConnectionParametersFactory implements ConnectionParametersFactory<KryoNetConnectionParameters> {
  /** Configuration attribute base name */
  public static final String CONFIG_BASE_NAME = "kryonet.connection.parameters";

  /**
   * @see com.comcast.viper.flume2storm.connection.parameters.ConnectionParametersFactory#create(org.apache.commons.configuration.Configuration)
   */
  @Override
  public KryoNetConnectionParameters create(Configuration config) throws F2SConfigurationException {
    return KryoNetConnectionParameters.from(config.subset(CONFIG_BASE_NAME));
  }
} public class SimpleConnectionParameters implements ConnectionParameters {
  /** Configuration attribute name for {@link #getHostname()} */
  public static final String HOSTNAME = "hostname";
  /** Default value for {@value #HOSTNAME} */
  public static final String HOSTNAME_DEFAULT = "localhost";
  /** Configuration attribute name for {@link #getPort()} */
  public static final String PORT = "port";
  /** Default value for {@value #PORT} */
  public static final int PORT_DEFAULT = 7000;

  protected String address;
  protected int port;

  /**
   * @param configuration
   *          The configuration to use
   * @return The newly built {@link SimpleConnectionParameters} based on the
   *         configuration specified
   * @throws F2SConfigurationException
   *           If the configuration is invalid
   */
  public static SimpleConnectionParameters from(Configuration configuration) throws F2SConfigurationException {
    SimpleConnectionParameters result = new SimpleConnectionParameters();
    try {
      result.setHostname(configuration.getString(HOSTNAME, HOSTNAME_DEFAULT));
    } catch (Exception e) {
      throw F2SConfigurationException.with(HOSTNAME, configuration.getProperty(HOSTNAME), e);
    }
    try {
      result.setPort(configuration.getInt(PORT, PORT_DEFAULT));
    } catch (Exception e) {
      throw F2SConfigurationException.with(PORT, configuration.getProperty(PORT), e);
    }
    return result;
  }

  /**
   * Constructor that initializes the connection parameters using the defaults
   * settings
   */
  public SimpleConnectionParameters() {
    address = HOSTNAME_DEFAULT;
    port = PORT_DEFAULT;
  }

  /**
   * Constructor that initializes the connection parameters using the specified
   * settings
   * 
   * @param serverAddress
   *          See {@link #getHostname()}
   * @param serverPort
   *          See {@link #getPort()}
   */
  public SimpleConnectionParameters(String serverAddress, int serverPort) {
    this.address = serverAddress;
    this.port = serverPort;
  }

  /**
   * Copy constructor
   * 
   * @param copy
   *          Another configuration object for KryoNet connection parameters
   */
  public SimpleConnectionParameters(final SimpleConnectionParameters copy) {
    address = copy.address;
    port = copy.port;
  }

  /**
   * @see com.comcast.viper.flume2storm.connection.parameters.ConnectionParameters#getId()
   */
  @Override
  public String getId() {
    return new StringBuilder().append(address).append(":").append(port).toString();
  }

  /**
   * @return Address (or host name) to use for the KryoNet event sender.
   *         Defaults to {@value #HOSTNAME_DEFAULT}
   */
  public String getHostname() {
    return address;
  }

  /**
   * @param address
   *          See {@link #getHostname()}
   * @return This configuration object
   */
  public SimpleConnectionParameters setHostname(final String address) {
    this.address = address;
    return this;
  }

  /**
   * @return Port to use for the KryoNet event sender. Defaults to
   *         {@value #PORT_DEFAULT}
   */
  public int getPort() {
    return port;
  }

  /**
   * @param port
   *          See {@link #getPort()}
   * @return This configuration object
   */
  public SimpleConnectionParameters setPort(final int port) {
    if (port < 0)
      throw new IllegalArgumentException("...");
    this.port = port;
    return this;
  }

  /**
   * @return The connection string (for identification purpose)
   */
  public String getConnectionStr() {
    return address + ":" + port;
  }

  /**
   * @see java.lang.Object#hashCode()
   */
  @Override
  public int hashCode() {
    return new HashCodeBuilder().append(address).append(port).hashCode();
  }

  /**
   * @see java.lang.Object#equals(java.lang.Object)
   */
  @Override
  public boolean equals(Object obj) {
    if (this == obj)
      return true;
    if (obj == null)
      return false;
    if (getClass() != obj.getClass())
      return false;
    SimpleConnectionParameters other = (SimpleConnectionParameters) obj;
    return new EqualsBuilder().append(this.address, other.address).append(this.port, other.port).isEquals();
  }

  /**
   * @see java.lang.Object#toString()
   */
  @Override
  public String toString() {
    return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE).append("address", address).append("port", port)
        .build();
  }
} public class SimpleConnectionParametersFactory implements ConnectionParametersFactory<SimpleConnectionParameters> {
  /** Configuration attribute base name */
  public static final String CONFIG_BASE_NAME = "connection";

  /**
   * @see com.comcast.viper.flume2storm.connection.parameters.ConnectionParametersFactory#create(org.apache.commons.configuration.Configuration)
   */
  @Override
  public SimpleConnectionParameters create(Configuration config) throws F2SConfigurationException {
    return SimpleConnectionParameters.from(config.subset(CONFIG_BASE_NAME));
  }
} public class KryoNetEventReceptorFactory implements EventReceptorFactory<KryoNetConnectionParameters> {
  /**
   * @see com.comcast.viper.flume2storm.connection.receptor.EventReceptorFactory#create(com.comcast.viper.flume2storm.connection.parameters.ConnectionParameters,
   *      org.apache.commons.configuration.Configuration)
   */
  @Override
  public EventReceptor<KryoNetConnectionParameters> create(KryoNetConnectionParameters connectionParams,
      Configuration config) throws F2SConfigurationException {
    return new KryoNetEventReceptor(connectionParams, KryoNetParameters.from(config
        .subset(KryoNetParameters.CONFIG_BASE_NAME)));
  }
} public class SimpleEventReceptorFactory implements EventReceptorFactory<SimpleConnectionParameters> {
  /**
   * @see com.comcast.viper.flume2storm.connection.receptor.EventReceptorFactory#create(com.comcast.viper.flume2storm.connection.parameters.ConnectionParameters,
   *      org.apache.commons.configuration.Configuration)
   */
  @Override
  public EventReceptor<SimpleConnectionParameters> create(SimpleConnectionParameters connectionParams,
      Configuration config) throws F2SConfigurationException {
    return new SimpleEventReceptor(connectionParams);
  }
} public interface EventSender<CP extends ConnectionParameters> extends ServiceProvider<CP> {
  /**
   * @return True if the {@link EventSender} started successfully, false
   *         otherwise
   */
  boolean start();

  /**
   * @return True if the {@link EventSender} stopped successfully, false
   *         otherwise
   */
  boolean stop();

  /**
   * @return Statistics related to this {@link EventSender}
   */
  EventSenderStatsMBean getStats();

  /**
   * @param events
   *          The events to send
   * @return The number of events successfully sent
   */
  int send(List<F2SEvent> events);

  // TODO add listener for receptor connection/disconnection event
} public class KryoNetEventSenderFactory implements EventSenderFactory<KryoNetConnectionParameters> {
  /**
   * @see com.comcast.viper.flume2storm.connection.sender.EventSenderFactory#create(com.comcast.viper.flume2storm.connection.parameters.ConnectionParameters,
   *      org.apache.commons.configuration.Configuration)
   */
  @Override
  public EventSender<KryoNetConnectionParameters> create(KryoNetConnectionParameters connectionParams,
      Configuration config) throws F2SConfigurationException {
    return new KryoNetEventSender(connectionParams, KryoNetParameters.from(config
        .subset(KryoNetParameters.CONFIG_BASE_NAME)));
  }
} public class SimpleEventSenderFactory implements EventSenderFactory<SimpleConnectionParameters> {
  /**
   * @see com.comcast.viper.flume2storm.connection.sender.EventSenderFactory#create(com.comcast.viper.flume2storm.connection.parameters.ConnectionParameters,
   *      org.apache.commons.configuration.Configuration)
   */
  @Override
  public EventSender<SimpleConnectionParameters> create(SimpleConnectionParameters connectionParams,
      Configuration config) throws F2SConfigurationException {
    return new SimpleEventSender(connectionParams);
  }
} public final class F2SEvent implements Serializable {
  private static final long serialVersionUID = -5906404402187281472L;
  private final Map<String, String> headers;
  private final byte[] body;

  F2SEvent() {
    headers = ImmutableMap.of();
    body = new byte[0];
  }

  /**
   * Constructor to build Flume2Storm events
   * 
   * @param headers
   *          See {@link #getHeaders()}. If null, the headers is an empty
   *          immutable map.
   * @param body
   *          See {@link #getBody()}. If null, the payload of the event is an
   *          empty byte array.
   */
  public F2SEvent(Map<String, String> headers, byte[] body) {
    if (headers == null) {
      this.headers = ImmutableMap.of();
    } else {
      this.headers = ImmutableMap.copyOf(headers);
    }
    if (body == null) {
      this.body = ArrayUtils.EMPTY_BYTE_ARRAY;
    } else {
      this.body = ArrayUtils.clone(body);
    }
  }

  /**
   * Copy constructor
   * 
   * @param event
   *          Another Flume2Storm event
   */
  public F2SEvent(F2SEvent event) {
    headers = ImmutableMap.copyOf(event.getHeaders());
    body = event.body;
  }

  /**
   * @return The event body
   */
  public byte[] getBody() {
    return body;
  }

  /**
   * @return The headers (key/value pairs) associated to the event
   */
  public Map<String, String> getHeaders() {
    return headers;
  }

  /**
   * @param key
   *          A header name (i.e. key)
   * @return The value of the header, or null if it does not exist
   */
  public String getHeader(String key) {
    return headers.get(key);
  }

  /**
   * @see java.lang.Object#hashCode()
   */
  @Override
  public int hashCode() {
    return new HashCodeBuilder().append(headers).append(body).hashCode();
  }

  /**
   * @see java.lang.Object#equals(java.lang.Object)
   */
  @Override
  public boolean equals(Object obj) {
    if (this == obj)
      return true;
    if (obj == null)
      return false;
    if (getClass() != obj.getClass())
      return false;
    F2SEvent other = (F2SEvent) obj;
    return new EqualsBuilder().append(this.headers, other.headers)
        .append(this.body, other.body).isEquals();
  }

  /**
   * @see java.lang.Object#toString()
   */
  @Override
  public String toString() {
    ToStringBuilder toStringBuilder = new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE).append("headers",
        headers);
    String bodyStr = StringUtils.toEncodedString(body, F2SEventFactory.DEFAULT_CHARACTER_SET);
    if (StringUtils.isAsciiPrintable(bodyStr)) {
      toStringBuilder.append("body", bodyStr);
    } else {
      toStringBuilder.append(Hex.encodeHexString(body));
    }
    return toStringBuilder.toString();
  }
} public class F2SEventFactory {
  static final String DEFAULT_CHARACTER_SET_NAME = "UTF-8";
  static final Charset DEFAULT_CHARACTER_SET = Charset.forName(DEFAULT_CHARACTER_SET_NAME);
  private static final F2SEventFactory INSTANCE = new F2SEventFactory();
  private final Random random;

  /**
   * @return The instance of the Flume2Storm event factory
   */
  public static F2SEventFactory getInstance() {
    return INSTANCE;
  }

  private F2SEventFactory() {
    random = new Random();
  }

  /**
   * @param body
   *          The payload of the event
   * @return The newly created Flume2Storm event
   */
  public F2SEvent create(byte[] body) {
    return new F2SEvent(null, body);
  }

  /**
   * @param body
   *          The payload of the event
   * @return The newly created Flume2Storm event
   */
  public F2SEvent create(String body) {
    return new F2SEvent(null, body.getBytes(DEFAULT_CHARACTER_SET));
  }

  /**
   * @param body
   *          The payload of the event
   * @param headers
   *          The headers associated with the event
   * @return The newly created Flume2Storm event
   */
  public F2SEvent create(byte[] body, Map<String, String> headers) {
    return new F2SEvent(headers, body);
  }

  /**
   * @return An event with a payload containing a random alpha-numeric string
   *         (and no headers)
   */
  public F2SEvent createRandom() {
    return create(RandomStringUtils.randomAlphanumeric(random.nextInt(128) + 1));
  }

  /**
   * @return An event with a payload containing a random alpha-numeric string
   *         (and no headers)
   */
  public F2SEvent createRandomWithHeaders() {
    F2SEventBuilder builder = new F2SEventBuilder();
    for (int i = 0; i < random.nextInt(5); i++) {
      builder.header("H" + i, RandomStringUtils.randomAlphabetic(random.nextInt(64) + 1));
    }
    builder.body(RandomStringUtils.randomAlphanumeric(random.nextInt(128) + 1).getBytes());
    return builder.get();
  }

  /**
   * @param nbEvents
   *          The number of events to create
   * @return A {@link SortedSet} of randomly generated {@link F2SEvent}
   */
  public SortedSet<F2SEvent> generateRandomEvents(int nbEvents) {
    final SortedSet<F2SEvent> result = new TreeSet<F2SEvent>(new F2SEventComparator());
    while (result.size() != nbEvents) {
      result.add(F2SEventFactory.getInstance().createRandomWithHeaders());
    }
    return result;
  }
} public class DynamicLocationServiceConfiguration extends ZkClientConfiguration {
  private static final long serialVersionUID = 4096690012104970344L;
  /** Configuration attribute name for {@link #getBasePath()} */
  public static final String BASE_PATH = "base.path";
  /** Default value for {@value #BASE_PATH} */
  public static final String BASE_PATH_DEFAULT = "/services";
  /** Configuration attribute name for {@link #getServiceName()} */
  public static final String SERVICE_NAME = "service.name";
  /** Default value for {@value #SERVICE_NAME} */
  public static final String SERVICE_NAME_DEFAULT = "flume2storm";

  protected String basePath;
  protected String serviceName;

  /**
   * @param config
   *          The configuration to use
   * @return The newly built {@link DynamicLocationServiceConfiguration} based
   *         on the configuration specified
   * @throws F2SConfigurationException
   *           If the configuration is invalid
   */
  public static DynamicLocationServiceConfiguration from(Configuration config) throws F2SConfigurationException {
    DynamicLocationServiceConfiguration result = new DynamicLocationServiceConfiguration();
    result.setConnectionStr(config.getString(CONNECTION_STRING, CONNECTION_STRING_DEFAULT));
    result.setSessionTimeout(config.getInt(SESSION_TIMEOUT, SESSION_TIMEOUT_DEFAULT));
    result.setConnectionTimeout(config.getInt(CONNECTION_TIMEOUT, CONNECTION_TIMEOUT_DEFAULT));
    result.setReconnectionDelay(config.getInt(RECONNECTION_DELAY, RECONNECTION_DELAY_DEFAULT));
    result.setTerminationTimeout(config.getInt(TERMINATION_TIMEOUT, TERMINATION_TIMEOUT_DEFAULT));
    result.setBasePath(config.getString(BASE_PATH, BASE_PATH_DEFAULT));
    result.setServiceName(config.getString(SERVICE_NAME, SERVICE_NAME_DEFAULT));
    return result;
  }

  /**
   * Empty constructor - initializes with default values when available
   */
  public DynamicLocationServiceConfiguration() {
    super();
    connectionStr = CONNECTION_STRING_DEFAULT;
    sessionTimeout = SESSION_TIMEOUT_DEFAULT;
    connectionTimeout = CONNECTION_TIMEOUT_DEFAULT;
    reconnectionDelay = RECONNECTION_DELAY_DEFAULT;
    terminationTimeout = TERMINATION_TIMEOUT_DEFAULT;
    basePath = BASE_PATH_DEFAULT;
    serviceName = SERVICE_NAME_DEFAULT;
  }

  /**
   * Copy constructor
   * 
   * @param other
   *          the configuration to copy
   */
  public DynamicLocationServiceConfiguration(final DynamicLocationServiceConfiguration other) {
    super(other);
    basePath = other.basePath;
    serviceName = other.serviceName;
  }

  /**
   * @return The service name (which is used to group all the
   *         {@link ServiceProvider} instances in ZooKeeper)
   */
  public String getServiceName() {
    return serviceName;
  }

  /**
   * @param serviceName
   *          See {@link #getServiceName()}
   */
  public void setServiceName(final String serviceName) {
    // TODO Use a regular expression to validate there are no invalid char
    Preconditions.checkArgument(serviceName.length() > 0, "Flume2Storm service name needs to be set");
    this.serviceName = serviceName;
  }

  /**
   * @return ZooKeeper's root path to store this service
   */
  public String getBasePath() {
    return basePath;
  }

  /**
   * @param basePath
   *          See {@link #getBasePath()}
   */
  public void setBasePath(final String basePath) {
    // TODO Use a regular expression to validate there are no invalid char
    this.basePath = basePath;
  }

  /**
   * @see com.comcast.viper.flume2storm.zookeeper.ZkClientConfiguration#hashCode()
   */
  @Override
  public int hashCode() {
    return new HashCodeBuilder().appendSuper(super.hashCode()).append(basePath).append(serviceName).hashCode();
  }

  /**
   * @see com.comcast.viper.flume2storm.zookeeper.ZkClientConfiguration#equals(java.lang.Object)
   */
  @Override
  public boolean equals(Object obj) {
    if (this == obj)
      return true;
    if (!super.equals(obj))
      return false;
    if (getClass() != obj.getClass())
      return false;
    DynamicLocationServiceConfiguration other = (DynamicLocationServiceConfiguration) obj;
    return new EqualsBuilder().appendSuper(super.equals(obj)).append(this.basePath, other.basePath)
        .append(this.serviceName, other.serviceName).isEquals();
  }

  /**
   * @see java.lang.Object#toString()
   */
  @Override
  public String toString() {
    return new ToStringBuilder(this).append("basePath", basePath).append("serviceName", serviceName)
        .append("connectionStr", connectionStr).append("sessionTimeout", sessionTimeout)
        .append("connectionTimeout", connectionTimeout).append("reconnectionDelay", reconnectionDelay)
        .append("terminationTimeout", terminationTimeout).toString();
  }
} public class DynamicLocationServiceFactory<SP extends ServiceProvider<?>> implements LocationServiceFactory<SP> {
  /** Configuration attribute base name */
  public static final String CONFIG_BASE_NAME = "dynamic.location.service";

  /**
   * @see com.comcast.viper.flume2storm.location.LocationServiceFactory#create(org.apache.commons.configuration.Configuration,
   *      com.comcast.viper.flume2storm.location.ServiceProviderSerialization)
   */
  @Override
  public LocationService<SP> create(Configuration config, ServiceProviderSerialization<SP> serviceProviderSerialization)
      throws F2SConfigurationException {
    return new DynamicLocationService<SP>(DynamicLocationServiceConfiguration.from(config.subset(CONFIG_BASE_NAME)),
        serviceProviderSerialization);
  }
} public class KryoNetServiceProvider implements ServiceProvider<KryoNetConnectionParameters> {
  private static final long serialVersionUID = -6694093091877189894L;
  protected final KryoNetConnectionParameters connectionParameters;

  protected KryoNetServiceProvider() {
    connectionParameters = new KryoNetConnectionParameters();
  }

  /**
   * Constructor
   * 
   * @param connectionParameters
   *          See {@link #getConnectionParameters()}t
   */
  public KryoNetServiceProvider(KryoNetConnectionParameters connectionParameters) {
    this.connectionParameters = connectionParameters;
  }

  /**
   * @see com.comcast.viper.flume2storm.location.ServiceProvider#getConnectionParameters()
   */
  public KryoNetConnectionParameters getConnectionParameters() {
    return connectionParameters;
  }

  /**
   * @see java.lang.Comparable#compareTo(java.lang.Object)
   */
  @Override
  public int compareTo(ServiceProvider<KryoNetConnectionParameters> o) {
    int res = connectionParameters.getAddress().compareTo(o.getConnectionParameters().getAddress());
    if (res != 0)
      return res;
    int thisPort = connectionParameters.getPort();
    int otherPort = o.getConnectionParameters().getPort();
    return (thisPort < otherPort ? -1 : (thisPort == otherPort ? 0 : 1));
  }

  /**
   * @see java.lang.Object#hashCode()
   */
  @Override
  public int hashCode() {
    return new HashCodeBuilder().append(connectionParameters).hashCode();
  }

  /**
   * @see java.lang.Object#equals(java.lang.Object)
   */
  @Override
  public boolean equals(Object obj) {
    if (this == obj)
      return true;
    if (obj == null)
      return false;
    if (getClass() != obj.getClass())
      return false;
    KryoNetServiceProvider other = (KryoNetServiceProvider) obj;
    return new EqualsBuilder().append(this.connectionParameters, other.connectionParameters).isEquals();
  }

  /**
   * @see java.lang.Object#toString()
   */
  @Override
  public String toString() {
    return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE).append(connectionParameters).toString();
  }
} public class KryoNetServiceProviderSerialization implements ServiceProviderSerialization<KryoNetServiceProvider> {
  private static final long serialVersionUID = -3054825185354578392L;
  protected static String SEPARATOR = ":";

  /**
   * @see com.comcast.viper.flume2storm.location.ServiceProviderSerialization#serialize(com.comcast.viper.flume2storm.location.ServiceProvider)
   */
  @Override
  public byte[] serialize(KryoNetServiceProvider serviceProvider) {
    KryoNetConnectionParameters cp = serviceProvider.getConnectionParameters();
    StringBuilder sb = new StringBuilder();
    sb.append(cp.getAddress());
    sb.append(SEPARATOR);
    sb.append(cp.getPort());
    sb.append(SEPARATOR);
    sb.append(cp.getObjectBufferSize());
    sb.append(SEPARATOR);
    sb.append(cp.getWriteBufferSize());
    return sb.toString().getBytes();
  }

  /**
   * @see com.comcast.viper.flume2storm.location.ServiceProviderSerialization#deserialize(byte[])
   */
  @Override
  public KryoNetServiceProvider deserialize(byte[] bytes) {
    Preconditions.checkNotNull(bytes, "Failed to deserialize KryoNet Service Provider: null byte array");
    String[] elements = new String(bytes).split(SEPARATOR);
    Preconditions.checkArgument(elements.length == 4,
        "Failed to deserialize KryoNet Service Provider: invalid number of elements");
    KryoNetConnectionParameters cp = new KryoNetConnectionParameters();
    cp.setAddress(elements[0]);
    cp.setServerPort(Integer.parseInt(elements[1]));
    cp.setObjectBufferSize(Integer.parseInt(elements[2]));
    cp.setWriteBufferSize(Integer.parseInt(elements[3]));
    return new KryoNetServiceProvider(cp);
  }
} public interface ServiceProvider<CP extends ConnectionParameters> extends Serializable, Comparable<ServiceProvider<CP>> {
  /**
   * @return The connection parameters for this service provider
   */
  CP getConnectionParameters();
} public interface ServiceProviderConfigurationLoader<SP extends ServiceProvider<?>> {
  /**
   * @param config
   *          The configuration to use to load a Service Provider. It contains
   *          only the subset of configuration variables to load one service
   *          provider
   * @return The newly created Service Provider
   * @throws F2SConfigurationException
   *           If the configuration is invalid
   */
  SP load(Configuration config) throws F2SConfigurationException;
} public class SimpleLocationServiceFactory implements LocationServiceFactory<SimpleServiceProvider> {
  /**
   * @see com.comcast.viper.flume2storm.location.LocationServiceFactory#create(org.apache.commons.configuration.Configuration,
   *      com.comcast.viper.flume2storm.location.ServiceProviderSerialization)
   */
  @Override
  public SimpleLocationService create(Configuration config,
      ServiceProviderSerialization<SimpleServiceProvider> serviceProviderSerialization)
      throws F2SConfigurationException {
    return SimpleLocationService.getInstance();
  }
} public class SimpleServiceProviderSerialization implements ServiceProviderSerialization<SimpleServiceProvider> {
  private static final long serialVersionUID = -1002890263399098547L;
  private static final String SEPARATOR = ":";

  /**
   * @see com.comcast.viper.flume2storm.location.ServiceProviderSerialization#serialize(com.comcast.viper.flume2storm.location.ServiceProvider)
   */
  public byte[] serialize(SimpleServiceProvider serviceProvider) {
    Preconditions.checkNotNull(serviceProvider, "Failed to serialize null SimpleServiceProvider");
    return new StringBuilder().append(serviceProvider.getHostname()).append(SEPARATOR)
        .append(serviceProvider.getPort()).toString().getBytes();
  }

  /**
   * @see com.comcast.viper.flume2storm.location.ServiceProviderSerialization#deserialize(byte[])
   */
  public SimpleServiceProvider deserialize(byte[] bytes) {
    Preconditions.checkNotNull(bytes, "Failed to deserialize null SimpleServiceProvider");
    String string = new String(bytes);
    String[] elements = string.split(SEPARATOR);
    if (elements.length != 2) {
      throw new IllegalArgumentException("Failed to deserialize SimpleServiceProvider: " + string);
    }
    return new SimpleServiceProvider(elements[0], Integer.parseInt(elements[1]));
  }
} public class StaticLocationServiceConfiguration {
  /** Configuration attribute name to get the list of service providers */
  public static final String SERVICE_PROVIDER_LIST = "service.providers";
  /** The {@link ServiceProvider} list separator */
  public static final String SERVICE_PROVIDER_LIST_SEPARATOR = " ";
  /** Configuration attribute name for {@link #getServiceProviderBase()} */
  public static final String SERVICE_PROVIDER_BASE = "service.providers.base";
  /** Default value for {@value #SERVICE_PROVIDER_BASE} */
  public static final String SERVICE_PROVIDER_BASE_DEFAULT = "service.providers";
  /**
   * Configuration attribute name for the
   * {@link ServiceProviderConfigurationLoader} class name
   */
  public static final String CONFIGURATION_LOADER_CLASS = "configuration.loader.class";

  protected String serviceProviderBase;
  private String configurationLoaderClassName;

  /**
   * @param config
   *          The configuration to use
   * @return The newly built {@link StaticLocationServiceConfiguration} based on
   *         the configuration specified
   * @throws F2SConfigurationException
   *           If the configuration is invalid
   */
  public static StaticLocationServiceConfiguration from(Configuration config) throws F2SConfigurationException {
    StaticLocationServiceConfiguration result = new StaticLocationServiceConfiguration();
    result.setServiceProviderBase(config.getString(SERVICE_PROVIDER_BASE, SERVICE_PROVIDER_LIST));
    String className = config.getString(CONFIGURATION_LOADER_CLASS);
    if (className == null) {
      throw F2SConfigurationException.missing(CONFIGURATION_LOADER_CLASS);
    }
    try {
      result.setConfigurationLoaderClassName(className);
    } catch (Exception e) {
      throw F2SConfigurationException.with(config, CONFIGURATION_LOADER_CLASS, e);
    }
    return result;
  }

  /**
   * Empty constructor - initializes with default values when available
   */
  public StaticLocationServiceConfiguration() {
    serviceProviderBase = SERVICE_PROVIDER_LIST;
  }

  /**
   * Copy constructor
   * 
   * @param other
   *          the configuration to copy
   */
  public StaticLocationServiceConfiguration(final StaticLocationServiceConfiguration other) {
    serviceProviderBase = other.serviceProviderBase;
    configurationLoaderClassName = other.configurationLoaderClassName;
  }

  /**
   * @return The base name for the {@link ServiceProvider} attributes
   */
  public String getServiceProviderBase() {
    return serviceProviderBase;
  }

  /**
   * @param serviceProviderBase
   *          See {@link #getServiceProviderBase()}
   */
  public void setServiceProviderBase(final String serviceProviderBase) {
    Preconditions.checkNotNull(serviceProviderBase);
    this.serviceProviderBase = serviceProviderBase;
  }

  /**
   * @return The class name of the {@link ServiceProviderConfigurationLoader}
   */
  public String getConfigurationLoaderClassName() {
    return configurationLoaderClassName;
  }

  /**
   * @param configurationLoaderClassName
   *          See {@link #getConfigurationLoaderClassName()}
   * @throws ClassNotFoundException
   *           If the class specified is not found in the classpath
   */
  public void setConfigurationLoaderClassName(String configurationLoaderClassName) throws ClassNotFoundException {
    Class<?> configurationLoaderClass = Class.forName(configurationLoaderClassName);
    Preconditions.checkArgument(ServiceProviderConfigurationLoader.class.isAssignableFrom(configurationLoaderClass),
        "The class must implement " + ServiceProviderConfigurationLoader.class.getCanonicalName());
    this.configurationLoaderClassName = configurationLoaderClassName;
  }

  /**
   * @see java.lang.Object#hashCode()
   */
  @Override
  public int hashCode() {
    return new HashCodeBuilder().append(serviceProviderBase).append(configurationLoaderClassName).hashCode();
  }

  /**
   * @see java.lang.Object#equals(java.lang.Object)
   */
  @Override
  public boolean equals(Object obj) {
    if (this == obj)
      return true;
    if (!super.equals(obj))
      return false;
    if (getClass() != obj.getClass())
      return false;
    StaticLocationServiceConfiguration other = (StaticLocationServiceConfiguration) obj;
    return new EqualsBuilder().append(this.serviceProviderBase, other.configurationLoaderClassName).isEquals();
  }

  /**
   * @see java.lang.Object#toString()
   */
  @Override
  public String toString() {
    return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE)
        .append("serviceProviderBase", serviceProviderBase)
        .append("configurationLoaderClassName", configurationLoaderClassName).toString();
  }
} public class StaticLocationServiceFactory<SP extends ServiceProvider<?>> implements LocationServiceFactory<SP> {
  /** Configuration attribute base name */
  public static final String CONFIG_BASE_NAME = "static.location.service";

  /**
   * @see com.comcast.viper.flume2storm.location.LocationServiceFactory#create(org.apache.commons.configuration.Configuration,
   *      com.comcast.viper.flume2storm.location.ServiceProviderSerialization)
   */
  @Override
  public LocationService<SP> create(Configuration config, ServiceProviderSerialization<SP> serviceProviderSerialization)
      throws F2SConfigurationException {
    return new StaticLocationService<SP>(config.subset(CONFIG_BASE_NAME), serviceProviderSerialization);
  }
} public class StormSink<CP extends ConnectionParameters, SP extends ServiceProvider<CP>> extends AbstractSink implements
    Configurable {
  private static final Logger LOG = LoggerFactory.getLogger(StormSink.class);
  protected SinkCounter sinkCounter;
  protected StormSinkConfiguration configuration;
  protected LocationServiceFactory<SP> locationServiceFactory;
  protected LocationService<SP> locationService;
  protected ServiceProviderSerialization<SP> serviceProviderSerialization;
  protected EventSenderFactory<CP> eventSenderFactory;
  protected ConnectionParametersFactory<CP> connectionParametersFactory;
  protected EventSender<CP> eventSender;
  protected EventConverter eventConverter;

  /**
   * @see org.apache.flume.conf.Configurable#configure(org.apache.flume.Context)
   */
  @Override
  public void configure(final Context context) {
    Preconditions.checkNotNull(context);
    if (sinkCounter == null) {
      sinkCounter = new SinkCounter(getName());
    }
    try {
      Configuration mapConfiguration = new MapConfiguration(context.getParameters());
      LOG.debug("Storm-sink configuration:\n{}", ConfigurationUtils.toString(mapConfiguration));
      configuration = StormSinkConfiguration.from(mapConfiguration);
      Class<? extends ConnectionParametersFactory<CP>> connectionParametersFactoryClass = (Class<? extends ConnectionParametersFactory<CP>>) Class
          .forName(configuration.getConnectionParametersFactoryClassName());
      this.connectionParametersFactory = connectionParametersFactoryClass.newInstance();
      Class<? extends EventSenderFactory<CP>> eventSenderFactoryClass = (Class<? extends EventSenderFactory<CP>>) Class
          .forName(configuration.getEventSenderFactoryClassName());
      this.eventSenderFactory = eventSenderFactoryClass.newInstance();
      Class<? extends LocationServiceFactory<SP>> locationServiceFactoryClass = (Class<? extends LocationServiceFactory<SP>>) Class
          .forName(configuration.getLocationServiceFactoryClassName());
      this.locationServiceFactory = locationServiceFactoryClass.newInstance();
      Class<? extends ServiceProviderSerialization<SP>> serviceProviderSerializationClass = (Class<? extends ServiceProviderSerialization<SP>>) Class
          .forName(configuration.getServiceProviderSerializationClassName());
      this.serviceProviderSerialization = serviceProviderSerializationClass.newInstance();
    } catch (Exception e) {
      LOG.error("Failed to configure Storm sink: " + e.getMessage(), e);
    }
  }

  /**
   * @see org.apache.flume.sink.AbstractSink#start()
   */
  @Override
  public synchronized void start() {
    try {
      LOG.debug("Opening...");
      sinkCounter.start();
      CP connectionParameters = connectionParametersFactory.create(configuration.get());
      eventSender = eventSenderFactory.create(connectionParameters, configuration.get());
      eventSender.start();

      locationService = locationServiceFactory.create(configuration.get(), serviceProviderSerialization);
      locationService.start();
      locationService.register((SP) eventSender);

      eventConverter = new EventConverter();
      LOG.info("Opened");
      super.start();
    } catch (F2SConfigurationException e) {
      LOG.error(e.getLocalizedMessage(), e);
    }
  }

  /**
   * @see org.apache.flume.sink.AbstractSink#stop()
   */
  @Override
  public synchronized void stop() {
    LOG.debug("Closing...");
    locationService.unregister((SP) eventSender);
    locationService.stop();
    eventSender.stop();
    sinkCounter.stop();
    LOG.info("Closed");
    super.stop();
  }

  /**
   * @see org.apache.flume.Sink#process()
   */
  @Override
  public Status process() throws EventDeliveryException {
    if (eventSender.getStats().getNbClients() == 0) {
      // No receptor connected
      return Status.BACKOFF;
    }
    Status result = Status.READY;
    final Channel channel = getChannel();
    final Transaction transaction = channel.getTransaction();
    try {
      transaction.begin();
      List<F2SEvent> batch = new ArrayList<F2SEvent>(configuration.getBatchSize());
      for (int i = 0; i < configuration.getBatchSize(); i++) {
        final Event event = channel.take();
        if (event == null) {
          break;
        }
        batch.add(eventConverter.convert(event));
      }
      if (batch.isEmpty()) {
        sinkCounter.incrementBatchEmptyCount();
        result = Status.BACKOFF;
      } else {
        sinkCounter.addToEventDrainAttemptCount(batch.size());
        if (eventSender.send(batch) == 0) {
          throw new EventDeliveryException("No event sent to receptor");
        }
        sinkCounter.addToEventDrainSuccessCount(batch.size());
        if (batch.size() == configuration.getBatchSize()) {
          sinkCounter.incrementBatchCompleteCount();
        } else {
          sinkCounter.incrementBatchUnderflowCount();
        }
      }
      transaction.commit();
    } catch (final Throwable t) {
      transaction.rollback();
      if (t instanceof Error) {
        throw (Error) t;
      } else if (t instanceof ChannelException) {
        if (!(t.getCause() instanceof InterruptedException)) {
          LOG.error("Storm Sink " + getName() + ": Unable to get event from channel " + channel.getName(), t);
        }
        result = Status.BACKOFF;
      } else {
        throw new EventDeliveryException("Failed to accept events", t);
      }
    } finally {
      transaction.close();
    }
    return result;
  }

  /**
   * @return The stats associated with teh {@link EventSender}
   */
  public EventSenderStatsMBean getEventSernderStats() {
    return eventSender == null ? new EventSenderStats("Unknown") : eventSender.getStats();
  }

  /**
   * @return The sink counter
   */
  public SinkCounter getSinkCounter() {
    return sinkCounter;
  }
} public class StormSinkConfiguration implements Supplier<Configuration> {
  /** Configuration attribute name for {@link #getBatchSize()} */
  public static final String BATCH_SIZE = "batch-size";
  /** Default value for {@value #BATCH_SIZE} */
  public static final int BATCH_SIZE_DEFAULT = 100;
  /**
   * Configuration attribute name for
   * {@link #getLocationServiceFactoryClassName()}
   */
  public static final String LOCATION_SERVICE_FACTORY_CLASS = "location.service.factory";
  /** Default value for {@value #LOCATION_SERVICE_FACTORY_CLASS} */
  public static final String LOCATION_SERVICE_FACTORY_CLASS_DEFAULT = "com.comcast.viper.flume2storm.location.DynamicLocationServiceFactory";
  /**
   * Configuration attribute name for
   * {@link #getServiceProviderSerializationClassName()}
   */
  public static final String SERVICE_PROVIDER_SERIALIZATION_CLASS = "service.provider.serialization";
  /** Default value for {@value #SERVICE_PROVIDER_SERIALIZATION_CLASS} */
  public static final String SERVICE_PROVIDER_SERIALIZATION_CLASS_DEFAULT = "com.comcast.viper.flume2storm.location.KryoNetServiceProviderSerialization";
  /** Configuration attribute name for {@link #getEventSenderFactoryClassName()} */
  public static final String EVENT_SENDER_FACTORY_CLASS = "event.sender.factory";
  /** Default value for {@value #EVENT_SENDER_FACTORY_CLASS} */
  public static final String EVENT_SENDER_FACTORY_CLASS_DEFAULT = "com.comcast.viper.flume2storm.connection.sender.KryoNetEventSenderFactory";
  /**
   * Configuration attribute name for
   * {@link #getConnectionParametersFactoryClassName()}
   */
  public static final String CONNECTION_PARAMETERS_FACTORY_CLASS = "connection.parameters.factory";
  /** Default value for {@value #CONNECTION_PARAMETERS_FACTORY_CLASS} */
  public static final String CONNECTION_PARAMETERS_FACTORY_CLASS_DEFAULT = "com.comcast.viper.flume2storm.connection.parameters.KryoNetConnectionParametersFactory";

  protected int batchSize;
  protected String locationServiceFactoryClassName;
  protected String serviceProviderSerializationClassName;
  protected String eventSenderFactoryClassName;
  protected String connectionParametersFactoryClassName;
  protected Configuration configuration;

  /**
   * @param config
   *          The configuration to use
   * @return The newly built {@link StormSinkConfiguration} based on the
   *         configuration specified
   * @throws F2SConfigurationException
   *           If the configuration is invalid
   */
  public static StormSinkConfiguration from(Configuration config) throws F2SConfigurationException {
    StormSinkConfiguration result = new StormSinkConfiguration();
    try {
      result.setBatchSize(config.getInteger(BATCH_SIZE, BATCH_SIZE_DEFAULT));
    } catch (Exception e) {
      throw F2SConfigurationException.with(config, BATCH_SIZE, e);
    }
    try {
      result.setLocationServiceFactoryClassName(config.getString(LOCATION_SERVICE_FACTORY_CLASS,
          LOCATION_SERVICE_FACTORY_CLASS_DEFAULT));
    } catch (Exception e) {
      throw F2SConfigurationException.with(config, LOCATION_SERVICE_FACTORY_CLASS, e);
    }
    try {
      result.setServiceProviderSerializationClassName(config.getString(SERVICE_PROVIDER_SERIALIZATION_CLASS,
          SERVICE_PROVIDER_SERIALIZATION_CLASS_DEFAULT));
    } catch (Exception e) {
      throw F2SConfigurationException.with(config, SERVICE_PROVIDER_SERIALIZATION_CLASS, e);
    }
    try {
      result.setEventSenderFactoryClassName(config.getString(EVENT_SENDER_FACTORY_CLASS,
          EVENT_SENDER_FACTORY_CLASS_DEFAULT));
    } catch (Exception e) {
      throw F2SConfigurationException.with(config, EVENT_SENDER_FACTORY_CLASS, e);
    }
    try {
      result.setConnectionParametersFactoryClassName(config.getString(CONNECTION_PARAMETERS_FACTORY_CLASS,
          CONNECTION_PARAMETERS_FACTORY_CLASS_DEFAULT));
    } catch (Exception e) {
      throw F2SConfigurationException.with(config, CONNECTION_PARAMETERS_FACTORY_CLASS, e);
    }
    result.configuration = config;
    return result;
  }

  /**
   * Empty constructor - initializes with default values when available
   */
  public StormSinkConfiguration() {
    batchSize = BATCH_SIZE_DEFAULT;
    locationServiceFactoryClassName = LOCATION_SERVICE_FACTORY_CLASS_DEFAULT;
    serviceProviderSerializationClassName = SERVICE_PROVIDER_SERIALIZATION_CLASS_DEFAULT;
    connectionParametersFactoryClassName = CONNECTION_PARAMETERS_FACTORY_CLASS_DEFAULT;
    eventSenderFactoryClassName = EVENT_SENDER_FACTORY_CLASS_DEFAULT;
    configuration = new BaseConfiguration();
  }

  /**
   * Copy constructor
   * 
   * @param other
   *          the configuration to copy
   */
  public StormSinkConfiguration(final StormSinkConfiguration other) {
    batchSize = other.batchSize;
    locationServiceFactoryClassName = other.locationServiceFactoryClassName;
    serviceProviderSerializationClassName = other.serviceProviderSerializationClassName;
    connectionParametersFactoryClassName = other.connectionParametersFactoryClassName;
    eventSenderFactoryClassName = other.eventSenderFactoryClassName;
    configuration = new MapConfiguration(ConfigurationConverter.getMap(other.configuration));
  }

  /**
   * @return The number of events per transaction
   */
  public int getBatchSize() {
    return batchSize;
  }

  /**
   * @param batchSize
   *          See {@link #getBatchSize()}
   */
  public void setBatchSize(int batchSize) {
    Preconditions.checkArgument(batchSize > 0, "Batch-size must be strictly positive");
    this.batchSize = batchSize;
  }

  /**
   * @return The class name of the location service factory
   */
  public String getLocationServiceFactoryClassName() {
    return locationServiceFactoryClassName;
  }

  /**
   * @param locationServiceFactoryClassName
   *          See {@link #getLocationServiceFactoryClassName()}
   * @throws ClassNotFoundException
   *           If the class specified is not found in the classpath
   */
  public void setLocationServiceFactoryClassName(String locationServiceFactoryClassName) throws ClassNotFoundException {
    Class<?> locationServiceFactoryClass = Class.forName(locationServiceFactoryClassName);
    Preconditions.checkArgument(LocationServiceFactory.class.isAssignableFrom(locationServiceFactoryClass),
        "The class must implement " + LocationServiceFactory.class.getCanonicalName());
    this.locationServiceFactoryClassName = locationServiceFactoryClassName;
  }

  /**
   * @return The class name of the service provider serialization
   */
  public String getServiceProviderSerializationClassName() {
    return serviceProviderSerializationClassName;
  }

  /**
   * @param serviceProviderSerializationClassName
   *          See {@link #getServiceProviderSerializationClassName()}
   * @throws ClassNotFoundException
   *           If the class specified is not found in the classpath
   */
  public void setServiceProviderSerializationClassName(String serviceProviderSerializationClassName)
      throws ClassNotFoundException {
    Class<?> serviceProviderSerializationClass = Class.forName(serviceProviderSerializationClassName);
    Preconditions.checkArgument(ServiceProviderSerialization.class.isAssignableFrom(serviceProviderSerializationClass),
        "The class must implement " + ServiceProviderSerialization.class.getCanonicalName());
    this.serviceProviderSerializationClassName = serviceProviderSerializationClassName;
  }

  /**
   * @return The class name of the event sender factory
   */
  public String getEventSenderFactoryClassName() {
    return eventSenderFactoryClassName;
  }

  /**
   * @param eventSenderFactoryClassName
   *          See {@link #getEventSenderFactoryClassName()}
   * @throws ClassNotFoundException
   *           If the class specified is not found in the classpath
   */
  public void setEventSenderFactoryClassName(String eventSenderFactoryClassName) throws ClassNotFoundException {
    Class<?> eventSenderFactoryClass = Class.forName(eventSenderFactoryClassName);
    Preconditions.checkArgument(EventSenderFactory.class.isAssignableFrom(eventSenderFactoryClass),
        "The class must implement " + EventSenderFactory.class.getCanonicalName());
    this.eventSenderFactoryClassName = eventSenderFactoryClassName;
  }

  /**
   * @return The class name of the connection parameters factory
   */
  public String getConnectionParametersFactoryClassName() {
    return connectionParametersFactoryClassName;
  }

  /**
   * @param connectionParametersFactoryClassName
   *          See {@link #getConnectionParametersFactoryClassName()}
   * @throws ClassNotFoundException
   *           If the class specified is not found in the classpath
   */
  public void setConnectionParametersFactoryClassName(String connectionParametersFactoryClassName)
      throws ClassNotFoundException {
    Class<?> connectionParametersFactoryClass = Class.forName(connectionParametersFactoryClassName);
    Preconditions.checkArgument(ConnectionParametersFactory.class.isAssignableFrom(connectionParametersFactoryClass),
        "The class must implement " + ConnectionParametersFactory.class.getCanonicalName());
    this.connectionParametersFactoryClassName = connectionParametersFactoryClassName;
  }

  /**
   * @see com.google.common.base.Supplier#get()
   * @return This object as a {@link Configuration}
   */
  public Configuration get() {
    return configuration;
  }

  /**
   * @param configuration
   *          The {@link FlumeSpout} configuration as a
   *          {@link StormSinkConfiguration}
   */
  public void setConfiguration(Configuration configuration) {
    this.configuration = configuration;
  }
} public class BasicF2SEventEmitter implements F2SEventEmitter {
  private static final long serialVersionUID = 8195071400428813016L;
  /** Field name of emitted tuples */
  public static final String F2S_EVENT = "f2sEvent";
  /** Tuple fields declaration */
  public static final Fields fields = new Fields(F2S_EVENT);

  /**
   * @see com.comcast.viper.flume2storm.spout.F2SEventEmitter#declareOutputFields(backtype.storm.topology.OutputFieldsDeclarer)
   */
  @Override
  public void declareOutputFields(OutputFieldsDeclarer declarer) {
    declarer.declare(fields);
  }

  /**
   * @see com.comcast.viper.flume2storm.spout.F2SEventEmitter#emitEvent(com.comcast.viper.flume2storm.event.F2SEvent,
   *      backtype.storm.spout.SpoutOutputCollector)
   */
  @Override
  public void emitEvent(F2SEvent event, SpoutOutputCollector collector) {
    final List<Object> tuple = new ArrayList<Object>(1);
    tuple.add(event);
    collector.emit(tuple);
  }
} public interface F2SEventEmitter extends Serializable {
  /**
   * Implement this method to specify what tuples the Flume-Spout will emit on
   * which stream.
   * 
   * @param declarer
   *          Storm {@link OutputFieldsDeclarer}
   */
  public void declareOutputFields(OutputFieldsDeclarer declarer);

  /**
   * Method to actually emit the Flume2Storm events down the topology. Tuples
   * must match what has been declared in
   * {@link #declareOutputFields(OutputFieldsDeclarer)}.
   * 
   * @param event
   *          The event being emitted
   * @param collector
   *          The collector to use
   */
  public void emitEvent(F2SEvent event, SpoutOutputCollector collector);

  // TODO Add acknowledgement methods
} public class FlumeSpout<CP extends ConnectionParameters, SP extends ServiceProvider<CP>> extends BaseRichSpout {
  private static final long serialVersionUID = -2858136141243917853L;
  /** Small sleep for storm spout */
  protected static final int SPOUT_SLEEP_TIME = 10;
  protected static final Logger LOG = LoggerFactory.getLogger(FlumeSpout.class);

  protected final FlumeSpoutConfiguration configuration;
  protected final Map<String, EventReceptor<CP>> eventReceptors;
  protected final Set<F2SEventEmitter> emitters;
  protected SpoutOutputCollector collector;
  protected LocationServiceFactory<SP> locationServiceFactory;
  protected ServiceProviderSerialization<SP> serviceProviderSerialization;
  protected LocationService<SP> locationService;
  protected EventReceptorFactory<CP> eventReceptorFactory;
  protected ServiceListener<SP> serviceListener;

  private class MyServiceListener implements ServiceListener<SP> {
    protected MyServiceListener() {
      super();
    }

    /**
     * @see com.comcast.viper.flume2storm.location.ServiceListener#onProviderRemoved(com.comcast.viper.flume2storm.location.ServiceProvider)
     */
    @Override
    public void onProviderRemoved(SP serviceProvider) {
      // The associated EventReceptor is not removed at this point
    }

    /**
     * @see com.comcast.viper.flume2storm.location.ServiceListener#onProviderAdded(com.comcast.viper.flume2storm.location.ServiceProvider)
     */
    @Override
    public void onProviderAdded(SP serviceProvider) {
      try {
        LOG.info("Added provider: {}", serviceProvider);
        EventReceptor<CP> eventReceptor = eventReceptorFactory.create(serviceProvider.getConnectionParameters(),
            configuration.get());
        LOG.debug("Adding event receptor: {}", eventReceptor);
        eventReceptor.start();
        while (!eventReceptor.getStats().isConnected()) {
          LOG.debug("Receptor not started... waiting...");
          try {
            Thread.sleep(500);
          } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
          }
        }
        eventReceptors.put(serviceProvider.getConnectionParameters().getId(), eventReceptor);
      } catch (F2SConfigurationException e) {
        LOG.error("Failed to add service provider: " + serviceProvider, e);
      }
    }
  }

  /**
   * Constructs a new {@link FlumeSpout}
   * 
   * @param emitters
   *          The list of {@link F2SEventEmitter}
   * @param configuration
   *          The configuration for the spout
   * @throws F2SConfigurationException
   *           If the configuration is invalid
   */
  public FlumeSpout(final Set<F2SEventEmitter> emitters, final Configuration configuration)
      throws F2SConfigurationException {
    this.emitters = emitters;
    eventReceptors = new ConcurrentHashMap<String, EventReceptor<CP>>();
    this.configuration = FlumeSpoutConfiguration.from(configuration);
  }

  /**
   * Convenience constructor
   * 
   * @param emitter
   *          One {@link F2SEventEmitter}
   * @param configuration
   *          The configuration for the spout
   * @throws F2SConfigurationException
   *           If the configuration is invalid
   */
  public FlumeSpout(final F2SEventEmitter emitter, final Configuration configuration) throws F2SConfigurationException {
    this(ImmutableSet.of(emitter), configuration);
  }

  /**
   * @see backtype.storm.spout.ISpout#open(java.util.Map,
   *      backtype.storm.task.TopologyContext,
   *      backtype.storm.spout.SpoutOutputCollector)
   */
  @SuppressWarnings("unchecked")
  @Override
  public void open(@SuppressWarnings("rawtypes") final Map conf, final TopologyContext context,
      final SpoutOutputCollector collector) {
    try {
      LOG.debug("Opening...");
      this.collector = collector;
      Class<? extends LocationServiceFactory<SP>> locationServiceFactoryClass = (Class<? extends LocationServiceFactory<SP>>) Class
          .forName(configuration.getLocationServiceFactoryClassName());
      this.locationServiceFactory = locationServiceFactoryClass.newInstance();
      Class<? extends ServiceProviderSerialization<SP>> serviceProviderSerializationClass = (Class<? extends ServiceProviderSerialization<SP>>) Class
          .forName(configuration.getServiceProviderSerializationClassName());
      this.serviceProviderSerialization = serviceProviderSerializationClass.newInstance();
      Class<? extends EventReceptorFactory<CP>> eventReceptorFactoryClass = (Class<? extends EventReceptorFactory<CP>>) Class
          .forName(configuration.getEventReceptorFactoryClassName());
      this.eventReceptorFactory = eventReceptorFactoryClass.newInstance();
      locationService = locationServiceFactory.create(configuration.get(), serviceProviderSerialization);
      serviceListener = new MyServiceListener();
      locationService.addListener(serviceListener);
      locationService.start();
      LOG.info("Opened");
    } catch (Exception e) {
      LOG.error("Failed to open properly: " + e.getMessage(), e);
    }
  }

  /**
   * @see backtype.storm.topology.base.BaseRichSpout#close()
   */
  @Override
  public void close() {
    LOG.debug("Closing...");
    locationService.removeListener(serviceListener);
    locationService.stop();
    for (EventReceptor<CP> eventReceptor : eventReceptors.values()) {
      eventReceptor.stop();
    }
    eventReceptors.clear();
    LOG.info("Clossed");
  }

  protected void removeServiceProvider(String serviceProviderId) {
    EventReceptor<CP> eventReceptor = eventReceptors.remove(serviceProviderId);
    if (eventReceptor != null) {
      eventReceptor.stop();
      LOG.debug("Removed service provider: {}", eventReceptor);
    }
  }

  /**
   * @see backtype.storm.spout.ISpout#nextTuple()
   */
  @Override
  public void nextTuple() {
    try {
      if (collector != null) {
        // Emit any events we have queued
        for (final EventReceptor<CP> eventReceptor : eventReceptors.values()) {
          List<F2SEvent> events = eventReceptor.getEvents();
          for (final F2SEvent event : events) {
            LOG.trace("Received F2S event: {}", event);
            for (final F2SEventEmitter emitter : emitters) {
              emitter.emitEvent(event, collector);
            }
          }
          // Removing EventReceptor if:
          // - it's disconnected from the EventSender
          // - the associated ServiceProvider is not registered
          // - it does not have queued up events
          if (!eventReceptor.getStats().isConnected()
              && !locationService.containsServiceProvider(eventReceptor.getConnectionParameters().getId())
              && events.isEmpty()) {
            removeServiceProvider(eventReceptor.getConnectionParameters().getId());
          }
        }
      }
    } catch (final Exception ex) {
      LOG.error("There was a problem emitting events: " + ex.getLocalizedMessage(), ex);
      collector.reportError(ex);
    }
    // TODO remove: this is not needed since storm 0.8.1
    // Always sleep for a little bit
    try {
      Thread.sleep(SPOUT_SLEEP_TIME);
    } catch (final InterruptedException ex) {
      LOG.warn("Thread interupted: " + ex.getLocalizedMessage());
    }
  }

  /**
   * @see backtype.storm.topology.IComponent#declareOutputFields(backtype.storm.topology.OutputFieldsDeclarer)
   */
  @Override
  public void declareOutputFields(final OutputFieldsDeclarer declarer) {
    for (final F2SEventEmitter emitter : emitters) {
      emitter.declareOutputFields(declarer);
    }
  }
} public class FlumeSpoutConfiguration implements Supplier<Configuration>, Serializable {
  private static final long serialVersionUID = -8862782920258510684L;
  /**
   * Configuration attribute name for
   * {@link #getLocationServiceFactoryClassName()}
   */
  public static final String LOCATION_SERVICE_FACTORY_CLASS = "location.service.factory";
  /**
   * Default value for {@value #LOCATION_SERVICE_FACTORY_CLASS}
   */
  public static final String LOCATION_SERVICE_FACTORY_CLASS_DEFAULT = "com.comcast.viper.flume2storm.location.DynamicLocationServiceFactory";
  /**
   * Configuration attribute name for
   * {@link #getServiceProviderSerializationClassName()}
   */
  public static final String SERVICE_PROVIDER_SERIALIZATION_CLASS = "service.provider.serialization";
  /**
   * Default value for {@value #SERVICE_PROVIDER_SERIALIZATION_CLASS}
   */
  public static final String SERVICE_PROVIDER_SERIALIZATION_CLASS_DEFAULT = "com.comcast.viper.flume2storm.location.KryoNetServiceProviderSerialization";
  /**
   * Configuration attribute name for
   * {@link #getEventReceptorFactoryClassName()}
   */
  public static final String EVENT_RECEPTOR_FACTORY_CLASS = "event.receptor.factory";
  /**
   * Default value for {@value #EVENT_RECEPTOR_FACTORY_CLASS}
   */
  public static final String EVENT_RECEPTOR_FACTORY_CLASS_DEFAULT = "com.comcast.viper.flume2storm.connection.receptor.KryoNetEventReceptorFactory";

  protected String locationServiceFactoryClassName;
  protected String serviceProviderSerializationClassName;
  protected String eventReceptorFactoryClassName;
  protected Map<String, Object> configuration;

  /**
   * Builds a new {@link FlumeSpoutConfiguration} based on a Configuration
   * 
   * @param config
   *          The configuration to use
   * @return The newly created {@link F2SConfiguration}
   * @throws F2SConfigurationException
   *           If the class is not found in the class path
   */
  public static FlumeSpoutConfiguration from(Configuration config) throws F2SConfigurationException {
    FlumeSpoutConfiguration result = new FlumeSpoutConfiguration();
    try {
      result.setLocationServiceFactoryClassName(config.getString(LOCATION_SERVICE_FACTORY_CLASS,
          LOCATION_SERVICE_FACTORY_CLASS_DEFAULT));
    } catch (Exception e) {
      throw F2SConfigurationException.with(config, LOCATION_SERVICE_FACTORY_CLASS, e);
    }
    try {
      result.setServiceProviderSerializationClassName(config.getString(SERVICE_PROVIDER_SERIALIZATION_CLASS,
          SERVICE_PROVIDER_SERIALIZATION_CLASS_DEFAULT));
    } catch (Exception e) {
      throw F2SConfigurationException.with(config, SERVICE_PROVIDER_SERIALIZATION_CLASS, e);
    }
    try {
      result.setEventReceptorFactoryClassName(config.getString(EVENT_RECEPTOR_FACTORY_CLASS,
          EVENT_RECEPTOR_FACTORY_CLASS_DEFAULT));
    } catch (Exception e) {
      throw F2SConfigurationException.with(config, EVENT_RECEPTOR_FACTORY_CLASS, e);
    }
    result.configuration = getMapFromConfiguration(config);
    return result;
  }

  /**
   * @param configuration
   *          A configuration object
   * @return A map that contains the configuration key/values (in order to
   *         serialize it via Kryo)
   */
  protected static Map<String, Object> getMapFromConfiguration(Configuration configuration) {
    Map<String, Object> result = new HashMap<String, Object>();
    Iterator<?> it = configuration.getKeys();
    while (it.hasNext()) {
      String key = it.next().toString();
      result.put(key, configuration.getProperty(key));
    }
    return result;
  }

  /**
   * Empty constructor - initializes with default values when available
   */
  public FlumeSpoutConfiguration() {
    locationServiceFactoryClassName = LOCATION_SERVICE_FACTORY_CLASS_DEFAULT;
    serviceProviderSerializationClassName = SERVICE_PROVIDER_SERIALIZATION_CLASS_DEFAULT;
    eventReceptorFactoryClassName = EVENT_RECEPTOR_FACTORY_CLASS_DEFAULT;
    configuration = new HashMap<String, Object>();
  }

  /**
   * Copy constructor
   * 
   * @param other
   *          the configuration to copy
   */
  public FlumeSpoutConfiguration(final FlumeSpoutConfiguration other) {
    locationServiceFactoryClassName = other.locationServiceFactoryClassName;
    serviceProviderSerializationClassName = other.serviceProviderSerializationClassName;
    eventReceptorFactoryClassName = other.eventReceptorFactoryClassName;
    configuration = new HashMap<String, Object>(other.configuration);
  }

  /**
   * @return The class name of the location service factory
   */
  public String getLocationServiceFactoryClassName() {
    return locationServiceFactoryClassName;
  }

  /**
   * @param locationServiceFactoryClassName
   *          See {@link #getLocationServiceFactoryClassName()}
   * @throws ClassNotFoundException
   *           If the class specified is not found in the classpath
   */
  public void setLocationServiceFactoryClassName(String locationServiceFactoryClassName) throws ClassNotFoundException {
    Class<?> locationServiceFactoryClass = Class.forName(locationServiceFactoryClassName);
    Preconditions.checkArgument(LocationServiceFactory.class.isAssignableFrom(locationServiceFactoryClass),
        "The class must implement " + LocationServiceFactory.class.getCanonicalName());
    this.locationServiceFactoryClassName = locationServiceFactoryClassName;
  }

  /**
   * @return The class name of the service provider serialization
   */
  public String getServiceProviderSerializationClassName() {
    return serviceProviderSerializationClassName;
  }

  /**
   * @param serviceProviderSerializationClassName
   *          See {@link #getServiceProviderSerializationClassName()}
   * @throws ClassNotFoundException
   *           If the class specified is not found in the classpath
   */
  public void setServiceProviderSerializationClassName(String serviceProviderSerializationClassName)
      throws ClassNotFoundException {
    Class<?> serviceProviderSerializationClass = Class.forName(serviceProviderSerializationClassName);
    Preconditions.checkArgument(ServiceProviderSerialization.class.isAssignableFrom(serviceProviderSerializationClass),
        "The class must implement " + ServiceProviderSerialization.class.getCanonicalName());
    this.serviceProviderSerializationClassName = serviceProviderSerializationClassName;
  }

  /**
   * @return The class name of the event receptor factory
   */
  public String getEventReceptorFactoryClassName() {
    return eventReceptorFactoryClassName;
  }

  /**
   * @param eventReceptorFactoryClassName
   *          See {@link #getEventReceptorFactoryClassName()}
   * @throws ClassNotFoundException
   *           If the class specified is not found in the classpath
   */
  public void setEventReceptorFactoryClassName(String eventReceptorFactoryClassName) throws ClassNotFoundException {
    Class<?> eventReceptorFactoryClass = Class.forName(eventReceptorFactoryClassName);
    Preconditions.checkArgument(EventReceptorFactory.class.isAssignableFrom(eventReceptorFactoryClass),
        "The class must implement " + EventReceptorFactory.class.getCanonicalName());
    this.eventReceptorFactoryClassName = eventReceptorFactoryClassName;
  }

  /**
   * @see com.google.common.base.Supplier#get()
   * @return This object as a {@link Configuration}
   */
  public Configuration get() {
    return new MapConfiguration(configuration);
  }

  /**
   * @param map
   *          The {@link FlumeSpout} configuration as a {@link Map}
   */
  public void setConfiguration(Map<String, Object> map) {
    this.configuration = map;
  }

  /**
   * @param configuration
   *          The {@link FlumeSpout} configuration as a
   *          {@link FlumeSpoutConfiguration}
   */
  public void setConfiguration(Configuration configuration) {
    this.configuration = getMapFromConfiguration(configuration);
  }

  /**
   * @see java.lang.Object#hashCode()
   */
  @Override
  public int hashCode() {
    return new HashCodeBuilder().append(locationServiceFactoryClassName).append(serviceProviderSerializationClassName)
        .append(eventReceptorFactoryClassName).append(configuration).hashCode();
  }

  /**
   * @see java.lang.Object#equals(java.lang.Object)
   */
  @Override
  public boolean equals(Object obj) {
    if (this == obj)
      return true;
    if (obj == null)
      return false;
    if (getClass() != obj.getClass())
      return false;
    FlumeSpoutConfiguration other = (FlumeSpoutConfiguration) obj;
    return new EqualsBuilder().append(this.eventReceptorFactoryClassName, other.eventReceptorFactoryClassName)
        .append(this.locationServiceFactoryClassName, other.locationServiceFactoryClassName)
        .append(this.serviceProviderSerializationClassName, other.serviceProviderSerializationClassName)
        .append(this.configuration, other.configuration).isEquals();
  }

  /**
   * @see java.lang.Object#toString()
   */
  @Override
  public String toString() {
    return new ToStringBuilder("FlumeSpoutConfiguration")
        .append("locationServiceFactoryClass", locationServiceFactoryClassName)
        .append("serviceProviderSerializationClass", serviceProviderSerializationClassName)
        .append("eventReceptorFactory", eventReceptorFactoryClassName).append("configuration", configuration)
        .toString();
  }
} public class MemoryStorage {
  private static MemoryStorage instance = new MemoryStorage();

  public static MemoryStorage getInstance() {
    return instance;
  }

  protected SortedSet<F2SEvent> receivedEvents;

  public MemoryStorage() {
    receivedEvents = new TreeSet<F2SEvent>(new F2SEventComparator());
  }

  public synchronized SortedSet<F2SEvent> getReceivedEvents() {
    return receivedEvents;
  }

  public synchronized void addEvent(F2SEvent event) {
    receivedEvents.add(event);
  }

  public synchronized void clear() {
    receivedEvents.clear();
  }
} public class TestBolt extends BaseRichBolt {
  private static final long serialVersionUID = -4437924501877270487L;
  protected OutputCollector collector;

  /**
   * @see backtype.storm.task.IBolt#prepare(java.util.Map,
   *      backtype.storm.task.TopologyContext,
   *      backtype.storm.task.OutputCollector)
   */
  @Override
  public void prepare(Map stormConf, TopologyContext context,
      OutputCollector collector) {
    this.collector = collector;
  }

  /**
   * @see backtype.storm.task.IBolt#execute(backtype.storm.tuple.Tuple)
   */
  @Override
  public void execute(Tuple input) {
    final F2SEvent event = (F2SEvent) input
        .getValueByField(BasicF2SEventEmitter.F2S_EVENT);
    MemoryStorage.getInstance().addEvent(event);
    collector.ack(input);
  }

  /**
   * @see backtype.storm.topology.IComponent#declareOutputFields(backtype.storm.topology.OutputFieldsDeclarer)
   */
  @Override
  public void declareOutputFields(OutputFieldsDeclarer declarer) {
    // This bolt is not emitting anything
  }
} public interface TestCondition {
	/**
	 * @return True when the condition is fulfilled, false otherwise
	 */
	boolean evaluate();
} public class TestUtils {
  private static final int DEFAULT_RETRY_TIMEOUT = 100;
  private static Random random = new Random();

  /**
   * @return An ephemeral port available for test usage
   */
  public static int getAvailablePort() {
    try (ServerSocket s = new ServerSocket(0)) {
      return s.getLocalPort();
    } catch (Exception e) {
      throw new AssertionError("Failed to find available port", e);
    }
  }

  /**
   * @param n
   *          Max int allowed
   * @return A random number between 1 and n
   */
  public static final int getRandomPositiveInt(int n) {
    Preconditions.checkArgument(n > 1, "Cannot generate this kind of number!");
    return random.nextInt(n - 1) + 1;
  }

  /**
   * Waits that the condition is fulfilled - careful, this may never return!
   * 
   * @param condition
   *          The condition to evaluate. It should be fast to evaluate
   * @throws InterruptedException
   *           If wait is interrupted
   */
  public static void waitFor(final TestCondition condition) throws InterruptedException {
    waitFor(condition, Integer.MAX_VALUE);
  }

  /**
   * Waits that the condition is fulfilled, for up to a specified amount of time
   * 
   * @param condition
   *          The condition to evaluate. It should be fast to evaluate
   * @param maxWaitInMs
   *          The maximum number of milliseconds to wait
   * @return True if the condition is fulfilled in time, false otherwise
   * @throws InterruptedException
   *           If wait is interrupted
   */
  public static boolean waitFor(final TestCondition condition, final int maxWaitInMs) throws InterruptedException {
    return waitFor(condition, maxWaitInMs, DEFAULT_RETRY_TIMEOUT);
  }

  /**
   * Waits that the condition is fulfilled, for up to a specified amount of time
   * 
   * @param condition
   *          The condition to evaluate. It should be fast to evaluate
   * @param maxWaitInMs
   *          The maximum number of milliseconds to wait
   * @param retryTimeout
   *          The number of milliseconds to wait before retry the test
   * @return True if the condition is fulfilled in time, false otherwise
   * @throws InterruptedException
   *           If wait is interrupted
   */
  public static boolean waitFor(final TestCondition condition, final int maxWaitInMs, final int retryTimeout)
      throws InterruptedException {
    final long t0 = System.currentTimeMillis();
    while (!condition.evaluate() && (System.currentTimeMillis() - t0) < maxWaitInMs) {
      Thread.sleep(retryTimeout);
    }
    if (condition.evaluate()) {
      return true;
    }
    throw new AssertionError("TestCondition did not validate in time");
  }
}
import static org.assertj.core.api.Assertions.assertThat;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.SortedSet;
import junit.framework.Assert;
import org.apache.commons.configuration.BaseConfiguration;
import org.apache.commons.configuration.CombinedConfiguration;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.ConfigurationConverter;
import org.apache.commons.configuration.ConfigurationUtils;
import org.apache.commons.configuration.MapConfiguration;
import org.apache.commons.lang3.StringUtils;
import org.apache.flume.Context;
import org.apache.flume.Sink;
import org.apache.flume.SinkRunner;
import org.apache.flume.channel.PseudoTxnMemoryChannel;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.lifecycle.LifecycleState;
import org.apache.flume.sink.LoadBalancingSinkProcessor;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import backtype.storm.Config;
import backtype.storm.ILocalCluster;
import backtype.storm.Testing;
import backtype.storm.generated.KillOptions;
import backtype.storm.testing.MkClusterParam;
import backtype.storm.testing.TestJob;
import backtype.storm.topology.TopologyBuilder;
import com.comcast.viper.flume2storm.connection.KryoNetParameters;
import com.comcast.viper.flume2storm.connection.parameters.ConnectionParameters;
import com.comcast.viper.flume2storm.connection.parameters.KryoNetConnectionParameters;
import com.comcast.viper.flume2storm.connection.parameters.KryoNetConnectionParametersFactory;
import com.comcast.viper.flume2storm.connection.parameters.SimpleConnectionParameters;
import com.comcast.viper.flume2storm.connection.parameters.SimpleConnectionParametersFactory;
import com.comcast.viper.flume2storm.connection.receptor.KryoNetEventReceptorFactory;
import com.comcast.viper.flume2storm.connection.receptor.SimpleEventReceptorFactory;
import com.comcast.viper.flume2storm.connection.sender.EventSender;
import com.comcast.viper.flume2storm.connection.sender.KryoNetEventSenderFactory;
import com.comcast.viper.flume2storm.connection.sender.SimpleEventSenderFactory;
import com.comcast.viper.flume2storm.event.F2SEvent;
import com.comcast.viper.flume2storm.event.F2SEventFactory;
import com.comcast.viper.flume2storm.location.DynamicLocationServiceConfiguration;
import com.comcast.viper.flume2storm.location.DynamicLocationServiceFactory;
import com.comcast.viper.flume2storm.location.KryoNetServiceProvider;
import com.comcast.viper.flume2storm.location.KryoNetServiceProviderSerialization;
import com.comcast.viper.flume2storm.location.ServiceProvider;
import com.comcast.viper.flume2storm.location.ServiceProviderConfigurationLoader;
import com.comcast.viper.flume2storm.location.SimpleLocationServiceFactory;
import com.comcast.viper.flume2storm.location.SimpleServiceProviderSerialization;
import com.comcast.viper.flume2storm.location.StaticLocationServiceConfiguration;
import com.comcast.viper.flume2storm.location.StaticLocationServiceFactory;
import com.comcast.viper.flume2storm.sink.StormSink;
import com.comcast.viper.flume2storm.sink.StormSinkConfiguration;
import com.comcast.viper.flume2storm.spout.BasicF2SEventEmitter;
import com.comcast.viper.flume2storm.spout.F2SEventEmitter;
import com.comcast.viper.flume2storm.spout.FlumeSpout;
import com.comcast.viper.flume2storm.spout.FlumeSpoutConfiguration;
import com.comcast.viper.flume2storm.spout.MemoryStorage;
import com.comcast.viper.flume2storm.spout.TestBolt;
import com.comcast.viper.flume2storm.utility.test.TestCondition;
import com.comcast.viper.flume2storm.utility.test.TestUtils;
import com.google.common.collect.ImmutableList;
/**
 * Copyright 2014 Comcast Cable Communications Management, LLC
 *
 * 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.comcast.viper.flume2storm;







/**
 * Integration tests that put all the components together. Notice that the only
 * difference between the use of various implementations is the configuration of
 * it all. <br />
 * Note that we're using the ZooKeeper server that is created as part of the
 * test topology when needed.
 * 
 */
public class IntegrationTest {
  private static final String TEST_TOPOLOGY_NAME = "TestTopology";
  protected static final Logger LOG = LoggerFactory.getLogger(IntegrationTest.class);
  private static final String CHANNEL_CONFIG = "channel";
  private static final String SINK1_CONFIG = "sink1";
  private static final String SINK2_CONFIG = "sink2";
  private static final String SINK_PROCESSOR_CONFIG = "processor";
  private static final String SPOUT_CONFIG = "spout";
  private static final int TEST_TIMEOUT = 60000;
  private static final int HALF_NB_EVENTS = 15000;
  private static final int NB_EVENTS = HALF_NB_EVENTS * 2;
  private static final int ZK_PORT = 2000;

  /**
   * This contains all the other configuration objects (sink1, sink2, spout,
   * ...)
   */
  protected static SortedSet<F2SEvent> eventsToSent;
  protected CombinedConfiguration config;

  /**
   * Initializes integration tests
   */
  @BeforeClass
  public static void init() {
    // Generating test events
    eventsToSent = F2SEventFactory.getInstance().generateRandomEvents(NB_EVENTS);
  }

  /**
   * @throws Exception
   *           If anything went wrong
   */
  @Before
  public void setup() throws Exception {
    config = new CombinedConfiguration();
    // Flume channel configuration
    BaseConfiguration channelConfig = new BaseConfiguration();
    channelConfig.addProperty("keep-alive", "1"); // Speeds up test
    channelConfig.addProperty("capacity", "" + NB_EVENTS);
    config.addConfiguration(channelConfig, CHANNEL_CONFIG);

    MemoryStorage.getInstance().clear();
  }

  /**
   * Integration test with the test implementation of the Location Service and
   * the Connection API
   * 
   * @throws Exception
   *           If anything went wrong
   */
  // @Test
  public void testTestImpl() throws Exception {
    // Base storm sink configuration
    BaseConfiguration sinkBaseConfig = new BaseConfiguration();
    sinkBaseConfig.addProperty(StormSinkConfiguration.LOCATION_SERVICE_FACTORY_CLASS,