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

import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.io.StringWriter;
import java.util.List;
import java.util.Optional;

import org.apache.commons.lang.StringUtils;
import org.apache.http.HttpResponse;
import org.apache.http.HttpStatus;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.message.BasicHeader;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.http.MediaType;
import org.springframework.http.converter.json.Jackson2ObjectMapperBuilder;
import org.springframework.stereotype.Component;
import org.talend.dataprep.api.APIMessages;
import org.talend.dataprep.api.dataset.ColumnMetadata;
import org.talend.dataprep.api.dataset.DataSetMetadata;
import org.talend.dataprep.api.service.PreparationAPI;
import org.talend.dataprep.exception.Exceptions;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.netflix.hystrix.HystrixCommand;

@Component
@Scope("request")
public class SuggestColumnActions extends ChainedCommand<InputStream, DataSetMetadata> {

    private final String transformServiceUrl;

    private final String column;

    private final HttpClient client;

    @Autowired(required = true)
    private Jackson2ObjectMapperBuilder builder;

    private SuggestColumnActions(HttpClient client, String transformServiceUrl, HystrixCommand<DataSetMetadata> retrieveMetadata,
            String columnName) {
        super(PreparationAPI.TRANSFORM_GROUP, retrieveMetadata);
        this.transformServiceUrl = transformServiceUrl;
        this.column = columnName;
        this.client = client;
    }

    @Override
    protected InputStream run() throws Exception {
        HttpPost post = new HttpPost(transformServiceUrl + "/suggest/column");
        post.setHeader(new BasicHeader("Content-Type", MediaType.APPLICATION_JSON_VALUE));
        DataSetMetadata metadata = getInput();

        if (metadata == null) {
            // FIXME add dataset id in error message?
            throw Exceptions.User(APIMessages.UNABLE_TO_RETRIEVE_DATASET_METADATA);
        }

        List<ColumnMetadata> columns = metadata.getRow().getColumns();

        Optional<ColumnMetadata> columnMetadata = columns.stream()
                .filter(columnMetadataCurrent -> StringUtils.equals(column, columnMetadataCurrent.getId())).findFirst();

        if (!columnMetadata.isPresent()) {
            // Column does not exist in data set metadata.
            return new ByteArrayInputStream(new byte[0]);
        }
        ObjectMapper objectMapper = builder.build();
        StringWriter columnMetadataJSON = new StringWriter();
        objectMapper.writer().writeValue(columnMetadataJSON, columnMetadata.get());
        post.setEntity(new StringEntity(columnMetadataJSON.toString()));
        HttpResponse response = client.execute(post);
        int statusCode = response.getStatusLine().getStatusCode();
        if (statusCode >= 200) {
            if (statusCode == HttpStatus.SC_NO_CONTENT || statusCode == HttpStatus.SC_ACCEPTED) {
                // Immediately release connection
                post.releaseConnection();
                return null;
            } else if (statusCode == HttpStatus.SC_OK) {
                return new ReleasableInputStream(response.getEntity().getContent(), post::releaseConnection);
            }
        }
        throw Exceptions.User(APIMessages.UNABLE_TO_RETRIEVE_SUGGESTED_ACTIONS);
    }
}

@@ -232,16 +232,52 @@
 gUtils;%0A
+import org.apache.http.HttpHeaders;%0A
 import o
@@ -478,52 +478,8 @@
 ty;%0A
-import org.apache.http.message.BasicHeader;%0A
 impo
@@ -1960,38 +1960,32 @@
 der(
-new BasicHeader(%22Content-Type%22
+HttpHeaders.CONTENT_TYPE
 , Me
@@ -2015,17 +2015,16 @@
 N_VALUE)
-)
 ;%0A      

use constants