public SalesforceUpdateSObjectComponent() {
    super("salesforce-update-sobject",SalesforceUpdateSObjectComponent.class.getName());

    // set sObjectId header
    setBeforeProducer(exchange -> {

        // parse input json and extract Id field
        final ObjectMapper mapper = JsonUtils.createObjectMapper();
        final JsonNode node = mapper.readTree(exchange.getIn().getBody(String.class));

        final JsonNode sObjectId = node.get("Id");
        if (sObjectId == null) {
            exchange.setException(new SalesforceException("Missing field Id",404));
        } else {
            exchange.getIn().setHeader(SalesforceEndpointConfig.soBJECT_ID,sObjectId.asText());
        }

        clearBaseFields((ObjectNode) node);

        // update input json
        exchange.getIn().setBody(mapper.writeValueAsstring(node));
    });
}

public SalesforceProducer(SalesforceEndpoint endpoint) throws SalesforceException {
    super(endpoint);

    final SalesforceEndpointConfig endpointConfig = endpoint.getConfiguration();
    final PayloadFormat payloadFormat = endpointConfig.getFormat();

    // check if its a Bulk Operation
    final OperationName operationName = endpoint.getoperationName();
    if (isBulkOperation(operationName)) {
        processor = new BulkApiProcessor(endpoint);
    } else if (isAnalyticsOperation(operationName)) {
        processor = new AnalyticsApiProcessor(endpoint);
    } else {
        // create an appropriate processor
        if (payloadFormat == PayloadFormat.JSON) {
            // create a JSON exchange processor
            processor = new JsonRestProcessor(endpoint);
        } else {
            processor = new XmlRestProcessor(endpoint);
        }
    }
}

private void processGetJob(final Exchange exchange,final AsyncCallback callback) throws SalesforceException {
    JobInfo jobBody;
    jobBody = exchange.getIn().getBody(JobInfo.class);
    String jobId;
    if (jobBody != null) {
        jobId = jobBody.getId();
    } else {
        jobId = getParameter(JOB_ID,exchange,USE_BODY,NOT_OPTIONAL);
    }
    bulkClient.getJob(jobId,new Bulkapiclient.JobInfoResponseCallback() {
        @Override
        public void onResponse(JobInfo jobInfo,SalesforceException ex) {
            processResponse(exchange,jobInfo,ex,callback);
        }
    });
}

private void processCloseJob(final Exchange exchange,final AsyncCallback callback) throws SalesforceException {
    JobInfo jobBody;
    String jobId;
    jobBody = exchange.getIn().getBody(JobInfo.class);
    if (jobBody != null) {
        jobId = jobBody.getId();
    } else {
        jobId = getParameter(JOB_ID,NOT_OPTIONAL);
    }
    bulkClient.closeJob(jobId,callback);
        }
    });
}

private void processAbortJob(final Exchange exchange,NOT_OPTIONAL);
    }
    bulkClient.abortJob(jobId,callback);
        }
    });
}

private void processCreateBatch(final Exchange exchange,final AsyncCallback callback) throws SalesforceException {
    String jobId;
    // since request is in the body,use headers or endpoint params
    ContentType contentType = ContentType.fromValue(
            getParameter(CONTENT_TYPE,IGnorE_BODY,NOT_OPTIONAL));
    jobId = getParameter(JOB_ID,NOT_OPTIONAL);

    InputStream request;
    try {
        request = exchange.getIn().getMandatoryBody(InputStream.class);
    } catch (CamelException e) {
        String msg = "Error preparing batch request: " + e.getMessage();
        throw new SalesforceException(msg,e);
    }

    bulkClient.createBatch(request,jobId,contentType,new Bulkapiclient.BatchInfoResponseCallback() {
        @Override
        public void onResponse(BatchInfo batchInfo,batchInfo,callback);
        }
    });
}

