/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to you 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 org.apache.logging.log4j.cassandra;

import com.datastax.driver.core.BatchStatement;
import org.apache.logging.log4j.core.Core;
import org.apache.logging.log4j.core.Filter;
import org.apache.logging.log4j.core.appender.AbstractAppender;
import org.apache.logging.log4j.core.appender.db.AbstractDatabaseAppender;
import org.apache.logging.log4j.core.appender.db.ColumnMapping;
import org.apache.logging.log4j.core.config.Property;
import org.apache.logging.log4j.core.config.plugins.Plugin;
import org.apache.logging.log4j.core.config.plugins.PluginBuilderAttribute;
import org.apache.logging.log4j.core.config.plugins.PluginBuilderFactory;
import org.apache.logging.log4j.core.config.plugins.PluginElement;
import org.apache.logging.log4j.core.config.plugins.validation.constraints.Required;
import org.apache.logging.log4j.core.net.SocketAddress;
import org.apache.logging.log4j.core.util.Clock;

/**
 * Appender plugin that uses a Cassandra database.
 *
 * @see SocketAddress
 * @see ColumnMapping
 */
@Plugin(
        name = "Cassandra",
        category = Core.CATEGORY_NAME,
        elementType = CassandraAppender.ELEMENT_TYPE,
        printObject = true)
public class CassandraAppender extends AbstractDatabaseAppender<CassandraManager> {

    private CassandraAppender(
            final String name,
            final Filter filter,
            final boolean ignoreExceptions,
            final Property[] properties,
            final CassandraManager manager) {
        super(name, filter, null, ignoreExceptions, properties, manager);
    }

    @PluginBuilderFactory
    public static <B extends Builder<B>> B newBuilder() {
        return new Builder<B>().asBuilder();
    }

    public static class Builder<B extends Builder<B>> extends AbstractAppender.Builder<B>
            implements org.apache.logging.log4j.core.util.Builder<CassandraAppender> {

        /**
         * List of Cassandra node contact points. Addresses without a port (or port set to 0) will use the default
         * Cassandra port (9042).
         */
        @PluginElement("ContactPoints")
        @Required(message = "No Cassandra servers provided")
        private SocketAddress[] contactPoints = new SocketAddress[] {SocketAddress.getLoopback()};

        /**
         * List of column mappings to convert a LogEvent into a database row.
         */
        @PluginElement("Columns")
        @Required(message = "No Cassandra columns provided")
        private ColumnMapping[] columns;

        @PluginBuilderAttribute
        private boolean useTls;

        @PluginBuilderAttribute
        @Required(message = "No cluster name provided")
        private String clusterName;

        @PluginBuilderAttribute
        @Required(message = "No keyspace provided")
        private String keyspace;

        @PluginBuilderAttribute
        @Required(message = "No table name provided")
        private String table;

        @PluginBuilderAttribute
        private String username;

        @PluginBuilderAttribute(sensitive = true)
        private String password;

        /**
         * Override the default TimestampGenerator with one based on the configured {@link Clock}.
         */
        @PluginBuilderAttribute
        private boolean useClockForTimestampGenerator;

        /**
         * Number of LogEvents to buffer before writing. Can be used with or without batch statements.
         */
        @PluginBuilderAttribute
        private int bufferSize;

        /**
         * Whether or not to use batch statements when inserting records.
         */
        @PluginBuilderAttribute
        private boolean batched;

        /**
         * If batch statements are enabled, use this type of batch statement.
         */
        @PluginBuilderAttribute
        private BatchStatement.Type batchType = BatchStatement.Type.LOGGED;

        public B setContactPoints(final SocketAddress... contactPoints) {
            this.contactPoints = contactPoints;
            return asBuilder();
        }

        public B setColumns(final ColumnMapping... columns) {
            this.columns = columns;
            return asBuilder();
        }

        public B setUseTls(final boolean useTls) {
            this.useTls = useTls;
            return asBuilder();
        }

        public B setClusterName(final String clusterName) {
            this.clusterName = clusterName;
            return asBuilder();
        }

        public B setKeyspace(final String keyspace) {
            this.keyspace = keyspace;
            return asBuilder();
        }

        public B setTable(final String table) {
            this.table = table;
            return asBuilder();
        }

        public B setUsername(final String username) {
            this.username = username;
            return asBuilder();
        }

        public B setPassword(final String password) {
            this.password = password;
            return asBuilder();
        }

        public B setUseClockForTimestampGenerator(final boolean useClockForTimestampGenerator) {
            this.useClockForTimestampGenerator = useClockForTimestampGenerator;
            return asBuilder();
        }

        public B setBufferSize(final int bufferSize) {
            this.bufferSize = bufferSize;
            return asBuilder();
        }

        public B setBatched(final boolean batched) {
            this.batched = batched;
            return asBuilder();
        }

        public B setBatchType(final BatchStatement.Type batchType) {
            this.batchType = batchType;
            return asBuilder();
        }

        @Override
        public CassandraAppender build() {
            final CassandraManager manager = CassandraManager.getManager(
                    getName(),
                    contactPoints,
                    columns,
                    useTls,
                    clusterName,
                    keyspace,
                    table,
                    username,
                    password,
                    useClockForTimestampGenerator,
                    bufferSize,
                    batched,
                    batchType);
            return new CassandraAppender(getName(), getFilter(), isIgnoreExceptions(), null, manager);
        }
    }
}
