Salesforce Bulk API 和 Apache Camel Salesforce 组件

Question

我必须将我的项目与 salesforce 集成。更具体地说，我需要能够将一些 salesforce.com 数据批量导入我们的系统，例如 Accounts。我已经做了足够多的研发来得出结论，SF Bulk API 是要走的路。我们的项目是一个 Java 系统，我们已经在使用 Camel，它恰好有一个 Salesforce 组件，看起来工作得很好。

这个问题既是关于 Salesforce 的，也是关于 Camel 的。

SF bulk api 的本质是它们是异步的。这意味着，我提交一份工作，然后我必须轮询状态，然后状态在某个时候变为 COMPLETED，然后我可以提取结果。到目前为止一切顺利，但我需要了解的是，从 Camel 的角度来看，我如何在提交工作后启动此投票？正如您在我的示例中看到的那样，我使用了延迟器，但这不是最理想的——有些批次可能需要几个小时才能准备好。

此外，如您所见，我坚持使用 jobInfo 变量。那是因为它似乎没有被getQueryResultIds传递。骆驼有什么办法可以在交易所被推下管道时保留交易所的某些部分？

这是我为批量导入帐户编写的代码（我取出了一些专有部分）：

final BatchInfo[] bi = new BatchInfo[1];
from("timer://runOnce?repeatCount=1&delay=10")
.process(new Processor() {

    public void process(Exchange exchange) throws Exception
    {
        JobInfo jobInfo = new JobInfo();
        jobInfo.setContentType(ContentType.CSV);
        jobInfo.setOperation(OperationEnum.QUERY);
        jobInfo.setObject("Account");
        jobInfo.setConcurrencyMode(ConcurrencyModeEnum.PARALLEL);
        exchange.getOut().setBody(jobInfo);
    }

})
.to("salesforce:createJob")
.to("salesforce:createBatchQuery?sObjectQuery=select Id,Name,Type,BillingCity,BillingState," +
        "BillingPostalCode,BillingCountry,Phone from Account")
.delay(10000)
.to("salesforce:getBatch")
.process(new Processor() {

    public void process(Exchange exchange) throws Exception
    {
        BatchInfo batchInfo = exchange.getIn().getBody(BatchInfo.class);
        bi[0] = batchInfo;
        exchange.getOut().setBody(batchInfo);

    }

})
.to("salesforce:getQueryResultIds")
.process(new Processor() {

    public void process(Exchange exchange) throws Exception
    {
        if (exchange.getException() != null)
        {
            exchange.getException().printStackTrace();
        }
        System.out.println(exchange.getIn().getBody());
        Collection resultIds = exchange.getIn().getBody(Collection.class);
        String resultId = (String) resultIds.iterator().next();
        exchange.getOut().setHeader(SalesforceEndpointConfig.RESULT_ID, resultId);
        exchange.getOut().setHeader(SalesforceEndpointConfig.JOB_ID, bi[0].getJobId());
        exchange.getOut().setHeader(SalesforceEndpointConfig.BATCH_ID, bi[0].getId());
        exchange.getOut().setBody(exchange.getIn().getBody());
    }

})
.to("salesforce:getQueryResult")
.process(new Processor() {

    public void process(Exchange exchange) throws Exception
    {
        InputStream is = exchange.getIn().getBody(InputStream.class);
        CachedOutputStream cos = new CachedOutputStream(exchange);
        BufferedReader r = new BufferedReader(new InputStreamReader(is));
        PrintWriter pw = new PrintWriter(cos);
        boolean header = false;
        String line = null;
        com.google.gson.stream.JsonWriter writer = new com.google.gson.stream.JsonWriter(pw);
        Gson gson = new Gson();
        writer.beginArray();
        while ((line = r.readLine()) != null)
        {
            if (!header)
            {
                header = true;
                continue;
            }
            String[] split = line.split(",");
            JsonObject account = new JsonObject();
            JsonObject headers = new JsonObject();
            JsonObject data = new JsonObject();
            account.add("headers", headers);
            headers.addProperty("category", "entity");
            headers.addProperty("type", "Account");
            account.add("data", data);

            data.addProperty("accountId", split[0].substring(1, split[0].length() - 1));
            data.addProperty("name", split[1].substring(1, split[1].length() - 1));
            data.addProperty("accountType", split[2].substring(1, split[2].length() - 1));
            gson.toJson(account, writer);
        }
        writer.endArray();
        writer.flush();
        exchange.getOut().setBody(cos.getInputStream());

    }

})
.to("stream:out");

如您所见，我在这里所做的是从 SF 批量导入数据并将其转换为 JSON 并将其保存到文件中。为了这个问题，我简化了一些事情。

我知道，对于现实世界的场景，我将不得不用 SEDA 或 VM 生产者替换路由的“来自计时器”部分。

非常感谢任何帮助。

问候， 奥列格

Answer 1

我在下面找到了这个解决方案：

public Predicate stopLoopPredicate() {
    Predicate stopLoop = new Predicate() {
        @Override
        public boolean matches(Exchange exchange) {
             return exchange.getIn().getBody() != null && !exchange.getIn().getHeader("stateBatch").toString().equals("COMPLETED");
        }
    };
    return stopLoop;
}

.......................

    .to("salesforce:createJob")
    .to("salesforce:createBatchQuery?sObjectQuery=select Id,FirstName,LastName,Email,MobilePhone from Contact")
    .to("salesforce:getBatch")
      
    .process(new Processor() {
        public void process(Exchange exchange) throws Exception {
            BatchInfo batchInfo = exchange.getIn().getBody(BatchInfo.class);
            bi[0] = batchInfo;
            BatchStateEnum stateBatch = batchInfo.getState();
            exchange.getMessage().setHeader("stateBatch", stateBatch.toString());
            exchange.getMessage().setBody(batchInfo);
        }
    })

    .loopDoWhile(stopLoopPredicate())

    .to("salesforce:getBatch")

    .process(new Processor() {
        public void process(Exchange exchange) throws Exception {
            BatchInfo batchInfo = exchange.getIn().getBody(BatchInfo.class);
            bi[0] = batchInfo;
            BatchStateEnum stateBatch = batchInfo.getState();
            exchange.getMessage().setHeader("stateBatch", stateBatch.toString());
            exchange.getMessage().setBody(batchInfo);
        }
    })

    .end()

......................

Answer 2

最后，我决定不为此使用 Camel Salesforce 连接器，这很遗憾。缺乏该库的文档以及不太清楚的源代码 cmets 让我担心我可能会花费更多时间让该库正常工作而不是实际实施我的项目。

Answer 3

等待实用程序会帮助您解决这个问题吗？ https://code.google.com/p/awaitility

Answer 4

所以，我想我想首先质疑您对批量 api 的假设。

bulk api 在这里可能比它的价值更麻烦。

您可以使用其余的 api，以及一些巧妙的循环来提取您的记录。我们在这里讨论了多少条记录？

此外，您可能会考虑使用平面文件 (csv) 导出和加载，甚至是 jitterbit/dataloader 情况，并让系统从由此生成的 .csv 加载。

尽管 Camel 与 Apache 基金会有关联，但我不得不去查找它是什么/做了什么。我喜欢认为自己是相当典型的，所以不幸的是，这可能意味着你很难在这里找到骆驼专业知识。