private void processGetBatch(final Exchange exchange,final AsyncCallback callback) throws SalesforceException {
    String jobId;
    BatchInfo batchBody = exchange.getIn().getBody(BatchInfo.class);
    String batchId;
    if (batchBody != null) {
        jobId = batchBody.getJobId();
        batchId = batchBody.getId();
    } else {
        jobId = getParameter(JOB_ID,NOT_OPTIONAL);
        batchId = getParameter(BATCH_ID,NOT_OPTIONAL);
    }
    bulkClient.getBatch(jobId,batchId,callback);
        }
    });
}

private void processGetAllBatches(final Exchange exchange,NOT_OPTIONAL);
    }
    bulkClient.getAllBatches(jobId,new Bulkapiclient.BatchInfoListResponseCallback() {
        @Override
        public void onResponse(List<BatchInfo> batchInfoList,batchInfoList,callback);
        }
    });
}

private void processCreateBatchQuery(final Exchange exchange,final AsyncCallback callback) throws SalesforceException {
    JobInfo jobBody;
    String jobId;
    ContentType contentType;
    jobBody = exchange.getIn().getBody(JobInfo.class);
    String soqlQuery;
    if (jobBody != null) {
        jobId = jobBody.getId();
        contentType = jobBody.getContentType();
        // use SOQL query from header or endpoint config
        soqlQuery = getParameter(SOBJECT_QUERY,NOT_OPTIONAL);
    } else {
        jobId = getParameter(JOB_ID,NOT_OPTIONAL);
        contentType = ContentType.fromValue(
                getParameter(CONTENT_TYPE,NOT_OPTIONAL));
        // reuse SOBJECT_QUERY property
        soqlQuery = getParameter(SOBJECT_QUERY,NOT_OPTIONAL);
    }
    bulkClient.createBatchQuery(jobId,soqlQuery,new Bulkapiclient.BatchInfoResponseCallback() {
                @Override
                public void onResponse(BatchInfo batchInfo,SalesforceException ex) {
                    processResponse(exchange,callback);
                }
            });
}

private void processGetQueryResultIds(final Exchange exchange,final AsyncCallback callback) throws SalesforceException {
    String jobId;
    BatchInfo batchBody;
    String batchId;
    batchBody = exchange.getIn().getBody(BatchInfo.class);
    if (batchBody != null) {
        jobId = batchBody.getJobId();
        batchId = batchBody.getId();
    } else {
        jobId = getParameter(JOB_ID,NOT_OPTIONAL);
    }
    bulkClient.getQueryResultIds(jobId,new Bulkapiclient.QueryResultIdsCallback() {
        @Override
        public void onResponse(List<String> ids,ids,callback);
        }
    });
}

private void processCreateSobject(final Exchange exchange,final AsyncCallback callback) throws SalesforceException {
    String sObjectName;
    // determine parameters from input AbstractSObject
    AbstractSObjectBase sObjectBase = exchange.getIn().getBody(AbstractSObjectBase.class);
    if (sObjectBase != null) {
        sObjectName = sObjectBase.getClass().getSimpleName();
    } else {
        sObjectName = getParameter(SOBJECT_NAME,NOT_OPTIONAL);
    }

    restClient.createSObject(sObjectName,getRequestStream(exchange),new RestClient.ResponseCallback() {
            @Override
            public void onResponse(InputStream response,SalesforceException exception) {
                processResponse(exchange,response,exception,callback);
            }
        });
}

private void processUpdateSobject(final Exchange exchange,final AsyncCallback callback) throws SalesforceException {
    String sObjectName;
    // determine parameters from input AbstractSObject
    final AbstractSObjectBase sObjectBase = exchange.getIn().getBody(AbstractSObjectBase.class);
    String sObjectId;
    if (sObjectBase != null) {
        sObjectName = sObjectBase.getClass().getSimpleName();
        // remember the sObject Id
        sObjectId = sObjectBase.getId();
        // clear base object fields,which cannot be updated
        sObjectBase.clearBaseFields();
    } else {
        sObjectName = getParameter(SOBJECT_NAME,NOT_OPTIONAL);
        sObjectId = getParameter(SOBJECT_ID,NOT_OPTIONAL);
    }

    final String finalsObjectId = sObjectId;
    restClient.updateSObject(sObjectName,sObjectId,callback);
                restoreFields(exchange,sObjectBase,finalsObjectId,null,null);
            }
        });
}

