// ============================================================================
//
// Copyright (C) 2006-2015 Talend Inc. - www.talend.com
//
// This source code is available under agreement available at
// %InstallDIR%\features\org.talend.rcp.branding.%PRODUCTNAME%\%PRODUCTNAME%license.txt
//
// 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.components.salesforce.runtime;

import org.apache.avro.Schema;
import org.apache.avro.generic.IndexedRecord;
import org.talend.components.api.component.runtime.WriteOperation;
import org.talend.components.api.component.runtime.WriterResult;
import org.talend.components.api.container.RuntimeContainer;
import org.talend.components.common.BulkFileProperties;
import org.talend.components.common.runtime.BulkFileWriter;
import org.talend.components.salesforce.tsalesforceoutputbulk.TSalesforceOutputBulkProperties;

import java.util.ArrayList;
import java.util.List;

/**
 * Prepare Data Files for bulk execution
 */
final class SalesforceBulkFileWriter extends BulkFileWriter {

	public SalesforceBulkFileWriter(WriteOperation<WriterResult> writeOperation, BulkFileProperties bulkProperties,
			RuntimeContainer adaptor) {
		super(writeOperation, bulkProperties, adaptor);
	}

	@Override
	public String[] getHeaders(Schema schema) {
		TSalesforceOutputBulkProperties salesforceBulkProperties = (TSalesforceOutputBulkProperties)bulkProperties;
		
		List<String> headers = new ArrayList<String>();
		StringBuilder sbuilder = new StringBuilder();
        for(Schema.Field f :schema.getFields()){
        	String header = f.name();
        	
        	String ref_module_name = f.getProp(SalesforceSchemaConstants.REF_MODULE_NAME);
        	String ref_field_name = f.getProp(SalesforceSchemaConstants.REF_FIELD_NAME);
        	if(ref_module_name!=null) {
        		header = sbuilder.append(ref_module_name).append(":").append(ref_field_name).append(".").append(f.name()).toString();
        		sbuilder.setLength(0);
        	} else {
        		int index = getIndex((List<String>)salesforceBulkProperties.upsertRelationTable.columnName.getValue(), header);
            	if(index > -1) {
            		List<String> polymorphics = (List<String>)salesforceBulkProperties.upsertRelationTable.polymorphic.getValue();
            		List<String> lookupFieldModuleNames = (List<String>)salesforceBulkProperties.upsertRelationTable.lookupFieldModuleName.getValue();
            		List<String> lookupFieldNames = (List<String>)salesforceBulkProperties.upsertRelationTable.lookupFieldName.getValue();
            		List<String> externalIdFromLookupFields = (List<String>)salesforceBulkProperties.upsertRelationTable.lookupFieldExternalIdName.getValue();
            		
            		if("true".equals(polymorphics.get(index))) {
            			sbuilder.append(lookupFieldModuleNames.get(index)).append(":");
            		}
            		sbuilder.append(lookupFieldNames.get(index)).append(".").append(externalIdFromLookupFields.get(index));
            		header = sbuilder.toString();
            		sbuilder.setLength(0);
            	}
        	}
        	
            headers.add(header);
        }
        return headers.toArray(new String[headers.size()]);
	}
	
	private int getIndex(List<String> columnNames, String columnName) {
		if(columnNames == null) {
			return -1;
		}
		return columnNames.indexOf(columnName);
	}

	@Override
	public List<String> getValues(Object datum) {
		IndexedRecord input = getFactory(datum).convertToAvro(datum);
		List<String> values = new ArrayList<String>();
		for (Schema.Field f : input.getSchema().getFields()) {
			if (input.get(f.pos()) == null) {
				if (((TSalesforceOutputBulkProperties) bulkProperties).ignoreNull.getBooleanValue()) {
					values.add("");
				} else {
					values.add("#N/A");
				}
			} else {
				values.add(String.valueOf(input.get(f.pos())));
			}
		}
		return values;
	}
}
@@ -567,16 +567,68 @@
 ntime;%0A%0A
+import java.util.ArrayList;%0Aimport java.util.List;%0A%0A
 import o
@@ -650,16 +650,16 @@
 Schema;%0A
-
 import o
@@ -1105,60 +1105,8 @@
 s;%0A%0A
-import java.util.ArrayList;%0Aimport java.util.List;%0A%0A
 /**%0A
@@ -2176,43 +2176,23 @@
   %09%09
-int index = getIndex((List%3CString%3E)
+Object value = 
 sale
@@ -2253,16 +2253,122 @@
 tValue()
+;%0A        %09%09if(value!=null && value instanceof List) %7B%0A%09        %09%09int index = getIndex((List%3CString%3E)value
 , header
@@ -2362,32 +2362,33 @@
 value, header);%0A
+%09
             %09if(
@@ -2397,24 +2397,25 @@
 dex %3E -1) %7B%0A
+%09
             
@@ -2519,32 +2519,33 @@
 hic.getValue();%0A
+%09
             %09%09Li
@@ -2665,32 +2665,33 @@
 ame.getValue();%0A
+%09
             %09%09Li
@@ -2799,32 +2799,33 @@
 ame.getValue();%0A
+%09
             %09%09Li
@@ -2953,32 +2953,33 @@
 ame.getValue();%0A
+%09
             %09%09%0A 
@@ -2977,16 +2977,17 @@
      %09%09%0A
+%09
         
@@ -3033,24 +3033,25 @@
 (index))) %7B%0A
+%09
             
@@ -3117,16 +3117,17 @@
 d(%22:%22);%0A
+%09
         
@@ -3134,16 +3134,17 @@
     %09%09%7D%0A
+%09
         
@@ -3253,16 +3253,17 @@
 ndex));%0A
+%09
         
@@ -3290,32 +3290,33 @@
 der.toString();%0A
+%09
             %09%09sb
@@ -3328,32 +3328,33 @@
 r.setLength(0);%0A
+%09
             %09%7D%0A 
@@ -3344,32 +3344,44 @@
 %09            %09%7D%0A
+        %09%09%7D%0A
         %09%7D%0A     

fix upsert with relation tables and dynamic schema for salesforce bulk components https://jira.talendforge.org/browse/TUP-4480