public class S3Configuration {

    /**
     * SSE options when using S3
     * See https://docs.aws.amazon.com/AmazonS3/latest/dev/serv-side-encryption.html
     */
    public enum SSEType {
        SSE_S3,     // Server-Side Encryption with Amazon S3-Managed Key
        SSE_KMS,    // Server-Side Encryption with AWS KMS-Managed Keys
        SSE_C,      // Server-Side Encryption with Customer-Provided Keys
    }

    /**
     * Configuration for working with SSE in AWS S3
     */
    public final static class S3SSEConfiguration {
        private final SSEType keyType;
        private final String keyId;

        public S3SSEConfiguration(SSEType keyType, String keyId) {
            if (keyType != SSEType.SSE_S3) {
                throw new UnsupportedOperationException("Only SSE_S3 is supported at this time.");
            }
            this.keyType = keyType;
            this.keyId = keyId;
        }

        public SSEType getKeyType() {
            return keyType;
        }

        public String getKeyId() {
            return keyId;
        }
    }

    private String accessKey = null;
    private String secretKey = null;
    private String sessionToken = null;
    private Region region = null;
    private String bucket = null;
    private String path = null;

    private String serviceEndpoint = null;
    private String signingRegion = null;
    private boolean pathStyleAccess = false;

    private boolean compressionEnabled = false;
    private boolean keyGzSuffixEnabled = false;
    private S3SSEConfiguration sseConfiguration = null;

    private CannedAccessControlList cannedAcl = null;

    public String getAccessKey() {
        return accessKey;
    }
    public void setAccessKey(String accessKey) {
        this.accessKey = accessKey;
    }
    public String getSecretKey() {
        return secretKey;
    }
    public void setSecretKey(String secretKey) {
        this.secretKey = secretKey;
    }
    public String getSessionToken() { return sessionToken; }
    public void setSessionToken(String sessionToken) { this.sessionToken = sessionToken; }

    public Region getRegion() {
        return region;
    }

    public S3SSEConfiguration getSseConfiguration() {
        return sseConfiguration;
    }

    public void setSseConfiguration(S3SSEConfiguration sseConfiguration) {
        this.sseConfiguration = sseConfiguration;
    }

    public CannedAccessControlList getCannedAcl() {
        return cannedAcl;
    }

    public boolean isPathStyleAccess() {
        return pathStyleAccess;
    }
    /**
     * Sets the canned ACL for S3 to use when storing objects.
     *
     * @param cannedAclValue - the String value of the ACL to set (values for the CannedAccessControlList enum)
     *
     * @throws IllegalArgumentException - if the cannedAclValue cannot be mapped to a valid CannedAccessControlList enum
     */
    public void setCannedAclFromValue(String cannedAclValue) throws IllegalArgumentException {
        this.cannedAcl = CannedAccessControlList.valueOf(cannedAclValue);
    }

    /**
     * Sets the region to use when building the S3 client. The value can be the
     * "lowercase dash" format used in AWS literature (e.g. "us-west-2", "eu-west-1")
     * or the Regions ordinal name (e.g. "US_WEST_2" or "EU_WEST_2").
     * <br>
     * If the value is <code>null</code>, then no region will be explicitly set.
     *
     * @param regionName the region name (e.g. "us-west-2", "eu-west-1")
     */
    public void setRegion(String regionName) {
        this.region = resolveRegion(regionName);
    }
    public String getBucket() {
        return bucket;
    }
    public void setBucket(String bucket) {
        this.bucket = bucket;
    }
    public String getPath() {
        return path;
    }
    public void setPath(String path) {
        this.path = path;
    }

    public String getServiceEndpoint() {
        return serviceEndpoint;
    }
    public void setServiceEndpoint(String serviceEndpoint) {
        this.serviceEndpoint = serviceEndpoint;
    }
    public String getSigningRegion() {
        return signingRegion;
    }
    public void setSigningRegion(String signingRegion) {
        this.signingRegion = signingRegion;
    }
    public void setPathStyleAccess(boolean pathStyleAccess) {
        this.pathStyleAccess = pathStyleAccess;
    }

