/**
 * 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.pulsar.client.api;

import java.time.Clock;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException;

/**
 * Builder interface that is used to configure and construct a {@link PulsarClient} instance.
 *
 * @since 2.0.0
 */
public interface ClientBuilder extends Cloneable {

    /**
     * Construct the final {@link PulsarClient} instance.
     *
     * @return the new {@link PulsarClient} instance
     */
    PulsarClient build() throws PulsarClientException;

    /**
     * Load the configuration from provided <tt>config</tt> map.
     *
     * <p>Example:
     *
     * <pre>
     * {@code
     * Map<String, Object> config = new HashMap<>();
     * config.put("serviceUrl", "pulsar://localhost:6650");
     * config.put("numIoThreads", 20);
     *
     * ClientBuilder builder = ...;
     * builder = builder.loadConf(config);
     *
     * PulsarClient client = builder.build();
     * }
     * </pre>
     *
     * @param config
     *            configuration to load
     * @return the client builder instance
     */
    ClientBuilder loadConf(Map<String, Object> config);

    /**
     * Create a copy of the current client builder.
     *
     * <p>Cloning the builder can be used to share an incomplete configuration and specialize it multiple times. For
     * example:
     *
     * <pre>{@code
     * ClientBuilder builder = PulsarClient.builder()
     *               .ioThreads(8)
     *               .listenerThreads(4);
     *
     * PulsarClient client1 = builder.clone()
     *                  .serviceUrl("pulsar://localhost:6650").build();
     * PulsarClient client2 = builder.clone()
     *                  .serviceUrl("pulsar://other-host:6650").build();
     * }</pre>
     *
     * @return a clone of the client builder instance
     */
    ClientBuilder clone();

    /**
     * Configure the service URL for the Pulsar service.
     *
     * <p>This parameter is required.
     *
     * <p>Examples:
     * <ul>
     * <li>{@code pulsar://my-broker:6650} for regular endpoint</li>
     * <li>{@code pulsar+ssl://my-broker:6651} for TLS encrypted endpoint</li>
     * </ul>
     *
     * @param serviceUrl
     *            the URL of the Pulsar service that the client should connect to
     * @return the client builder instance
     */
    ClientBuilder serviceUrl(String serviceUrl);

    /**
     * Configure the service URL provider for Pulsar service.
     *
     * <p>Instead of specifying a static service URL string (with {@link #serviceUrl(String)}), an application
     * can pass a {@link ServiceUrlProvider} instance that dynamically provide a service URL.
     *
     * @param serviceUrlProvider
     *            the provider instance
     * @return the client builder instance
     */
    ClientBuilder serviceUrlProvider(ServiceUrlProvider serviceUrlProvider);

    /**
     * Set the authentication provider to use in the Pulsar client instance.
     *
     * <p>Example:
     * <pre>{@code
     * PulsarClient client = PulsarClient.builder()
     *         .serviceUrl("pulsar+ssl://broker.example.com:6651/")
     *         .authentication(
     *               AuthenticationFactory.TLS("/my/cert/file", "/my/key/file")
     *         .build();
     * }</pre>
     *
     * <p>For token based authentication, this will look like:
     * <pre>{@code
     * AuthenticationFactory
     *      .token("eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJKb2UifQ.ipevRNuRP6HflG8cFKnmUPtypruRC4fb1DWtoLL62SY")
     * }</pre>
     *
     * @param authentication
     *            an instance of the {@link Authentication} provider already constructed
     * @return the client builder instance
     */
    ClientBuilder authentication(Authentication authentication);

    /**
     * Configure the authentication provider to use in the Pulsar client instance.
     *
     * <p>Example:
     * <pre>
     * <code>
     * PulsarClient client = PulsarClient.builder()
     *          .serviceUrl("pulsar+ssl://broker.example.com:6651/)
     *          .authentication(
     *              "org.apache.pulsar.client.impl.auth.AuthenticationTls",
     *              "tlsCertFile:/my/cert/file,tlsKeyFile:/my/key/file")
     *          .build();
     * </code>
     * </pre>
     *
     * @param authPluginClassName
     *            name of the Authentication-Plugin you want to use
     * @param authParamsString
     *            string which represents parameters for the Authentication-Plugin, e.g., "key1:val1,key2:val2"
     * @return the client builder instance
     * @throws UnsupportedAuthenticationException
     *             failed to instantiate specified Authentication-Plugin
     */
    ClientBuilder authentication(String authPluginClassName, String authParamsString)
            throws UnsupportedAuthenticationException;