private void processDeleteSobject(final Exchange exchange,final AsyncCallback callback) throws SalesforceException {
    String sObjectName;
    // determine parameters from input AbstractSObject
    final AbstractSObjectBase sObjectBase = exchange.getIn().getBody(AbstractSObjectBase.class);
    String sObjectIdValue;
    if (sObjectBase != null) {
        sObjectName = sObjectBase.getClass().getSimpleName();
        sObjectIdValue = sObjectBase.getId();
    } else {
        sObjectName = getParameter(SOBJECT_NAME,NOT_OPTIONAL);
        sObjectIdValue = getParameter(SOBJECT_ID,NOT_OPTIONAL);
    }
    final String sObjectId = sObjectIdValue;

    restClient.deleteSObject(sObjectName,new RestClient.ResponseCallback() {
        @Override
        public void onResponse(InputStream response,SalesforceException exception) {
            processResponse(exchange,callback);
            restoreFields(exchange,null);
        }
    });
}

private void restoreFields(Exchange exchange,AbstractSObjectBase sObjectBase,String sObjectId,String sObjectExtIdName,Object oldValue) {
    // restore fields
    if (sObjectBase != null) {
        // restore the Id if it was cleared
        if (sObjectId != null) {
            sObjectBase.setId(sObjectId);
        }
        // restore the external id if it was cleared
        if (sObjectExtIdName != null && oldValue != null) {
            try {
                setPropertyValue(sObjectBase,sObjectExtIdName,oldValue);
            } catch (SalesforceException e) {
                // YES,the exchange may fail if the property cannot be reset!!!
                exchange.setException(e);
            }
        }
    }
}

private void setResponseClass(Exchange exchange,String sObjectName) throws SalesforceException {
    Class<?> sObjectClass;

    if (sObjectName != null) {
        // lookup class from class map
        sObjectClass = classMap.get(sObjectName);
        if (null == sObjectClass) {
            throw new SalesforceException(String.format("No class found for SObject %s",sObjectName),null);
        }

    } else {

        // use custom response class property
        final String className = getParameter(SOBJECT_CLASS,NOT_OPTIONAL);
        try {
            sObjectClass = endpoint.getComponent().getCamelContext()
                    .getClassResolver().resolveMandatoryClass(className);
        } catch (ClassNotFoundException e) {
            throw new SalesforceException(
                    String.format("SObject class not found %s,%s",className,e.getMessage()),e);
        }
    }
    exchange.setProperty(RESPONSE_CLASS,sObjectClass);
}

private void processExecuteSyncReport(final Exchange exchange,final AsyncCallback callback) throws SalesforceException {

        String reportId;
        final Boolean includeDetails = getParameter(INCLUDE_DETAILS,IS_OPTIONAL,Boolean.class);

        // try getting report Metadata from body first
        ReportMetadata reportMetadata = exchange.getIn().getBody(ReportMetadata.class);
        if (reportMetadata != null) {
            reportId = reportMetadata.getId();
            if (reportId == null) {
                reportId = getParameter(REPORT_ID,NOT_OPTIONAL);
            }
        } else {
            reportId = getParameter(REPORT_ID,NOT_OPTIONAL);
            reportMetadata = getParameter(REPORT_MetaDATA,ReportMetadata.class);
        }

        analyticsClient.executeSyncReport(reportId,includeDetails,reportMetadata,new Analyticsapiclient.ReportResultsResponseCallback() {
                @Override
                public void onResponse(AbstractReportResultsBase reportResults,reportResults,callback);
                }
            });
    }

private void processExecuteAsyncReport(final Exchange exchange,ReportMetadata.class);
        }

        analyticsClient.executeAsyncReport(reportId,new Analyticsapiclient.ReportInstanceResponseCallback() {
                @Override
                public void onResponse(ReportInstance reportInstance,reportInstance,callback);
                }
            });
    }

