<commit_before>/*
 * =============================================================================
 * 
 *   Copyright (c) 2011-2016, The THYMELEAF team (http://www.thymeleaf.org)
 * 
 *   Licensed 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.thymeleaf.spring5;

import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.nio.charset.Charset;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.logging.Level;

import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.thymeleaf.IThrottledTemplateProcessor;
import org.thymeleaf.TemplateEngine;
import org.thymeleaf.context.AbstractContext;
import org.thymeleaf.context.IContext;
import org.thymeleaf.context.IEngineContext;
import org.thymeleaf.engine.DataDrivenTemplateIterator;
import org.thymeleaf.exceptions.TemplateProcessingException;
import org.thymeleaf.spring5.context.reactive.IReactiveDataDriverContextVariable;
import org.thymeleaf.spring5.context.reactive.SpringWebReactiveEngineContextFactory;
import org.thymeleaf.spring5.linkbuilder.reactive.SpringWebReactiveLinkBuilder;
import org.thymeleaf.util.LoggingUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;


/**
 *
 * @author <NAME>
 * 
 * @since 3.0.3
 *
 */
public class SpringWebReactiveTemplateEngine
        extends SpringTemplateEngine
        implements ISpringWebReactiveTemplateEngine {


    private static final Logger logger = LoggerFactory.getLogger(SpringWebReactiveTemplateEngine.class);
    private static final String LOG_CATEGORY_FULL_OUTPUT = SpringWebReactiveTemplateEngine.class.getName() + ".OUTPUT.FULL";
    private static final String LOG_CATEGORY_CHUNKED_OUTPUT = SpringWebReactiveTemplateEngine.class.getName() + ".OUTPUT.CHUNKED";
    private static final String LOG_CATEGORY_DATADRIVEN_INPUT = SpringWebReactiveTemplateEngine.class.getName() + ".INPUT.DATADRIVEN";
    private static final String LOG_CATEGORY_DATADRIVEN_OUTPUT = SpringWebReactiveTemplateEngine.class.getName() + ".OUTPUT.DATADRIVEN";




    public SpringWebReactiveTemplateEngine() {

        super();
        // In Spring Web Reactive environments, we will need to use a special context factory in order to
        // use an environment-tailored implementation of IEngineContext.
        this.setEngineContextFactory(new SpringWebReactiveEngineContextFactory());
        // In Spring Web Reactive environments, we will need to use a special link builder able to adapt
        // the creation of URLs as a result of @{...} expressions in a way that makes sense in this
        // environment.
        this.setLinkBuilder(new SpringWebReactiveLinkBuilder());

    }




    @Override
    public Publisher<DataBuffer> processStream(
            final String template, final Set<String> markupSelectors, final IContext context,
            final DataBufferFactory bufferFactory, final Charset charset) {

        /*
         * PERFORM VALIDATIONS
         */
        if (template == null) {
            return Flux.error(new IllegalArgumentException("Template cannot be null"));
        }
        if (context == null) {
            return Flux.error(new IllegalArgumentException("Context cannot be null"));
        }
        if (bufferFactory == null) {
            return Flux.error(new IllegalArgumentException("Buffer Factory cannot be null"));
        }
        if (charset == null) {
            return Flux.error(new IllegalArgumentException("Charset cannot be null"));
        }

        /*
         * CHECK FOR DATA-DRIVEN EXECUTION
         */
        try {
            final String dataDriverVariableName = findDataDriverInModel(context);
            if (dataDriverVariableName != null) {
                // We should be executing in data-driven mode
                return createDataDrivenStream(
                        template, markupSelectors, context, dataDriverVariableName, bufferFactory, charset, Integer.MAX_VALUE);
            }
        } catch (final Throwable t) {
            return Flux.error(t);
        }

        /*
         * CREATE A CHUNKED STREAM
         */
        return createFullStream(template, markupSelectors, context, bufferFactory, charset);

    }


    @Override
    public Publisher<DataBuffer> processStream(
            final String template, final Set<String> markupSelectors, final IContext context,
            final DataBufferFactory bufferFactory, final Charset charset, final int responseMaxChunkSizeBytes) {

        /*
         * PERFORM VALIDATIONS
         */
        if (template == null) {
            return Flux.error(new IllegalArgumentException("Template cannot be null"));
        }
        if (context == null) {
            return Flux.error(new IllegalArgumentException("Context cannot be null"));
        }
        if (bufferFactory == null) {
            return Flux.error(new IllegalArgumentException("Buffer Factory cannot be null"));
        }
        if (charset == null) {
            return Flux.error(new IllegalArgumentException("Charset cannot be null"));
        }

        if (responseMaxChunkSizeBytes == 0) {
            return Flux.error(new IllegalArgumentException("Max Chunk Size cannot be zero"));
        }

        // Normalize the chunk size in bytes (MAX_VALUE == no limit)
        final int chunkSizeBytes = (responseMaxChunkSizeBytes < 0? Integer.MAX_VALUE : responseMaxChunkSizeBytes);

        /*
         * CHECK FOR DATA-DRIVEN EXECUTION
         */
        try {
            final String dataDriverVariableName = findDataDriverInModel(context);
            if (dataDriverVariableName != null) {
                // We should be executing in data-driven mode
                return createDataDrivenStream(
                        template, markupSelectors, context, dataDriverVariableName, bufferFactory, charset, chunkSizeBytes);
            }
        } catch (final Throwable t) {
            return Flux.error(t);
        }

        /*
         * IS THERE A LIMIT IN BUFFER SIZE? if not, given we are not data-driven, we should switch to FULL
         */
        if (chunkSizeBytes == Integer.MAX_VALUE) {
            // No limit on buffer size, so there is no reason to throttle: using FULL mode instead.
            return processStream(template, markupSelectors, context, bufferFactory, charset);
        }

        /*
         * CREATE A CHUNKED STREAM
         */
        return createChunkedStream(
                template, markupSelectors, context, bufferFactory, charset, responseMaxChunkSizeBytes);

    }




    private Mono<DataBuffer> createFullStream(
            final String templateName, final Set<String> markupSelectors, final IContext context,
            final DataBufferFactory bufferFactory, final Charset charset) {

        final Mono<DataBuffer> stream =
                Mono.create(
                        subscriber -> {

                            if (logger.isTraceEnabled()) {
                                logger.trace("[THYMELEAF][{}] STARTING STREAM PROCESS (FULL MODE) OF TEMPLATE \"{}\" WITH LOCALE {}",
                                        new Object[]{TemplateEngine.threadIndex(), LoggingUtils.loggifyTemplateName(templateName), context.getLocale()});
                            }

                            final DataBuffer dataBuffer = bufferFactory.allocateBuffer();
                            // OutputStreamWriter object have an 8K buffer, but process(...) will flush it at the end
                            final OutputStreamWriter writer = new OutputStreamWriter(dataBuffer.asOutputStream(), charset);
<commit_msg>Restructured processStream calls in SpringWebReactiveTemplateEngine<commit_after>/*
 * =============================================================================
 * 
 *   Copyright (c) 2011-2016, The THYMELEAF team (http://www.thymeleaf.org)
 * 
 *   Licensed 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.thymeleaf.spring5;

import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.nio.charset.Charset;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.logging.Level;

import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.thymeleaf.IThrottledTemplateProcessor;
import org.thymeleaf.TemplateEngine;
import org.thymeleaf.context.AbstractContext;
import org.thymeleaf.context.IContext;
import org.thymeleaf.context.IEngineContext;
import org.thymeleaf.engine.DataDrivenTemplateIterator;
import org.thymeleaf.exceptions.TemplateProcessingException;
import org.thymeleaf.spring5.context.reactive.IReactiveDataDriverContextVariable;
import org.thymeleaf.spring5.context.reactive.SpringWebReactiveEngineContextFactory;
import org.thymeleaf.spring5.linkbuilder.reactive.SpringWebReactiveLinkBuilder;
import org.thymeleaf.util.LoggingUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;


/**
 *
 * @author <NAME>
 * 
 * @since 3.0.3
 *
 */
public class SpringWebReactiveTemplateEngine
        extends SpringTemplateEngine
        implements ISpringWebReactiveTemplateEngine {


    private static final Logger logger = LoggerFactory.getLogger(SpringWebReactiveTemplateEngine.class);
    private static final String LOG_CATEGORY_FULL_OUTPUT = SpringWebReactiveTemplateEngine.class.getName() + ".OUTPUT.FULL";
    private static final String LOG_CATEGORY_CHUNKED_OUTPUT = SpringWebReactiveTemplateEngine.class.getName() + ".OUTPUT.CHUNKED";
    private static final String LOG_CATEGORY_DATADRIVEN_INPUT = SpringWebReactiveTemplateEngine.class.getName() + ".INPUT.DATADRIVEN";
    private static final String LOG_CATEGORY_DATADRIVEN_OUTPUT = SpringWebReactiveTemplateEngine.class.getName() + ".OUTPUT.DATADRIVEN";




    public SpringWebReactiveTemplateEngine() {

        super();
        // In Spring Web Reactive environments, we will need to use a special context factory in order to
        // use an environment-tailored implementation of IEngineContext.
        this.setEngineContextFactory(new SpringWebReactiveEngineContextFactory());
        // In Spring Web Reactive environments, we will need to use a special link builder able to adapt
        // the creation of URLs as a result of @{...} expressions in a way that makes sense in this
        // environment.
        this.setLinkBuilder(new SpringWebReactiveLinkBuilder());

    }




    @Override
    public Publisher<DataBuffer> processStream(
            final String template, final Set<String> markupSelectors, final IContext context,
            final DataBufferFactory bufferFactory, final Charset charset) {
        return processStream(template, markupSelectors, context, bufferFactory, charset, Integer.MAX_VALUE);
    }


    @Override
    public Publisher<DataBuffer> processStream(
            final String template, final Set<String> markupSelectors, final IContext context,
            final DataBufferFactory bufferFactory, final Charset charset, final int responseMaxChunkSizeBytes) {

        /*
         * PERFORM VALIDATIONS
         */
        if (template == null) {
            return Flux.error(new IllegalArgumentException("Template cannot be null"));
        }
        if (context == null) {
            return Flux.error(new IllegalArgumentException("Context cannot be null"));
        }
        if (bufferFactory == null) {
            return Flux.error(new IllegalArgumentException("Buffer Factory cannot be null"));
        }
        if (charset == null) {
            return Flux.error(new IllegalArgumentException("Charset cannot be null"));
        }

        if (responseMaxChunkSizeBytes == 0) {
            return Flux.error(new IllegalArgumentException("Max Chunk Size cannot be zero"));
        }

        // Normalize the chunk size in bytes (MAX_VALUE == no limit)
        final int chunkSizeBytes = (responseMaxChunkSizeBytes < 0? Integer.MAX_VALUE : responseMaxChunkSizeBytes);

        /*
         * CHECK FOR DATA-DRIVEN EXECUTION
         */
        try {
            final String dataDriverVariableName = findDataDriverInModel(context);
            if (dataDriverVariableName != null) {
                // We should be executing in data-driven mode
                return createDataDrivenStream(
                        template, markupSelectors, context, dataDriverVariableName, bufferFactory, charset, chunkSizeBytes);
            }
        } catch (final Throwable t) {
            return Flux.error(t);
        }

        /*
         * IS THERE A LIMIT IN BUFFER SIZE? if not, given we are not data-driven, we should switch to FULL
         */
        if (chunkSizeBytes == Integer.MAX_VALUE) {
            // No limit on buffer size, so there is no reason to throttle: using FULL mode instead.
            return createFullStream(template, markupSelectors, context, bufferFactory, charset);
        }

        /*
         * CREATE A CHUNKED STREAM
         */
        return createChunkedStream(
                template, markupSelectors, context, bufferFactory, charset, responseMaxChunkSizeBytes);

    }




    private Mono<DataBuffer> createFullStream(
            final String templateName, final Set<String> markupSelectors, final IContext context,
            final DataBufferFactory bufferFactory, final Charset charset) {

        final Mono<DataBuffer> stream =
                Mono.create(
                        subscriber -> {

                            if (logger.isTraceEnabled()) {
                                logger.trace("[THYMELEAF][{}] STARTING STREAM PROCESS (FULL MODE) OF TEMPLATE \"{}\" WITH LOCALE {}",
                                        new Object[]{TemplateEngine.threadIndex(), LoggingUtils.loggifyTemplateName(templateName), context.getLocale()});
                            }

                            final DataBuffer dataBuffer = bufferFactory.allocateBuffer();
                            // OutputStreamWriter object have an 8K buffer, but process(...) will flush it at the end
                            final OutputStreamWriter writer = new OutputStreamWriter(dataBuffer.asOutputStream(), charset);

{'fraction_non_alphanumeric': 0.09799140088558043, 'fraction_numerical': 0.0030802797920811142, 'mean_word_length': 2.8252331860579285, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 4, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6350714', 'n_tokens_mistral': 4015, 'n_tokens_neox': 3836, 'n_words': 1124}