/*
 * Copyright (c) 2014-2020 Snowplow Analytics Ltd. All rights reserved.
 *
 * This program is licensed to you under the Apache License Version 2.0,
 * and you may not use this file except in compliance with the Apache License Version 2.0.
 * You may obtain a copy of the Apache License Version 2.0 at http://www.apache.org/licenses/LICENSE-2.0.
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the Apache License Version 2.0 is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
 */
package com.snowplowanalytics.snowplow.tracker.emitter;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import com.google.common.base.Preconditions;

import com.snowplowanalytics.snowplow.tracker.http.HttpClientAdapter;
import com.snowplowanalytics.snowplow.tracker.http.OkHttpClientAdapter;
import com.snowplowanalytics.snowplow.tracker.payload.TrackerEvent;

import okhttp3.OkHttpClient;

/**
 * AbstractEmitter class which contains common elements to
 * the emitters wrapped in a builder format.
 */
public abstract class AbstractEmitter implements Emitter {

    protected HttpClientAdapter httpClientAdapter;
    protected RequestCallback requestCallback;
    protected ExecutorService executor;

    public static abstract class Builder<T extends Builder<T>> {

        private HttpClientAdapter httpClientAdapter; // Optional
        private RequestCallback requestCallback = null; // Optional
        private int threadCount = 50; // Optional
        private String collectorUrl = null; // Required if not specifying a httpClientAdapter
        protected abstract T self();

        /**
         * Adds the HttpClientAdapter to the AbstractEmitter
         *
         * @param httpClientAdapter the adapter to use
         * @return itself
         */
        public T httpClientAdapter(final HttpClientAdapter httpClientAdapter) {
            this.httpClientAdapter = httpClientAdapter;
            return self();
        }

        /**
         * An optional Request Callback for adding the ability to handle failure cases
         * for sending.
         *
         * @param requestCallback the emitter request callback
         * @return itself
         */
        public T requestCallback(final RequestCallback requestCallback) {
            this.requestCallback = requestCallback;
            return self();
        }

        /**
         * Sets the Thread Count for the ExecutorService
         *
         * @param threadCount the size of the thread pool
         * @return itself
         */
        public T threadCount(final int threadCount) {
            this.threadCount = threadCount;
            return self();
        }

        /**
         * Sets the emitter url for when a httpClientAdapter is not specified
         * Will be used to create the default OkHttpClientAdapter.
         *
         * @param collectorUrl the url for the default httpClientAdapter
         * @return itself
         */
        public T url(final String collectorUrl) {
            this.collectorUrl = collectorUrl;
            return self();
        }
    }

    private static class Builder2 extends Builder<Builder2> {
        @Override
        protected Builder2 self() {
            return this;
        }
    }

    public static Builder<?> builder() {
        return new Builder2();
    }

    protected AbstractEmitter(final Builder<?> builder) {

        // Precondition checks
        Preconditions.checkArgument(builder.threadCount > 0, "threadCount must be greater than 0");

        if (builder.httpClientAdapter != null) {
            this.httpClientAdapter = builder.httpClientAdapter;
        } else {
            Preconditions.checkNotNull(builder.collectorUrl, "Collector url must be specified if not using a httpClientAdapter");

            this.httpClientAdapter = OkHttpClientAdapter.builder()
                    .url(builder.collectorUrl)
                    .httpClient(
                        new OkHttpClient()) // use okhttp as a default
                    .build();
        }

        this.requestCallback = builder.requestCallback;
        this.executor = Executors.newScheduledThreadPool(builder.threadCount);
    }

    /**
     * Adds an event to the buffer
     *
     * @param event an event
     */
    @Override
    public abstract void emit(TrackerEvent event);

    /**
     * Customize the emitter buffer size to any valid integer greater than zero.
     * Has no effect on SimpleEmitter
     *
     * @param bufferSize number of events to collect before sending
     */
    @Override
    public abstract void setBufferSize(final int bufferSize);

    /**
     * Removes all events from the buffer and sends them
     */
    @Override
    public abstract void flushBuffer();

    /**
     * Gets the Emitter Buffer Size - Will always be 1 for SimpleEmitter
     *
     * @return the buffer size
     */
    @Override
    public abstract int getBufferSize();

    /**
     * Returns List of Events that are in the buffer.
     *
     * @return the buffered events
     */
    @Override
    public abstract List<TrackerEvent> getBuffer();

    /**
     * Sends a runnable to the executor service.
     *
     * @param runnable the runnable to be queued
     */
    protected void execute(final Runnable runnable) {
        this.executor.execute(runnable);
    }

    /**
     * Checks whether the response code was a success or not.
     *
     * @param code the response code
     * @return whether it is in the success range
     */
    protected boolean isSuccessfulSend(final int code) {
        return code >= 200 && code < 300;
    }
}

@@ -865,46 +865,8 @@
 ors;
-%0Aimport java.util.concurrent.TimeUnit;
 %0A%0Aim
@@ -1702,32 +1702,107 @@
 50; // Optional%0A
+        private ExecutorService requestExecutorService = null; // Optional%0A
         private 
@@ -1913,24 +1913,519 @@
  T self();%0A%0A
+        /**%0A         * Set a custom ExecutorService to send http request.%0A         *%0A         *  /!%5C Be aware that calling %60close()%60 on a BatchEmitter instance has a side-effect and will shutdown that ExecutorService.%0A         * @param executorService the ExecutorService to use%0A         * @return itself%0A         */%0A        public T requestExecutorService(final ExecutorService executorService) %7B%0A            this.requestExecutorService = executorService;%0A            return self();%0A        %7D%0A%0A
         /**%0A
@@ -4889,32 +4889,167 @@
 equestCallback;%0A
+        if (builder.requestExecutorService != null) %7B%0A            this.executor = builder.requestExecutorService;%0A        %7D else %7B%0A    
         this.exe
@@ -5111,16 +5111,26 @@
 Count);%0A
+        %7D%0A
     %7D%0A%0A 

Allow Emitter to use a custom ExecutorService (close #278)