@SuppressWarnings("javadoc")
public class F2SConfigurationException extends Exception {
  private static final long serialVersionUID = 4941423443352427505L;

  /**
   * @param config
   *          The configuration that contains the invalid parameter
   * @param parameter
   *          The name of the configuration parameter that is invalid
   * @param throwable
   *          The exception that occurred when setting the parameter's value
   * @return The newly built F2SConfigurationException related to a specific
   *         invalid parameter
   */
  public static F2SConfigurationException with(Configuration config, String parameter, Throwable throwable) {
    return F2SConfigurationException.with(parameter, config.getProperty(parameter), throwable);
  }

  /**
   * @param parameter
   *          The name of the configuration parameter that is invalid
   * @param value
   *          The configured value of the parameter
   * @param throwable
   *          The exception that occurred when setting the parameter's value
   * @return The newly built F2SConfigurationException related to a specific
   *         invalid parameter
   */
  public static F2SConfigurationException with(String parameter, Object value, Throwable throwable) {
    return new F2SConfigurationException(buildErrorMessage(parameter, value, throwable), throwable);
  }

  protected static final String buildErrorMessage(String parameter, Object value, Throwable throwable) {
    return new StringBuilder("Configuration attribute \"").append(parameter).append("\" has invalid value (\"")
        .append(value).append("\"). ").append(throwable.getClass().getSimpleName()).append(": ")
        .append(throwable.getLocalizedMessage()).toString();
  }

  /**
   * @param parameter
   *          The name of the configuration parameter that is missing
   * @return The newly built F2SConfigurationException related to a specific
   *         missing parameter
   */
  public static F2SConfigurationException missing(String parameter) {
    return new F2SConfigurationException(new StringBuilder("Configuration attribute \"").append(parameter)
        .append("\" is required but not specified").toString());
  }

  public F2SConfigurationException() {
    super();
  }

  public F2SConfigurationException(String message) {
    super(message);
  }

  public F2SConfigurationException(Throwable throwable) {
    super(throwable);
  }