@Override
public void getJob(String jobId,final JobInfoResponseCallback callback) {
    final Request get = getRequest(HttpMethod.GET,jobUrl(jobId));

    // make the call and parse the result
    doHttpRequest(get,new ClientResponseCallback() {
        @Override
        public void onResponse(InputStream response,SalesforceException ex) {
            JobInfo value = null;
            try {
                value = unmarshalResponse(response,get,JobInfo.class);
            } catch (SalesforceException e) {
                ex = e;
            }
            callback.onResponse(value,ex);
        }
    });
}

@Override
public void createBatch(InputStream batchStream,String jobId,ContentType contentTypeEnum,final BatchInfoResponseCallback callback) {
    final Request post = getRequest(HttpMethod.POST,batchUrl(jobId,null));
    post.content(new InputStreamContentProvider(batchStream));
    post.header(HttpHeader.CONTENT_TYPE,getContentType(contentTypeEnum) + ";charset=" + StringUtil.__UTF8);

    // make the call and parse the result
    doHttpRequest(post,SalesforceException ex) {
            BatchInfo value = null;
            try {
                value = unmarshalResponse(response,post,BatchInfo.class);
            } catch (SalesforceException e) {
                ex = e;
            }
            callback.onResponse(value,ex);
        }
    });
}

@Override
public void getBatch(String jobId,String batchId,final BatchInfoResponseCallback callback) {
    final Request get = getRequest(HttpMethod.GET,batchId));

    // make the call and parse the result
    doHttpRequest(get,ex);
        }
    });
}

@Override
public void getAllBatches(String jobId,final BatchInfoListResponseCallback callback) {
    final Request get = getRequest(HttpMethod.GET,null));

    // make the call and parse the result
    doHttpRequest(get,SalesforceException ex) {
            BatchInfoList value = null;
            try {
                value = unmarshalResponse(response,BatchInfoList.class);
            } catch (SalesforceException e) {
                ex = e;
            }
            callback.onResponse(value != null ? value.getBatchInfo() : null,ex);
        }
    });
}

@Override
public void getQueryResultIds(String jobId,final QueryResultIdsCallback callback) {
    final Request get = getRequest(HttpMethod.GET,batchResultUrl(jobId,SalesforceException ex) {
            QueryResultList value = null;
            try {
                value = unmarshalResponse(response,QueryResultList.class);
            } catch (SalesforceException e) {
                ex = e;
            }
            callback.onResponse(value != null ? Collections.unmodifiableList(value.getResult()) : null,ex);
        }
    });
}

@Override
protected SalesforceException createRestException(Response response,InputStream responseContent) {
    // this must be of type Error
    try {
        final Error error = unmarshalResponse(responseContent,response.getRequest(),Error.class);

        final RestError restError = new RestError();
        restError.setErrorCode(error.getExceptionCode());
        restError.setMessage(error.getExceptionMessage());

        return new SalesforceException(Arrays.asList(restError),response.getStatus());
    } catch (SalesforceException e) {
        String msg = "Error un-marshaling Salesforce Error: " + e.getMessage();
        return new SalesforceException(msg,e);
    }
}

@Override
public void query(String soqlQuery,ResponseCallback callback) {
    try {

        String encodedQuery = urlEncode(soqlQuery);
        final Request get = getRequest(HttpMethod.GET,versionUrl() + "query/?q=" + encodedQuery);

        // requires authorization token
        setAccesstoken(get);

        doHttpRequest(get,new DelegatingClientCallback(callback));

    } catch (UnsupportedEncodingException e) {
        String msg = "Unexpected error: " + e.getMessage();
        callback.onResponse(null,new SalesforceException(msg,e));
    }
}

@Override
public void search(String soslQuery,ResponseCallback callback) {
    try {

        String encodedQuery = urlEncode(soslQuery);
        final Request get = getRequest(HttpMethod.GET,versionUrl() + "search/?q=" + encodedQuery);

        // requires authorization token
        setAccesstoken(get);

        doHttpRequest(get,e));
    }
}

