// ============================================================================
// Copyright (C) 2006-2016 Talend Inc. - www.talend.com
//
// This source code is available under agreement available at
// https://github.com/Talend/data-prep/blob/master/LICENSE
//
// You should have received a copy of the agreement
// along with this program; if not, write to Talend SA
// 9 rue Pages 92150 Suresnes, France
//
// ============================================================================

package org.talend.dataprep;

import java.io.IOException;
import java.io.InputStream;
import java.util.*;
import java.util.stream.Collectors;

import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.talend.dataprep.api.dataset.RowMetadata;
import org.talend.dataprep.api.preparation.Action;
import org.talend.dataprep.api.preparation.PreparationMessage;
import org.talend.dataprep.api.preparation.json.MixedContentMapModule;
import org.talend.dataprep.transformation.actions.common.ActionFactory;
import org.talend.dataprep.transformation.actions.common.RunnableAction;
import org.talend.dataprep.transformation.pipeline.ActionRegistry;

import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

/**
 * Provides utility to parses actions and associated {@link RowMetadata}.
 */
public class PreparationParser {

    public static final ActionRegistry actionRegistry = new ClassPathActionRegistry("org.talend.dataprep.transformation.actions");

    private static final ActionFactory actionFactory = new ActionFactory();

    private static ObjectMapper mapper = new ObjectMapper();

    private static void assertPreparation(Object preparation) {
        if (preparation == null) {
            throw new IllegalArgumentException("Preparation can not be null.");
        }
    }

    static {
        mapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
        mapper.registerModule(new MixedContentMapModule());
    }

    public static PreparationMessage parseCorePreparation(InputStream preparation) {
        assertPreparation(preparation);
        try {
            final PreparationMessage preparationMessage = mapper.reader(PreparationMessage.class).readValue(preparation);
            if (preparationMessage.getRowMetadata() == null) {
                preparationMessage.setRowMetadata(new RowMetadata());
            }
            return preparationMessage;
        } catch (IOException e) {
            throw new IllegalArgumentException("Unable to parse preparation", e);
        }
    }

    public static StandalonePreparation parseExportableCorePreparation(InputStream inputStream) {
        try {
            return mapper.reader(StandalonePreparation.class).readValue(inputStream);
        } catch (IOException e) {
            throw new IllegalArgumentException("Unable to create the provided exportable preparation into a function" + e);
        }
    }

    public static List<RunnableAction> ensureActionRowsExistence(List<Action> actions) {
        return actions.stream().map(a -> actionFactory.create(actionRegistry.get(a.getName()), a.getParameters()))
                .collect(Collectors.toList());
    }

}

@@ -643,77 +643,20 @@
 org.
-apache.commons.io.IOUtils;%0Aimport org.apache.commons.lang.StringUtils
+slf4j.Logger
 ;%0Aim
@@ -668,32 +668,39 @@
 org.slf4j.Logger
+Factory
 ;%0Aimport org.slf
@@ -696,35 +696,59 @@
 ort org.
-slf4j.LoggerFactory
+talend.dataprep.api.action.ActionDefinition
 ;%0Aimport
@@ -1257,56 +1257,8 @@
 re;%0A
-import com.fasterxml.jackson.databind.JsonNode;%0A
 impo
@@ -1306,16 +1306,204 @@
 apper;%0A%0A
+import static org.talend.dataprep.api.action.ActionDefinition.Behavior.FORBID_DISTRIBUTED;%0Aimport static org.talend.dataprep.api.action.ActionDefinition.Behavior.METADATA_CREATE_COLUMNS;%0A%0A
 /**%0A * P
@@ -1610,16 +1610,108 @@
 rser %7B%0A%0A
+    private static final Logger LOGGER = LoggerFactory.getLogger(PreparationParser.class);%0A%0A
     publ
@@ -3366,43 +3366,1524 @@
 ions
-) %7B%0A        return actions.stream()
+, boolean allowNonDistributedActions) %7B%0A        return actions.stream() //%0A                .map(action -%3E %7B%0A                    final ActionDefinition actionDefinition = actionRegistry.get(action.getName());%0A                    // Distributed run check for action%0A                    final Set%3CActionDefinition.Behavior%3E behavior = actionDefinition.getBehavior();%0A                    // if non distributed actions are forbidden (e.g. running locally)%0A                    if (!allowNonDistributedActions) %7B%0A                        // if some actions cannot be run in distributed environment, let's see how bad it is...%0A                        if (behavior.contains(FORBID_DISTRIBUTED)) %7B%0A                            // actions that changes the schema (potentially really harmful for the preparation) throws an exception%0A                            if (behavior.contains(METADATA_CREATE_COLUMNS)) %7B%0A                                throw new IllegalArgumentException(%22Action '%22 + actionDefinition.getName() + %22' cannot run in distributed environments.%22);%0A                            %7D else %7B%0A                                // else the action is just skipped%0A                                LOGGER.warn(%22Action '%7B%7D' cannot run in distributed environment, skip its execution.%22);%0A                                return null;%0A                            %7D%0A                        %7D%0A                    %7D%0A                    return action;%0A                %7D) //%0A                .filter(Objects::nonNull) //%0A                
 .map
@@ -4961,16 +4961,19 @@
 ters()))
+ //
 %0A       

Fix wrongfully removed code during feature development.