package org.apache.kafka.streams.processor.api;

/**
 * Wrapper class that can be used to inject custom wrappers around the processors of their application topology.
 * The returned instance should wrap the supplied {@code ProcessorSupplier} and the {@code Processor} it supplies
 * to avoid disrupting the regular processing of the application, although this is not required and any processor
 * implementation can be substituted in to replace the original processor entirely (which may be useful for example
 * while testing or debugging an application topology).
 * <p>
 * NOTE: in order to use this feature, you must set the {@link StreamsConfig#PROCESSOR_WRAPPER} config and pass it
 * in as a {@link TopologyConfig} when creating the {@link StreamsBuilder} or {@link Topology} by using the
 * appropriate constructor (ie {@link StreamsBuilder#StreamsBuilder(TopologyConfig)} or {@link Topology#Topology(TopologyConfig)})
 * <p>
 * Can be configured, if desired, by implementing the {@link #configure(Map)} method. This will be invoked when
 * the {@code ProcessorWrapper} is instantiated, and will provide it with the TopologyConfigs that were passed in 
 * to the {@link StreamsBuilder} or {@link Topology} constructor.
 */
public interface ProcessorWrapper extends Configurable {

    @Override
    default void configure(final Map<String, ?> configs) {
        // do nothing
    }

    /**
     * Wrap or replace the provided {@link ProcessorSupplier} and return a {@link WrappedProcessorSupplier}
     * To convert a {@link ProcessorSupplier} instance into a {@link WrappedProcessorSupplier},
     * use the {@link ProcessorWrapper#asWrapped(ProcessorSupplier)} method
     */  
    <KIn, VIn, KOut, VOut> WrappedProcessorSupplier<KIn, VIn, KOut, VOut> wrapProcessorSupplier(final String processorName,
                                                                                                        final ProcessorSupplier<KIn, VIn, KOut, VOut> processorSupplier);

    /**
     * Wrap or replace the provided {@link FixedKeyProcessorSupplier} and return a {@link WrappedFixedKeyProcessorSupplier}
     * To convert a {@link FixedKeyProcessorSupplier} instance into a {@link WrappedFixedKeyProcessorSupplier},
     * use the {@link ProcessorWrapper#asWrappedFixedKey(FixedKeyProcessorSupplier)} method
     */
     <KIn, VIn,  VOut> WrappedFixedKeyProcessorSupplier<KIn, VIn,  VOut> wrapFixedKeyProcessorSupplier(final String processorName,
                                                                                                      final FixedKeyProcessorSupplier<KIn, VIn, VOut> processorSupplier);

     /**
     * Use to convert a {@link ProcessorSupplier} instance into a {@link WrappedProcessorSupplier}
     */
     static <KIn, VIn, KOut, VOut> WrappedProcessorSupplier<KIn, VIn, KOut, VOut> asWrapped(
        final ProcessorSupplier<KIn, VIn, KOut, VOut> processorSupplier
    ) {
        return new WrappedProcessorSupplierImpl<>(processorSupplier);
    }

    /**
     * Use to convert a {@link FixedKeyProcessorSupplier} instance into a {@link WrappedFixedKeyProcessorSupplier}
     */
    static <KIn, VIn,  VOut> WrappedFixedKeyProcessorSupplier<KIn, VIn,  VOut> asWrappedFixedKey(
        final FixedKeyProcessorSupplier<KIn, VIn, VOut> processorSupplier
    ) {
        return new WrappedFixedKeyProcessorSupplierImpl<>(processorSupplier);
    } }

package org.apache.kafka.streams.processor.api;  

/**
 * Marker interface for classes implementing {@link ProcessorSupplier}
 * that have been wrapped via a {@link ProcessorWrapper}.
 * <p>
 * To convert a {@link ProcessorSupplier} instance into a {@link WrappedProcessorSupplier},
 * use the {@link ProcessorWrapper#asWrapped(ProcessorSupplier)} method
 */
public interface WrappedProcessorSupplier<KIn, VIn, KOut, VOut> extends ProcessorSupplier<KIn, VIn, KOut, VOut> {

}

package org.apache.kafka.streams.processor.api; 

/**
 * Marker interface for classes implementing {@link FixedKeyProcessorSupplier}
 * that have been wrapped via a {@link ProcessorWrapper}.
 * <p>
 * To convert a {@link FixedKeyProcessorSupplier} instance into a {@link WrappedFixedKeyProcessorSupplier},
 * use the {@link ProcessorWrapper#asWrappedFixedKey(FixedKeyProcessorSupplier)} method
 */
public interface WrappedFixedKeyProcessorSupplier<KIn, VIn, VOut> extends FixedKeyProcessorSupplier<KIn, VIn, VOut> {

}