    /**
     * Configure the authentication provider to use in the Pulsar client instance
     * using a config map.
     *
     * <p>Example:
     * <pre>{@code
     * Map<String, String> conf = new TreeMap<>();
     * conf.put("tlsCertFile", "/my/cert/file");
     * conf.put("tlsKeyFile", "/my/key/file");
     *
     * PulsarClient client = PulsarClient.builder()
     *          .serviceUrl("pulsar+ssl://broker.example.com:6651/)
     *          .authentication(
     *              "org.apache.pulsar.client.impl.auth.AuthenticationTls", conf)
     *          .build();
     * }</pre>
     *
     * @param authPluginClassName
     *            name of the Authentication-Plugin you want to use
     * @param authParams
     *            map which represents parameters for the Authentication-Plugin
     * @return the client builder instance
     * @throws UnsupportedAuthenticationException
     *             failed to instantiate specified Authentication-Plugin
     */
    ClientBuilder authentication(String authPluginClassName, Map<String, String> authParams)
            throws UnsupportedAuthenticationException;

    /**
     * Set the operation timeout <i>(default: 30 seconds)</i>.
     *
     * <p>Producer-create, subscribe and unsubscribe operations will be retried until this interval, after which the
     * operation will be marked as failed
     *
     * @param operationTimeout
     *            operation timeout
     * @param unit
     *            time unit for {@code operationTimeout}
     * @return the client builder instance
     */
    ClientBuilder operationTimeout(int operationTimeout, TimeUnit unit);

    /**
     * Set the number of threads to be used for handling connections to brokers <i>(default: 1 thread)</i>.
     *
     * @param numIoThreads the number of IO threads
     * @return the client builder instance
     */
    ClientBuilder ioThreads(int numIoThreads);

    /**
     * Set the number of threads to be used for message listeners <i>(default: 1 thread)</i>.
     *
     * <p>The listener thread pool is shared across all the consumers and readers that are
     * using a "listener" model to get messages. For a given consumer, the listener will be
     * always invoked from the same thread, to ensure ordering.
     *
     * @param numListenerThreads the number of listener threads
     * @return the client builder instance
     */
    ClientBuilder listenerThreads(int numListenerThreads);

    /**
     * Sets the max number of connection that the client library will open to a single broker.
     *
     * <p>By default, the connection pool will use a single connection for all the producers and consumers.
     * Increasing this parameter may improve throughput when using many producers over a high latency connection.
     *
     * @param connectionsPerBroker
     *            max number of connections per broker (needs to be greater than 0)
     * @return the client builder instance
     */
    ClientBuilder connectionsPerBroker(int connectionsPerBroker);

    /**
     * Configure whether to use TCP no-delay flag on the connection, to disable Nagle algorithm.
     *
     * <p>No-delay features make sure packets are sent out on the network as soon as possible, and it's critical
     * to achieve low latency publishes. On the other hand, sending out a huge number of small packets
     * might limit the overall throughput, so if latency is not a concern,
     * it's advisable to set the <code>useTcpNoDelay</code> flag to false.
     *
     * <p>Default value is true.
     *
     * @param enableTcpNoDelay whether to enable TCP no-delay feature
     * @return the client builder instance
     */
    ClientBuilder enableTcpNoDelay(boolean enableTcpNoDelay);

    /**
     * Configure whether to use TLS encryption on the connection
     * <i>(default: true if serviceUrl starts with "pulsar+ssl://", false otherwise)</i>.
     *
     * @param enableTls
     * @deprecated use "pulsar+ssl://" in serviceUrl to enable
     * @return the client builder instance
     */
    @Deprecated
    ClientBuilder enableTls(boolean enableTls);

    /**
     * Set the path to the trusted TLS certificate file.
     *
     * @param tlsTrustCertsFilePath
     * @return the client builder instance
     */
    ClientBuilder tlsTrustCertsFilePath(String tlsTrustCertsFilePath);

    /**
     * Configure whether the Pulsar client accept untrusted TLS certificate from broker <i>(default: false)</i>.
     *
     * @param allowTlsInsecureConnection whether to accept a untrusted TLS certificate
     * @return the client builder instance
     */
    ClientBuilder allowTlsInsecureConnection(boolean allowTlsInsecureConnection);