    public boolean isCompressionEnabled() {
        return compressionEnabled;
    }
    public boolean isKeyGzSuffixEnabled() {
        return keyGzSuffixEnabled;
    }

    public void setCompressionEnabled(boolean compressionEnabled) {
        this.compressionEnabled = compressionEnabled;
    }
    public void setKeyGzSuffixEnabled(boolean gzSuffix) {
        this.keyGzSuffixEnabled = gzSuffix;
    }

    /**
     * Best-effort to map a region name to an actual Region instance. The input
     * string can be either the public region name (e.g. "us-west-1") or the
     * Regions enum ordinal name (e.g. "US_WEST_1").
     *
     * @param str the region name to map to a Region
     *
     * @return the mapped Region
     *
     * @throws IllegalArgumentException if the input name cannot be mapped to a
     *  Region.
     */
    static Region resolveRegion(String str) {
        Region region = null;
        if (StringUtils.isTruthy(str)) {
            Regions regions;
            try {
                regions = Regions.valueOf(str);
            } catch (IllegalArgumentException ex) {
                // Try to interpret as the public name (this is more expensive). If
                // this fails, then just propagate the IllegalArgumentException out.
                regions = Regions.fromName(str);
            }
            region = Region.getRegion(regions);
        }
        return region;
    }

