public static FlumeAvroManager create(
    final List<RemoteFlumeAgent> agents,
    final Properties overrides,
    final Integer batchSize,
    final Long reportingWindow,
    final Integer reporterMaxThreadPoolSize,
    final Integer reporterMaxQueueSize,
    final ContextAware context) {

    if (agents != null && agents.size() > 0) {
      Properties props = buildFlumeProperties(agents);
      props.putAll(overrides);
      return new FlumeAvroManager(props, reportingWindow, batchSize, reporterMaxThreadPoolSize, reporterMaxQueueSize, context);
    } else {
      context.addError("No valid agents configured");
    }

  return null;
}
 

private FlumeAvroManager(final Properties props,
                         final Long reportingWindowReq,
                         final Integer batchSizeReq,
                         final Integer reporterMaxThreadPoolSizeReq,
                         final Integer reporterMaxQueueSizeReq,
                         final ContextAware context) {
  this.loggingContext = context;

  final int reporterMaxThreadPoolSize = reporterMaxThreadPoolSizeReq == null ?
          DEFAULT_REPORTER_MAX_THREADPOOL_SIZE : reporterMaxThreadPoolSizeReq;
  final int reporterMaxQueueSize = reporterMaxQueueSizeReq == null ?
          DEFAULT_REPORTER_MAX_QUEUE_SIZE : reporterMaxQueueSizeReq;

  this.reporter = new EventReporter(props, loggingContext, reporterMaxThreadPoolSize, reporterMaxQueueSize);
  this.evQueue = new ArrayBlockingQueue<Event>(1000);
  final long reportingWindow = hamonizeReportingWindow(reportingWindowReq);
  final int batchSize = batchSizeReq == null ? DEFAULT_BATCH_SIZE : batchSizeReq;
  this.asyncThread = new AsyncThread(evQueue, batchSize, reportingWindow);
  loggingContext.addInfo("Created a new flume agent with properties: " + props.toString());
  asyncThread.start();
}
 

private static void initAndAddListener(KonkerLoggerContext loggerContext, StatusListener listener) {
    if(listener != null) {
        if(listener instanceof ContextAware) {
            ((ContextAware)listener).setContext(loggerContext);
        }

        if(listener instanceof LifeCycle) {
            ((LifeCycle)listener).start();
        }

        loggerContext.getStatusManager().add(listener);
    }

}
 

private void inject(Context context, Converter<ILoggingEvent> head, boolean injectAll) {
	for (Converter<ILoggingEvent> c = head; c != null; c = c.getNext()) {
		if (c instanceof CompositeConverter<?>) {
			Converter<ILoggingEvent> childConverter = ((CompositeConverter<ILoggingEvent>) c).getChildConverter();
			inject(context, childConverter, true);
		}
		if (injectAll && c instanceof ContextAware) {
			((ContextAware) c).setContext(context);
		}
	}
}
 

public EventReporter(final Properties properties, final ContextAware context,
                     final int maximumThreadPoolSize, final int maxQueueSize) {
  BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<Runnable>(maxQueueSize);
  this.connectionProps = properties;
  this.loggingContext = context;

  int corePoolSize = 1;
  TimeUnit threadKeepAliveUnits = TimeUnit.SECONDS;
  int threadKeepAliveTime = 30;
  RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy();

  this.es = new ThreadPoolExecutor(corePoolSize, maximumThreadPoolSize, threadKeepAliveTime,
          threadKeepAliveUnits, blockingQueue, handler);
}
 

public void start(LifeCycle lifeCycle) {
    if (lifeCycle instanceof ContextAware) {
        ((ContextAware) lifeCycle).setContext(context);
    }
    lifeCycle.start();
}
 

public void start(LifeCycle lifeCycle) {
	if (lifeCycle instanceof ContextAware) {
		((ContextAware) lifeCycle).setContext(this.context);
	}
	lifeCycle.start();
}