    /**
     * It allows to validate hostname verification when client connects to broker over tls. It validates incoming x509
     * certificate and matches provided hostname(CN/SAN) with expected broker's host name. It follows RFC 2818, 3.1.
     * Server Identity hostname verification.
     *
     * @see <a href="https://tools.ietf.org/html/rfc2818">RFC 818</a>
     *
     * @param enableTlsHostnameVerification whether to enable TLS hostname verification
     * @return the client builder instance
     */
    ClientBuilder enableTlsHostnameVerification(boolean enableTlsHostnameVerification);

    /**
     * Set the interval between each stat info <i>(default: 60 seconds)</i> Stats will be activated with positive
     * statsInterval It should be set to at least 1 second.
     *
     * @param statsInterval
     *            the interval between each stat info
     * @param unit
     *            time unit for {@code statsInterval}
     * @return the client builder instance
     */
    ClientBuilder statsInterval(long statsInterval, TimeUnit unit);

    /**
     * Number of concurrent lookup-requests allowed to send on each broker-connection to prevent overload on broker.
     * <i>(default: 5000)</i> It should be configured with higher value only in case of it requires to produce/subscribe
     * on thousands of topic using created {@link PulsarClient}.
     *
     * @param maxConcurrentLookupRequests
     * @return the client builder instance
     */
    ClientBuilder maxConcurrentLookupRequests(int maxConcurrentLookupRequests);

    /**
     * Number of max lookup-requests allowed on each broker-connection to prevent overload on broker.
     * <i>(default: 50000)</i> It should be bigger than maxConcurrentLookupRequests.
     * Requests that inside maxConcurrentLookupRequests already send to broker, and requests beyond
     * maxConcurrentLookupRequests and under maxLookupRequests will wait in each client cnx.
     *
     * @param maxLookupRequests
     * @return the client builder instance
     */
    ClientBuilder maxLookupRequests(int maxLookupRequests);

    /**
     * Set max number of broker-rejected requests in a certain time-frame (30 seconds) after which current connection
     * will be closed and client creates a new connection that give chance to connect a different broker <i>(default:
     * 50)</i>.
     *
     * @param maxNumberOfRejectedRequestPerConnection
     * @return the client builder instance
     */
    ClientBuilder maxNumberOfRejectedRequestPerConnection(int maxNumberOfRejectedRequestPerConnection);

    /**
     * Set keep alive interval for each client-broker-connection. <i>(default: 30 seconds)</i>.
     *
     * @param keepAliveInterval
     * @param unit the time unit in which the keepAliveInterval is defined
     * @return the client builder instance
     */
    ClientBuilder keepAliveInterval(int keepAliveInterval, TimeUnit unit);

    /**
     * Set the duration of time to wait for a connection to a broker to be established. If the duration passes without a
     * response from the broker, the connection attempt is dropped.
     *
     * @since 2.3.0
     * @param duration
     *            the duration to wait
     * @param unit
     *            the time unit in which the duration is defined
     * @return the client builder instance
     */
    ClientBuilder connectionTimeout(int duration, TimeUnit unit);

    /**
     * Set the duration of time for a backoff interval.
     *
     * @param duration the duration of the interval
     * @param unit the time unit in which the duration is defined
     * @return the client builder instance
     */
    ClientBuilder startingBackoffInterval(long duration, TimeUnit unit);

    /**
     * Set the maximum duration of time for a backoff interval.
     *
     * @param duration the duration of the interval
     * @param unit the time unit in which the duration is defined
     * @return the client builder instance
     */
    ClientBuilder maxBackoffInterval(long duration, TimeUnit unit);

    /**
     * The clock used by the pulsar client.
     *
     * <p>The clock is currently used by producer for setting publish timestamps.
     * {@link Clock#millis()} is called to retrieve current timestamp as the publish
     * timestamp when producers produce messages. The default clock is a system default zone
     * clock. So the publish timestamp is same as calling {@link System#currentTimeMillis()}.
     *
     * <p>Warning: the clock is used for TTL enforcement and timestamp based seeks.
     * so be aware of the impacts if you are going to use a different clock.
     *
     * @param clock the clock used by the pulsar client to retrieve time information
     * @return the client builder instance
     */
    ClientBuilder clock(Clock clock);
}