    public String toString() {
        if (StringUtils.isTruthy(this.path)) {
            return String.format("S3 configuration (%s:%s in region %s; compressed: %s)",
                    this.bucket, this.path, this.region, this.compressionEnabled
            );
        } else {
            return String.format("S3 configuration (%s in region %s; compressed: %s)",
                    this.bucket, this.region, this.compressionEnabled
            );
        }
    }
} public class S3PublishHelper extends AbstractFilePublishHelper {

    private final AmazonS3Client client;
    private final String bucket;
    private final String path;
    private final S3Configuration.S3SSEConfiguration sseConfig;
    private final CannedAccessControlList cannedAcl;
    private final IStorageDestinationAdjuster storageDestinationAdjuster;
    private final boolean isCompressionEnabled;
    private final boolean keyGzSuffix;

    private volatile boolean bucketExists = false;

    public S3PublishHelper(
        S3Configuration s3,
        IStorageDestinationAdjuster storageDestinationAdjuster,
        boolean verbose
    ) {
        super(s3.isCompressionEnabled(), verbose);
        this.client = (AmazonS3Client)buildClient(
            s3.getAccessKey(), s3.getSecretKey(), s3.getSessionToken(),
            s3.getRegion(),
            s3.getServiceEndpoint(), s3.getSigningRegion(),
            s3.isPathStyleAccess()
        );
        this.bucket = s3.getBucket().toLowerCase();
        this.storageDestinationAdjuster = storageDestinationAdjuster;
        this.path = s3.getPath();
        this.sseConfig = s3.getSseConfiguration();
        this.cannedAcl = s3.getCannedAcl();
        this.isCompressionEnabled = s3.isCompressionEnabled();
        this.keyGzSuffix = s3.isKeyGzSuffixEnabled();
    }

    @Override
    public void start(PublishContext context) {
        super.start(context);
        try {
            if (!bucketExists) {
                bucketExists = client.doesBucketExist(bucket);
                if (!bucketExists) {
                    client.createBucket(bucket);
                    bucketExists = true;
                }
            }
        } catch (Exception ex) {
            throw new RuntimeException(String.format("Cannot start publishing: %s", ex.getMessage()), ex);
        }
    }

    static String getKey(PublishContext context, String path, boolean isCompressionEnabled, boolean keyGzSuffix) {
        String key = String.format("%s%s", path, context.getCacheName());
        if (isCompressionEnabled && keyGzSuffix) {
            key = String.format("%s.gz", key);
        }
        return key;
    }

    @Override
    protected void publishFile(File file, PublishContext context) {
        String path = StringUtils.addTrailingIfNeeded(
            PublishHelperUtils.adjustStoragePathIfNecessary(
                this.path,
                storageDestinationAdjuster
            ),
            "/"
        );
        String key = getKey(context, path, isCompressionEnabled, keyGzSuffix);
        if (this.verbose) {
            System.out.println(String.format("Publishing to S3 (bucket=%s; key=%s):", bucket, key));
        }

        try {
            // System.out.println(String.format("Publishing content of %s to S3.", tempFile));
            ObjectMetadata metadata = new ObjectMetadata();
            metadata.setContentLength(file.length());
            metadata.setContentType(ContentType.DEFAULT_BINARY.getMimeType());
            // Currently, only SSE_S3 is supported
            if ((sseConfig != null) && (sseConfig.getKeyType() == S3Configuration.SSEType.SSE_S3)) {
                // https://docs.aws.amazon.com/AmazonS3/latest/dev/SSEUsingJavaSDK.html
                metadata.setSSEAlgorithm(ObjectMetadata.AES_256_SERVER_SIDE_ENCRYPTION);
            }

            PutObjectRequest por;
            if (cannedAcl != null)
                por = new PutObjectRequest(bucket, key, file).withCannedAcl(cannedAcl);
            else
                por = new PutObjectRequest(bucket, key, file);
            por.setMetadata(metadata);

            PutObjectResult result = client.putObject(por);
            // System.out.println(String.format("Content MD5: %s", result.getContentMd5()));
        } catch (Exception ex) {
            throw new RuntimeException(String.format("Cannot publish to S3: %s", ex.getMessage()), ex);
        }
    }
} public class BlobConfiguration {
    public static final String DEFAULT_LOG_CONTAINER_PATH = "logs/";

    private String storageConnectionString = "";
    private String containerName = "";
    private String blobNamePrefix = DEFAULT_LOG_CONTAINER_PATH;
    private boolean compressionEnabled = false;
    private boolean blobNameGzSuffixEnabled = false;

    public void setStorageConnectionString(String storageConnectionString) {
        this.storageConnectionString = storageConnectionString;
    }

    public void setBlobNameGzSuffixEnabled(boolean gzSuffix) {
        this.blobNameGzSuffixEnabled = gzSuffix;
    }

    public void setContainerName(String containerName) {
        this.containerName = containerName;
    }

    public void setBlobNamePrefix(String blobNamePrefix) {
        this.blobNamePrefix = blobNamePrefix;
    }

    public void setCompressionEnabled(boolean compressionEnabled) {
        this.compressionEnabled = compressionEnabled;
    }

    public String getStorageConnectionString() {
        return storageConnectionString;
    }

    public String getContainerName() {
        return containerName;
    }

    public String getBlobNamePrefix() {
        return blobNamePrefix;
    }

    public boolean isCompressionEnabled() {
        return compressionEnabled;
    }

    public boolean isBlobNameGzSuffixEnabled() {
        return blobNameGzSuffixEnabled;
    }

    public String toString() {
        return String.format("Azure blob config (%s:%s, compressed: %s)",
            this.containerName, this.blobNamePrefix, this.compressionEnabled);
    }
} public class BlobPublishHelper extends AbstractFilePublishHelper {

    private final BlobConfiguration blobConfiguration;
    private final IStorageDestinationAdjuster storageDestinationAdjuster;

    public BlobPublishHelper(
        BlobConfiguration blobConfiguration,
        IStorageDestinationAdjuster storageDestinationAdjuster,
        boolean verbose
    ) {
        super(blobConfiguration.isCompressionEnabled(), verbose);
        this.blobConfiguration = blobConfiguration;
        this.storageDestinationAdjuster = storageDestinationAdjuster;
    }

    static String getBlobName(PublishContext context, String path, BlobConfiguration blobConfiguration) {
        String blobName = String.format("%s%s", path, context.getCacheName());
        if (blobConfiguration.isCompressionEnabled() && blobConfiguration.isBlobNameGzSuffixEnabled()) {
            blobName = String.format("%s.gz", blobName);
        }
        return blobName;
    }

    @Override
    protected void publishFile(File file, PublishContext context) throws Exception {
        String connectionString = blobConfiguration.getStorageConnectionString();

        if (null == connectionString || connectionString.isEmpty()) {
            connectionString = System.getenv("AZURE_STORAGE_CONNECTION_STRING");
        }
        CloudStorageAccount storageAccount = CloudStorageAccount.parse(connectionString);
        CloudBlobClient blobClient = storageAccount.createCloudBlobClient();
        CloudBlobContainer container = blobClient.getContainerReference(blobConfiguration.getContainerName());
        boolean created = container.createIfNotExists(
            BlobContainerPublicAccessType.CONTAINER, new BlobRequestOptions(), new OperationContext());

        String path = StringUtils.addTrailingIfNeeded(
            PublishHelperUtils.adjustStoragePathIfNecessary(
                blobConfiguration.getBlobNamePrefix(),
                storageDestinationAdjuster
            ),
            "/"
        );
        String blobName = getBlobName(context, path, blobConfiguration);
        if (this.verbose) {
            System.out.println(String.format("Publishing %s to Azure blob (container=%s; blob=%s):",
                file.getAbsolutePath(), blobConfiguration.getContainerName(), blobName));
        }

        CloudBlockBlob blob = container.getBlockBlobReference(blobName);
        if (this.blobConfiguration.isCompressionEnabled()) {
            blob.getProperties().setContentType("application/gzip");
        } else {
            blob.getProperties().setContentType("text/plain");
        }
        blob.uploadFromFile(file.getAbsolutePath());
        if (this.verbose) {
            System.out.println(String.format("Publishing to Azure blob %s done.", blobName));
        }
    }
} public class ElasticsearchConfiguration {
    public interface IHostConsumer {
        void consume(String host, int port);
    }

    private static final String DEFAULT_CLUSTERNAME = "elasticsearch";
    private static final String DEFAULT_INDEX = "logindex";

    // Elasticsearch is desperately trying to get rid of type, but this is here just in case.
    private static final String DEFAULT_TYPE = "_doc";


    private String clusterName = DEFAULT_CLUSTERNAME;
    private String index = DEFAULT_INDEX;
    private String type = DEFAULT_TYPE;

    private final List<String> hosts = new LinkedList<>();




    public ElasticsearchConfiguration() {
    }

    public void setClusterName(String clusterName) {
        this.clusterName = clusterName;
    }

    public String getClusterName() {
        return clusterName;
    }

    public String getIndex() {
        return index;
    }

    public void setIndex(String index) {
        this.index = index;
    }

    public String getType() {
        return type;
    }

    public void setType(String type) {
        this.type = type;
    }

    /**
     * Adds a host to the list of hosts to use to contact ElasticSearch.
     *
     * @param host the host and port (e.g. "host1:9300" or "host1")
     */
    public void addHost(String host) {
        hosts.add(host);
    }

    /**
     * Iterate through all the host:port entries and call the supplied consumer with each.
     *
     * @param consumer a consumer of host and port for an Elasticsearch host entry
     */
    public void iterateHosts(IHostConsumer consumer) {
        for (String host: hosts) {
            String parts[] = host.split(":");
            int port = 9300;
            String hostName = parts[0].trim();
            try {
                port = Integer.parseInt(parts[1]);
            } catch (Exception ex) {
            }
            consumer.consume(hostName, port);
        }
    }

    @Override
    public String toString() {
        return String.format(
            "ElasticsearchConfiguration: cluster=%s, hosts=%s, index=%s, type=%s",
            clusterName, hosts.toString(), index, type);
    }
} public class ElasticsearchPublishHelper implements IElasticsearchPublishHelper {

    private ElasticsearchConfiguration configuration;

    private final List<Node> nodes = new ArrayList<>();
    private RestHighLevelClient client;
    private BulkRequest bulkRequest;
    private int offset;
    private Date timeStamp;

    public ElasticsearchPublishHelper() {
    }

    @Override
    public void initialize(ElasticsearchConfiguration configuration) {
        this.configuration = configuration;
        this.configuration.iterateHosts((host, port) -> {
            nodes.add(createNodeFromHost(host, port));
        });
    }

    Node createNodeFromHost(String host, int port) {
        String scheme = null;
        String hostName = host.toLowerCase().trim();
        if (hostName.startsWith("http:") || hostName.startsWith("https:")) {
            String[] schemeAndHostname = hostName.split(":");
            if (schemeAndHostname.length >= 2) {
                scheme = schemeAndHostname[0];
                hostName = schemeAndHostname[1];
            }
        }
        return new Node(new HttpHost(hostName, port, scheme));
    }

    @Override
    public void start(PublishContext context) {
        offset = 0;
        timeStamp = new Date();

        // The way RestClient.builder(Node... hosts) is designed requires hard-coding the list of hosts (great
        // for demo programs but not for anything more serious).
        // Because our host list is materialized from configuration as an array/list, we have to actually call
        // RestClient.builder(Node... hosts) with an array via inflection. All this BS would not be necessary
        // if they had just added a RestClient.builder(Node[] nodes) or RestClient.builder(Collection<Node> nodes).
        try {
            Method restClientBuilderMethod = RestClient.class.getDeclaredMethod("builder", Node[].class);
            restClientBuilderMethod.setAccessible(true);
            RestClientBuilder builder =
                (RestClientBuilder)restClientBuilderMethod.invoke(null, new Object[]{nodes.toArray(new Node[0])});
            client = new RestHighLevelClient(builder);
        } catch (Exception e) {
            e.printStackTrace();
        }
        bulkRequest = new BulkRequest();
    }

    @Override
    public void publish(PublishContext context, int sequence, Event event) {
        try {
            String id = String.format("%s-%s-%016d", context.getCacheName(),
                context.getHostName(), offset);
            XContentBuilder contentBuilder = jsonBuilder()
                .startObject()
                .field("timestamp", timeStamp)
                .field("type", event.getType())
                .field("hostname", context.getHostName())
                .field("offset", offset)
                .field("thread_name", event.getThreadName())
                .field("logger", event.getSource())
                .field("message", event.getMessage())
                .array("tags", context.getTags())
                .endObject();
            bulkRequest.add(new IndexRequest(configuration.getIndex()).id(id).source(contentBuilder));
            offset++;
        } catch (Exception ex) {
            System.err.printf("Cannot publish event: %s%n", ex.getMessage());
        }
    }

    @Override
    public void end(PublishContext context) {
        try {
            if ((null != client) && (null != bulkRequest)) {
                BulkResponse response = client.bulk(bulkRequest, RequestOptions.DEFAULT);
                if (response.hasFailures()) {
                    System.err.println("Elasticsearch publish failures: " + response.buildFailureMessage());
                }
            }
        } catch (IOException ex) {
            ex.printStackTrace();
        } finally {
            try {
                if (null != client) {
                    client.close();
                }
                bulkRequest = null;
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static IElasticsearchPublishHelper getPublishHelper(
        String publishHelperNameSpec, ClassLoader classLoader) {
        return getPublishHelper(publishHelperNameSpec, classLoader, false);
    }

    public static IElasticsearchPublishHelper getPublishHelper(
        String publishHelperNameSpec, ClassLoader classLoader, boolean verbose) {
        ElasticsearchPublishHelper helper = null;
        if (StringUtils.isTruthy(publishHelperNameSpec)) {
            System.out.printf("Attempting to instantiate %s%n", publishHelperNameSpec);
            try {
                Class<ElasticsearchPublishHelper> cls;
                cls = (Class<ElasticsearchPublishHelper>)classLoader.loadClass(
                    publishHelperNameSpec
                );
                Constructor<ElasticsearchPublishHelper> ctor = cls.getConstructor();
                helper = ctor.newInstance();
                if (verbose) {
                    System.out.printf(
                        "Successfully registered %s as Elasticsearch publish helper%n", publishHelperNameSpec
                    );
                }
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }

        if (null == helper) {
            if (verbose) {
                System.out.printf("Instantiating the default ElasticsearchPublishHelper%n");
            }
            helper = new ElasticsearchPublishHelper();
        }
        return helper;
    }
} public interface IElasticsearchPublishHelper extends IPublishHelper<Event> {

    /**
     * Initialize the publish helper immediately after construction
     *
     * @param configuration configuration information to initialize the publish helper with
     */
    public void initialize(ElasticsearchConfiguration configuration);
} public class CloudStorageConfiguration {
    public static final String DEFAULT_BLOB_PATH = "logs/";

    private String bucketName;
    private String blobNamePrefix = DEFAULT_BLOB_PATH;
    private boolean compressionEnabled = false;
    private boolean blobNameGzSuffixEnabled = false;

    public String getBucketName() {
        return bucketName;
    }

    public void setBucketName(String bucketName) {
        this.bucketName = bucketName;
    }

    public String getBlobNamePrefix() {
        return blobNamePrefix;
    }

    public void setBlobNamePrefix(String blobNamePrefix) {
        this.blobNamePrefix = blobNamePrefix;
    }

    public boolean isCompressionEnabled() {
        return compressionEnabled;
    }

    public void setCompressionEnabled(boolean compressionEnabled) {
        this.compressionEnabled = compressionEnabled;
    }

    public boolean isBlobNameGzSuffixEnabled() {
        return blobNameGzSuffixEnabled;
    }

    public void setBlobNameGzSuffixEnabled(boolean blobNameGzSuffixEnabled) {
        this.blobNameGzSuffixEnabled = blobNameGzSuffixEnabled;
    }
} public class CloudStoragePublishHelper extends AbstractFilePublishHelper {
    private final CloudStorageConfiguration configuration;
    private final IStorageDestinationAdjuster storageDestinationAdjuster;

    public CloudStoragePublishHelper(
        CloudStorageConfiguration configuration,
        IStorageDestinationAdjuster storageDestinationAdjuster,
        boolean verbose
    ) {
        super(configuration.isCompressionEnabled(), verbose);
        this.configuration = configuration;
        this.storageDestinationAdjuster = storageDestinationAdjuster;
    }

    static String getBlobName(PublishContext context, String path, CloudStorageConfiguration configuration) {
        String blobName = String.format("%s%s", path, context.getCacheName());
        if (configuration.isCompressionEnabled() && configuration.isBlobNameGzSuffixEnabled()) {
            blobName = String.format("%s.gz", blobName);
        }
        return blobName;
    }

    @Override
    protected void publishFile(File file, PublishContext context) throws Exception {
        String bucketName = configuration.getBucketName();
        Storage storage = StorageOptions.getDefaultInstance().getService();
        Bucket bucket = storage.get(bucketName);
        if (null == bucket) {
            bucket = storage.create(BucketInfo.of(bucketName));
        }
        String path = StringUtils.addTrailingIfNeeded(
            PublishHelperUtils.adjustStoragePathIfNecessary(
                configuration.getBlobNamePrefix(),
                storageDestinationAdjuster
            ),
            "/"
        );
        String blobName = getBlobName(context, path, configuration);
        if (this.verbose) {
            System.out.println(String.format("Publishing %s to GCS blob (bucket=%s; blob=%s):",
                file.getAbsolutePath(), bucketName, blobName));
        }
        BlobId blobId = BlobId.of(bucketName, blobName);
        String contentType = configuration.isCompressionEnabled() ? "application/gzip" : "text/plain";
        BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setContentType(contentType).build();
        uploadToStorage(storage, file, blobInfo);
    }

    private void uploadToStorage(Storage storage, File uploadFrom, BlobInfo blobInfo) throws IOException {
        // From: https://stackoverflow.com/questions/53628684/how-to-upload-a-large-file-into-gcp-cloud-storage

        if (uploadFrom.length() < 1_000_000) {
            storage.create(blobInfo, Files.readAllBytes(uploadFrom.toPath()));
        } else {
            // For big files:
            // When content is not available or large (1MB or more) it is recommended to write it in chunks via
            // the blob's channel writer.
            try (WriteChannel writer = storage.writer(blobInfo)) {
                byte[] buffer = new byte[10_240];
                try (InputStream input = Files.newInputStream(uploadFrom.toPath())) {
                    int limit;
                    while ((limit = input.read(buffer)) >= 0) {
                        writer.write(ByteBuffer.wrap(buffer, 0, limit));
                    }
                }

            }
        }
    }
} public class SolrConfiguration {
    private URL url;

    public URL getUrl() {
        return url;
    }

    public void setUrl(String solrUrl) {
        if (StringUtils.isTruthy(solrUrl)) {
            try {
                this.url = new URL(solrUrl);
            } catch (MalformedURLException e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public String toString() {
        return String.format("SolrConfiguration: %s", url);
    }
} public class SolrPublishHelper implements IPublishHelper<Event> {
    private final SolrClient client;

    private int offset;
    private Date timeStamp;
    private List<SolrInputDocument> docs;

    public SolrPublishHelper(URL collectionUrl) {
        this.client = new HttpSolrClient.Builder()
            .withBaseSolrUrl(collectionUrl.toExternalForm())
            .build();
    }

    public void publish(PublishContext context, int sequence, Event event) {
        SolrInputDocument doc = new SolrInputDocument();
        doc.addField("id", String.format("%s-%s-%016d", context.getCacheName(),
            context.getHostName(), offset));
        doc.addField("timestamp", timeStamp);
        doc.addField("type", event.getType());
        doc.addField("hostname", context.getHostName());
        doc.addField("offset", offset);
        doc.addField("thread_name", event.getThreadName());
        doc.addField("logger", event.getSource());
        doc.addField("message", event.getMessage());
        if (context.getTags().length > 0) {
            for (String tag: context.getTags()) {
                doc.addField("tags", tag);
            }
        }
        docs.add(doc);
        offset++;
    }

    public void start(PublishContext context) {
        offset = 0;
        timeStamp = new Date();
        docs = new LinkedList<>();
    }

    public void end(PublishContext context) {
        try {
            client.add(docs);
            client.commit();
        } catch (Exception e) {
            throw new RuntimeException("Cannot commit batch", e);
        }
    }

} public class StringUtils {

    /**
     * Checks to see if a String is defined and not empty after trimming
     *
     * @param s the String to test
     *
     * @return true if the string is not null and is not empty
     */
    public static boolean isTruthy(String s) {
        return (null != s && !s.trim().isEmpty());
    }

    /**
     * If the string passed in is not blank or null and did not end with the trailing value, then the trailing value
     * will be appended to it in the returned value.
     *
     * @param s the string to conditionally add the trailing value to. If null, then null will be returned.
     * @param trailing the trailing value to append. If null, then s will be returned unmodified.
     *
     * @return the string s with the trailing value appended as necessary.
     */
    public static String addTrailingIfNeeded(String s, String trailing) {
        String value = s;
        if ((null != s) && (null != trailing)) {
            if (isTruthy(s) && !s.endsWith(trailing)) {
                value = s + trailing;
            }
        }
        return value;
    }
}
import com.van.logging.*;
import com.van.logging.aws.S3Configuration;
import com.van.logging.aws.S3PublishHelper;
import com.van.logging.azure.BlobConfiguration;
import com.van.logging.azure.BlobPublishHelper;
import com.van.logging.elasticsearch.ElasticsearchConfiguration;
import com.van.logging.elasticsearch.ElasticsearchPublishHelper;
import com.van.logging.elasticsearch.IElasticsearchPublishHelper;
import com.van.logging.gcp.CloudStorageConfiguration;
import com.van.logging.gcp.CloudStoragePublishHelper;
import com.van.logging.solr.SolrConfiguration;
import com.van.logging.solr.SolrPublishHelper;
import com.van.logging.utils.StringUtils;
import org.apache.logging.log4j.core.LogEvent;
import org.apache.logging.log4j.core.config.plugins.PluginBuilderAttribute;
import org.apache.logging.log4j.core.filter.AbstractFilter;
import java.net.UnknownHostException;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;
    @PluginBuilderAttribute
    private boolean s3PathStyleAccess;

    // Azure blob properties
    @PluginBuilderAttribute
    private String azureStorageConnectionString;

    @PluginBuilderAttribute
    private String azureBlobContainer;

    @PluginBuilderAttribute
    private String azureBlobNamePrefix;

    @PluginBuilderAttribute
    private String azureBlobCompressionEnabled;

    @PluginBuilderAttribute
    private String azureBlobNameGzSuffixEnabled;

    // GCP Storage properties
    @PluginBuilderAttribute
    private String gcpStorageBucket;

    @PluginBuilderAttribute
    private String gcpStorageBlobNamePrefix;

    @PluginBuilderAttribute
    private String gcpStorageCompressionEnabled;

    @PluginBuilderAttribute
    private String gcpStorageBlobNameGzSuffixEnabled;

    // Solr properties
    @PluginBuilderAttribute
    private String solrUrl;

    // Elasticsearch properties
    @PluginBuilderAttribute
    private String elasticsearchCluster;

    @PluginBuilderAttribute
    private String elasticsearchIndex;

    @PluginBuilderAttribute
    private String elasticsearchType;

    @PluginBuilderAttribute
    private String elasticsearchHosts;

    @PluginBuilderAttribute
    private String elasticSearchPublishHelperClass;


    @Override
    public Log4j2Appender build() {
        try {
            String cacheName = UUID.randomUUID().toString().replaceAll("-","");
            LoggingEventCache<Event> cache = new LoggingEventCache<>(
                cacheName, createCacheMonitor(), createCachePublisher(), verbose);
            Log4j2Appender appender = new Log4j2Appender(
                getName(), getFilter(), getLayout(),
                true, cache);
            appender.setVerbose(verbose);
            return installFilter(appender);
        } catch (Exception e) {
            throw new RuntimeException("Cannot build appender due to errors", e);
        }
    }

    Log4j2Appender installFilter(Log4j2Appender appender) {
        appender.addFilter(new AbstractFilter() {
            @Override
            public Result filter(final LogEvent event) {
                // To prevent infinite looping, we filter out events from
                // the publishing thread
                Result decision = Result.NEUTRAL;
                if (LoggingEventCache.PUBLISH_THREAD_NAME.equals(event.getThreadName())) {
                    decision = Result.DENY;
                }
                return decision;
            }});
        return appender;
    }

    Optional<S3Configuration> getS3ConfigIfEnabled() {
        Optional<S3Configuration> s3Config = Optional.empty();
        if (StringUtils.isTruthy(s3Bucket)) {
            S3Configuration config = new S3Configuration();
            config.setBucket(s3Bucket);
            config.setPath(s3Path);
            config.setRegion(s3Region);
            config.setAccessKey(s3AwsKey);
            config.setSecretKey(s3AwsSecret);
            config.setSessionToken(s3AwsSessionToken);
            config.setServiceEndpoint(s3ServiceEndpoint);
            config.setSigningRegion(s3SigningRegion);
            config.setPathStyleAccess(s3PathStyleAccess);
            config.setCompressionEnabled(Boolean.parseBoolean(s3Compression));
            config.setKeyGzSuffixEnabled(Boolean.parseBoolean(s3KeyGzSuffixEnabled));
            if (StringUtils.isTruthy(s3CannedAcl)) {
                try {
                    config.setCannedAclFromValue(s3CannedAcl);
                } catch (IllegalArgumentException ex) {
                    System.err.println(String.format("Ignoring unrecognized canned ACL value %s", s3CannedAcl));
                }
            }

            S3Configuration.S3SSEConfiguration sseConfig = null;
            if (s3SseKeyType != null) {
                sseConfig = new S3Configuration.S3SSEConfiguration(
                        S3Configuration.SSEType.valueOf(s3SseKeyType),
                        null
                );
            }
            config.setSseConfiguration(sseConfig);
            s3Config = Optional.of(config);
        }
        return s3Config;
    }

    Optional<BlobConfiguration> getBlobConfigurationIfEnabled() {