@Override
public void getRecentReports(final RecentReportsResponseCallback callback) {

    final Request request = getRequest(HttpMethod.GET,reportsUrl());

    doHttpRequest(request,new ClientResponseCallback() {
        @Override
        @SuppressWarnings("unchecked")
        public void onResponse(InputStream response,SalesforceException ex) {
            List<RecentReport> recentReports = null;
            if (response != null) {
                try {
                    recentReports = unmarshalResponse(response,request,new TypeReference<List<RecentReport>>() {
                        }
                    );
                } catch (SalesforceException e) {
                    ex = e;
                }
            }
            callback.onResponse(recentReports,ex);
        }
    });
}

@Override
public void getReportDescription(String reportId,final ReportDescriptionResponseCallback callback) {

    final Request request = getRequest(HttpMethod.GET,reportsDescribeUrl(reportId));

    doHttpRequest(request,SalesforceException ex) {
            ReportDescription reportDescription = null;
            try {
                reportDescription = unmarshalResponse(response,ReportDescription.class);
            } catch (SalesforceException e) {
                ex = e;
            }
            callback.onResponse(reportDescription,ex);
        }
    });
}

@Override
public void getReportInstances(String reportId,final ReportInstanceListResponseCallback callback) {

    final Request request = getRequest(HttpMethod.GET,reportInstancesUrl(reportId));

    doHttpRequest(request,SalesforceException ex) {
            List<ReportInstance> reportInstances = null;
            if (response != null) {
                try {
                    reportInstances = unmarshalResponse(response,new TypeReference<List<ReportInstance>>() {
                        }
                    );
                } catch (SalesforceException e) {
                    ex = e;
                }
            }
            callback.onResponse(reportInstances,ex);
        }
    });
}

@Override
public void getReportResults(String reportId,String instanceId,final ReportResultsResponseCallback callback) {

    final Request request = getRequest(HttpMethod.GET,reportInstancesUrl(reportId,instanceId));

    doHttpRequest(request,SalesforceException ex) {
            AsyncReportResults reportResults = null;
            try {
                reportResults = unmarshalResponse(response,AsyncReportResults.class);
            } catch (SalesforceException e) {
                ex = e;
            }
            callback.onResponse(reportResults,ex);
        }
    });
}