  public F2SConfigurationException(String message, Throwable throwable) {
    super(message, throwable);
  }
} public interface ConnectionParameters {
  /**
   * @return A unique identifier of the associated {@link EventSender} /
   *         {@link ServiceProvider} (hostname:port for instance)
   * 
   */
  String getId();
} public interface ConnectionParametersFactory<CP extends ConnectionParameters> {
  /**
   * Creates a new {@link ConnectionParameters} based on the configuration
   * provided
   * 
   * @param config
   *          Configuration for the connections parameters
   * @return The newly created {@link ConnectionParameters}
   * @throws F2SConfigurationException
   *           If the configuration specified is invalid
   */
  CP create(Configuration config) throws F2SConfigurationException;
} 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 interface EventSenderFactory<CP extends ConnectionParameters> {
  /**
   * Creates a new {@link EventSender} based on the connection parameters
   * provided
   * 
   * @param connectionParams
   *          Connections parameters to use to configure the {@link EventSender}
   * @param config
   *          Additional configuration for the creation of the event sender
   * @return The newly created {@link EventSender}
   * @throws F2SConfigurationException
   *           If the configuration specified is invalid
   */
  EventSender<CP> create(CP connectionParams, Configuration config) throws F2SConfigurationException;
} public final class EventSenderStats implements EventSenderStatsMBean {
  protected final String eventSenderId;
  protected final AtomicInteger nbClients;
  protected final AtomicLong nbEventIn;
  protected final AtomicLong nbEventOut;
  protected final AtomicLong nbEventFailed;

  /**
   * Default constructor
   * 
   * @param eventSenderId
   *          The {@link EventSender} identifier
   */
  public EventSenderStats(final String eventSenderId) {
    this.eventSenderId = eventSenderId;
    nbClients = new AtomicInteger();
    nbEventIn = new AtomicLong();
    nbEventOut = new AtomicLong();
    nbEventFailed = new AtomicLong();
  }

  /**
   * @see com.comcast.viper.flume2storm.connection.sender.EventSenderStatsMBean#getEventSenderId()
   */
  @Override
  public String getEventSenderId() {
    return eventSenderId;
  }

  /**
   * @see com.comcast.viper.flume2storm.connection.sender.EventSenderStatsMBean#reset()
   */
  @Override
  public final EventSenderStats reset() {
    nbClients.set(0);
    nbEventIn.set(0);
    nbEventOut.set(0);
    nbEventFailed.set(0);
    return this;
  }

  /**
   * @see com.comcast.viper.flume2storm.connection.sender.EventSenderStatsMBean#getNbClients()
   */
  @Override
  public final int getNbClients() {
    return nbClients.get();
  }

  /**
   * See {@link #getNbClients()}
   * 
   * @return This object
   */
  public final EventSenderStats incrClients() {
    nbClients.incrementAndGet();
    return this;
  }

  /**
   * See {@link #getNbClients()}
   * 
   * @return This object
   */
  public final EventSenderStats decrClients() {
    nbClients.decrementAndGet();
    return this;
  }

  /**
   * See {@link #getNbClients()}
   * 
   * @return This object
   */
  public final EventSenderStatsMBean resetClients() {
    nbClients.set(0);
    return this;
  }

  /**
   * @see com.comcast.viper.flume2storm.connection.sender.EventSenderStatsMBean#getNbEventsIn()
   */
  @Override
  public final long getNbEventsIn() {
    return nbEventIn.get();
  }

  /**
   * See {@link #getNbEventsIn()}
   * 
   * @return This object
   */
  public final EventSenderStats incrEventsIn() {
    return incrEventsIn(1);
  }

  /**
   * See {@link #getNbEventsIn()}
   * 
   * @param i
   *          The number of events to increment
   * 
   * @return This object
   */
  public final EventSenderStats incrEventsIn(final int i) {
    nbEventIn.addAndGet(i);
    return this;
  }

  /**
   * @see com.comcast.viper.flume2storm.connection.sender.EventSenderStatsMBean#getNbEventsOut()
   */
  @Override
  public final long getNbEventsOut() {
    return nbEventOut.get();
  }

  /**
   * See {@link #getNbEventsOut()}
   * 
   * @return This object
   */
  public final EventSenderStats incrEventsOut() {
    return incrEventsOut(1);
  }

  /**
   * See {@link #getNbEventsOut()}
   * 
   * @param i
   *          The number of events to increment
   * 
   * @return This object
   */
  public final EventSenderStats incrEventsOut(final int i) {
    nbEventOut.addAndGet(i);
    return this;
  }

  /**
   * @see com.comcast.viper.flume2storm.connection.sender.EventSenderStatsMBean#getNbEventsFailed()
   */
  @Override
  public final long getNbEventsFailed() {
    return nbEventFailed.get();
  }

  /**
   * See {@link #getNbEventsFailed()}
   * 
   * @return This object
   */
  public final EventSenderStatsMBean incrEventsFailed() {
    return incrEventsFailed(1);
  }

  /**
   * See {@link #getNbEventsFailed()}
   * 
   * @param i
   *          The number of events to increment
   * 
   * @return This object
   */
  public final EventSenderStatsMBean incrEventsFailed(final int i) {
    nbEventFailed.addAndGet(i);
    return this;
  }

  /**
   * @param other
   *          Another KryoNet Event Sender statistics object
   * @return True if the stats are the same
   */
  public boolean sameAs(EventSenderStats other) {
    Preconditions.checkNotNull(other);
    if (getNbClients() != other.getNbClients())
      return false;
    if (getNbEventsIn() != other.getNbEventsIn())
      return false;
    if (getNbEventsOut() != other.getNbEventsOut())
      return false;
    if (getNbEventsFailed() != other.getNbEventsFailed())
      return false;
    return true;
  }

  /**
   * @see java.lang.Object#hashCode()
   */
  @Override
  public int hashCode() {
    return new HashCodeBuilder().append(nbClients).append(nbEventIn).append(nbEventOut).append(nbEventFailed)
        .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;
    EventSenderStats other = (EventSenderStats) obj;
    return new EqualsBuilder().append(this.nbClients, other.nbClients).append(this.nbEventIn, other.nbEventIn)
        .append(this.nbEventOut, other.nbEventOut).append(this.nbEventFailed, other.nbEventFailed).isEquals();
  }

  /**
   * @see java.lang.Object#toString()
   */
  @Override
  public String toString() {
    return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE).append("NbClients", getNbClients())
        .append("NbEventsIn", getNbEventsIn()).append("NbEventsOut", getNbEventsOut())
        .append("NbEventsFailed", getNbEventsFailed()).toString();
  }
} public interface EventSenderStatsMBean {

  /**
   * @return The {@link EventSender} / {@link ServiceProvider} identifier
   */
  String getEventSenderId();

  /**
   * @return The number of {@link EventReceptor} connected
   */
  int getNbClients();

  /**
   * @return The total number of {@link F2SEvent} ingested by the
   *         {@link EventSender}
   */
  long getNbEventsIn();

  /**
   * @return The total number of {@link F2SEvent} sent by the
   *         {@link EventSender}
   */
  long getNbEventsOut();

  /**
   * @return The total number of {@link F2SEvent} that failed to be sent by the
   *         {@link EventSender}
   */
  long getNbEventsFailed();

  /**
   * Resets all the metrics
   * 
   * @return This object
   */
  EventSenderStatsMBean reset();
} 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 interface LocationService<SP extends ServiceProvider<?>> {
  /**
   * Starts the {@link LocationService}
   * 
   * @return True if the operation succeeded, false otherwise
   */
  boolean start();

  /**
   * Stops the {@link LocationService}
   * 
   * @return True if the operation succeeded, false otherwise
   */
  boolean stop();

  /**
   * Registers a {@link ServiceProvider} to the location service
   * 
   * @param serviceProvider
   *          The {@link ServiceProvider} that joined the service
   */
  void register(SP serviceProvider);

  /**
   * Unregisters a {@link ServiceProvider} from the location service
   * 
   * @param serviceProvider
   *          The {@link ServiceProvider} that left the service
   */
  void unregister(SP serviceProvider);

  /**
   * @return The (unmodifiable) list of information about the currently active
   *         service providers
   */
  List<SP> getServiceProviders();

  /**
   * @param serviceProvider
   *          A {@link ServiceProvider}
   * @return True if the specified {@link ServiceProvider} is registered, false
   *         otherwise
   */
  boolean containsServiceProvider(SP serviceProvider);

  /**
   * @param serviceProviderId
   *          A {@link ServiceProvider} identifier
   * @return True if the {@link ServiceProvider} identified with the specified
   *         string is registered, false otherwise
   */
  boolean containsServiceProvider(String serviceProviderId);

  /**
   * Adds the specified listener. Upon addition, the listener receives a
   * notification for all the {@link ServiceProvider} currently registered
   * 
   * @param serviceListener
   *          A {@link ServiceListener} for the specific type of
   *          {@link ServiceProvider}
   */
  void addListener(final ServiceListener<SP> serviceListener);

  /**
   * Removes the specified listener
   * 
   * @param serviceListener
   *          A {@link ServiceListener} for the specific type of
   *          {@link ServiceProvider}
   */
  void removeListener(final ServiceListener<SP> serviceListener);

  /**
   * @return The serialization factory for this service provider
   */
  ServiceProviderSerialization<SP> getSerialization();
} public interface LocationServiceFactory<SP extends ServiceProvider<?>> {
  /**
   * Creates a new {@link LocationService} based on the configuration provided
   * 
   * @param config
   *          Configuration for the location service
   * @param serviceProviderSerialization
   *          The {@link ServiceProviderSerialization} to use
   * @return The newly created {@link LocationService}
   * @throws F2SConfigurationException
   *           If the configuration specified is invalid
   */
  LocationService<SP> create(Configuration config, ServiceProviderSerialization<SP> serviceProviderSerialization)
      throws F2SConfigurationException;
} public interface ServiceProvider<CP extends ConnectionParameters> extends Serializable, Comparable<ServiceProvider<CP>> {
  /**
   * @return The connection parameters for this service provider
   */
  CP getConnectionParameters();
} public interface ServiceProviderSerialization<SP extends ServiceProvider<?>> extends Serializable {
  /**
   * Serializes a {@link ServiceProvider} into a byte array
   * 
   * @param serviceProvider
   *          The service provider to serialize
   * @return The serialized service provider
   */
  byte[] serialize(SP serviceProvider);

  /**
   * Deserializes a {@link ServiceProvider} from a byte array
   * 
   * @param bytes
   *          The serialized service provider
   * @return The deserialized service provider
   */
  SP deserialize(byte[] bytes);
}
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.ConfigurationUtils;
import org.apache.commons.configuration.MapConfiguration;
import org.apache.flume.Channel;
import org.apache.flume.ChannelException;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurable;
import org.apache.flume.instrumentation.SinkCounter;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.comcast.viper.flume2storm.F2SConfigurationException;
import com.comcast.viper.flume2storm.connection.parameters.ConnectionParameters;
import com.comcast.viper.flume2storm.connection.parameters.ConnectionParametersFactory;
import com.comcast.viper.flume2storm.connection.sender.EventSender;
import com.comcast.viper.flume2storm.connection.sender.EventSenderFactory;
import com.comcast.viper.flume2storm.connection.sender.EventSenderStats;
import com.comcast.viper.flume2storm.connection.sender.EventSenderStatsMBean;
import com.comcast.viper.flume2storm.event.F2SEvent;
import com.comcast.viper.flume2storm.location.LocationService;
import com.comcast.viper.flume2storm.location.LocationServiceFactory;
import com.comcast.viper.flume2storm.location.ServiceProvider;
import com.comcast.viper.flume2storm.location.ServiceProviderSerialization;
import com.google.common.base.Preconditions;
/**
 * 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.sink;




/**
 * This Flume sink sends the Flume events picked up from the channel to the
 * clients connected to it, load-balancing the data between them.
 * 
 * @param <CP>
 *          The Connection Parameters class
 * @param <SP>
 *          The Service Provider class
 */
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;