package org.talend.dataprep.api.service.command.preparation;

import java.io.InputStream;
import java.io.OutputStream;
import java.util.List;

import org.apache.http.client.HttpClient;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import org.talend.dataprep.api.preparation.Action;
import org.talend.dataprep.api.service.APIService;
import org.talend.dataprep.api.service.command.CloneInputStream;
import org.talend.dataprep.api.service.command.common.PreparationCommand;
import org.talend.dataprep.api.service.command.transformation.Transform;
import org.talend.dataprep.cache.ContentCache;
import org.talend.dataprep.cache.ContentCacheKey;

import com.netflix.hystrix.HystrixCommand;

/**
 * Command used to retrieve the preparation content.
 */
@Component
@Scope("request")
public class PreparationGetContent extends PreparationCommand<InputStream> {

    /** The preparation id. */
    private final String id;

    /** The preparation version. */
    private final String version;

    /** Optional sample size (if null or <=0, the full preparation content is returned). */
    private Long sample;

    /**
     * Constructor.
     *
     * @param client the http client to use.
     * @param id the preparation id.
     * @param version the preparation version.
     */
    private PreparationGetContent(HttpClient client, String id, String version) {
        this(client, id, version, null);
    }

    /**
     * Constructor with sample size specified.
     *
     * @param client the http client to use.
     * @param id the preparation id.
     * @param version the preparation version.
     */
    private PreparationGetContent(HttpClient client, String id, String version, Long sample) {
        super(APIService.PREPARATION_GROUP, client);
        this.id = id;
        this.version = version;
        this.sample = sample;
    }

    /**
     * @see HystrixCommand#run()
     */
    @Override
    protected InputStream run() throws Exception {

        PreparationContext preparationContext = getContext(id, version, sample);
        InputStream content = preparationContext.getContent();
        List<Action> actions = preparationContext.getActions();

        // preparation in cache
        if (preparationContext.fromCache()) {
            return content;
        }

        // Run transformation (if any action to perform)
        if (!actions.isEmpty()) {
            final String encodedActions = serializeActions(actions);
            // pass content to the transformation service as input as well as actions to perform on input...
            final Transform transformCommand = context.getBean(Transform.class, client, content, encodedActions);
            content = transformCommand.execute();
        }

        //@formatter:off
        ContentCacheKey key = new ContentCacheKey(preparationContext.getPreparation(),
                                                  preparationContext.getVersion(),
                                                  sample);
        //@formatter:on
        final OutputStream newCacheEntry = contentCache.put(key, ContentCache.TimeToLive.DEFAULT);
        return new CloneInputStream(content, newCacheEntry);
    }

}

@@ -521,16 +521,91 @@
 ommand;%0A
+import org.talend.dataprep.api.service.command.dataset.DataSetGetMetadata;%0A
 import o
@@ -3132,16 +3132,338 @@
 tter:on%0A
+        DataSetGetMetadata getMetadata = context.getBean(DataSetGetMetadata.class, client, preparationContext.getPreparation().getDataSetId());%0A        final boolean inProgress = getMetadata.execute().getLifecycle().inProgress();%0A        if (!inProgress) %7B%0A            // Statistics are done, can safely save to cache%0A    
         
@@ -3557,24 +3557,28 @@
 T);%0A        
+    
 return new C
@@ -3618,16 +3618,71 @@
 Entry);%0A
+        %7D else %7B%0A            return content;%0A        %7D%0A
     %7D%0A%0A%7D

Test for inProgress status to decide if content should be cached or not.