private <T> T unmarshalResponse(InputStream response,Request request,Class<T> responseClass)
    throws SalesforceException {

    if (response == null) {
        return null;
    }

    try {
        return objectMapper.readValue(response,responseClass);
    } catch (IOException e) {
        throw new SalesforceException(
            String.format("Error unmarshaling response {%s:%s} : %s",request.getmethod(),request.getURI(),e);
    }
}

@Test
public void testApexCall() throws Exception {
    try {
        doTestApexCall("");
        doTestApexCall("Xml");
    } catch (CamelExecutionException e) {
        if (e.getCause() instanceof SalesforceException) {
            SalesforceException cause = (SalesforceException) e.getCause();
            if (cause.getStatusCode() == HttpStatus.NOT_FOUND_404) {
                LOG.error("Make sure test REST resource MerchandiseRestResource.apxc has been loaded: "
                    + e.getMessage());
            }
        }
        throw e;
    }
}

@Override
public void process(final Exchange exchange) throws Exception {
    // parse input json and extract Id field
    final Message in = exchange.getIn();
    final String body = in.getBody(String.class);

    if (body == null) {
        return;
    }

    final ObjectNode node = (ObjectNode) MAPPER.readTree(body);

    final String idPropertyName = determineIdProperty(exchange);

    final JsonNode idProperty = node.remove(idPropertyName);
    if (idProperty == null) {
        exchange.setException(
            new SalesforceException("Missing option value for Id or " + SalesforceEndpointConfig.soBJECT_EXT_ID_NAME,404));

        return;
    }

    final String idValue = idProperty.textValue();
    if ("Id".equals(idPropertyName)) {
        in.setHeader(SalesforceEndpointConfig.soBJECT_ID,idValue);
    } else {
        in.setHeader(SalesforceEndpointConfig.soBJECT_EXT_ID_VALUE,idValue);
    }

    // base fields are not allowed to be updated
    clearBaseFields(node);

    // update input json
    in.setBody(MAPPER.writeValueAsstring(node));
}

@Override
protected void doStop() throws Exception {
    if (classMap != null) {
        classMap.clear();
    }

    try {
        if (subscriptionHelper != null) {
            // shutdown all streaming connections
            // note that this is done in the component,and not in consumer
            ServiceHelper.stopService(subscriptionHelper);
            subscriptionHelper = null;
        }
        if (session != null && session.getAccesstoken() != null) {
            try {
                // logout of Salesforce
                ServiceHelper.stopService(session);
            } catch (SalesforceException ignored) {
            }
        }
    } finally {
        if (httpClient != null) {
            // shutdown http client connections
            httpClient.stop();
            // destroy http client if it was created by the component
            if (config.getHttpClient() == null) {
                httpClient.destroy();
            }
            httpClient = null;
        }
    }
}

/**
 * Gets value for a parameter from header,endpoint config,or exchange body (optional).
 *
 * @param exchange          exchange to inspect
 * @param convertInBody     converts In body to parameterClass value if true
 * @param propName          name of property
 * @param optional          if {@code true} returns null,otherwise throws RestException
 * @param parameterClass    parameter type
 * @return value of property,or {@code null} for optional parameters if not found.
 * @throws org.apache.camel.component.salesforce.api.SalesforceException
 *          if the property can't be found or on conversion errors.
 */
protected final <T> T getParameter(String propName,Exchange exchange,boolean convertInBody,boolean optional,Class<T> parameterClass) throws SalesforceException {

    final Message in = exchange.getIn();
    T propValue = in.getHeader(propName,parameterClass);

    if (propValue == null) {
        // check if type conversion Failed
        if (in.getHeader(propName) != null) {
            throw new IllegalArgumentException("Header " + propName
                + " Could not be converted to type " + parameterClass.getName());
        }

        final Object value = endpointConfigMap.get(propName);

        if (value == null || parameterClass.isinstance(value)) {
            propValue = parameterClass.cast(value);
        } else {

            try {
                propValue = exchange.getContext().getTypeConverter().mandatoryConvertTo(parameterClass,value);
            } catch (NoTypeConversionAvailableException e) {
                throw new SalesforceException(e);
            }
        }
    }

    propValue = (propValue == null && convertInBody) ? in.getBody(parameterClass) : propValue;

    // error if property was not set
    if (propValue == null && !optional) {
        String msg = "Missing property " + propName
            + (convertInBody ? ",message body Could not be converted to type " + parameterClass.getName() : "");
        throw new SalesforceException(msg,null);
    }

    return propValue;
}

private void processCreateJob(final Exchange exchange,final AsyncCallback callback) throws InvalidPayloadException {
    JobInfo jobBody = exchange.getIn().getMandatoryBody(JobInfo.class);
    bulkClient.createJob(jobBody,callback);
        }
    });
}

private void processResponse(Exchange exchange,Object body,SalesforceException ex,AsyncCallback callback) {
    final Message out = exchange.getout();
    if (ex != null) {
        exchange.setException(ex);
    } else {
        out.setBody(body);
    }

    // copy headers and attachments
    out.getHeaders().putAll(exchange.getIn().getHeaders());
    out.getAttachments().putAll(exchange.getIn().getAttachments());

    // signal exchange completion
    callback.done(false);
}

private void processGetVersions(final Exchange exchange,final AsyncCallback callback) {
    restClient.getVersions(new RestClient.ResponseCallback() {
        @Override
        public void onResponse(InputStream response,SalesforceException exception) {
            // process response entity and create out message
            processResponse(exchange,callback);
        }
    });
}

private void processGetResources(final Exchange exchange,final AsyncCallback callback) {
    restClient.getResources(new RestClient.ResponseCallback() {
        @Override
        public void onResponse(InputStream response,callback);
        }
    });
}

private void processGetGlobalObjects(final Exchange exchange,final AsyncCallback callback) {
    restClient.getGlobalObjects(new RestClient.ResponseCallback() {
        @Override
        public void onResponse(InputStream response,callback);
        }
    });
}

private void processGetBasicInfo(final Exchange exchange,final AsyncCallback callback) throws SalesforceException {
    String sObjectName = getParameter(SOBJECT_NAME,NOT_OPTIONAL);
    restClient.getBasicInfo(sObjectName,callback);
        }
    });
}