Mule json-to-object-transformer 和 Dataweaver Transformer 异常答案

【问题标题】：Mule json-to-object-transformer and Dataweaver Transformer exceptionMule json-to-object-transformer 和 Dataweaver Transformer 异常
【发布时间】：2018-01-18 21:30:42
【问题描述】：

我遇到了一个 Mule Dataweave Transformer 获取问题

INFO  2017-08-10 15:05:52,787 [amqpReceiver.02] org.mule.api.processor.LoggerMessageProcessor: Authorisation Changed Event received. authorisation id: 1
INFO  2017-08-10 15:05:57,844 [amqpReceiver.02] org.mule.api.processor.LoggerMessageProcessor: type before class com.mulesoft.weave.reader.ByteArraySeekableStream
INFO  2017-08-10 15:06:02,058 [amqpReceiver.02] org.mule.api.processor.LoggerMessageProcessor: type after: class com.mulesoft.weave.reader.ByteArraySeekableStream
INFO  2017-08-10 15:06:13,309 [amqpReceiver.02] org.mule.api.processor.LoggerMessageProcessor: Org Name :- Id:- Org Name 3
ERROR 2017-08-10 15:06:15,520 [amqpReceiver.02] org.mule.exception.RollbackMessagingExceptionStrategy: 
********************************************************************************
Message               : Exception while executing: 
{
^
Unexpected end-of-input at index 0 (line 1, position 1), expected JSON Value
Payload               : com.mulesoft.weave.reader.ByteArraySeekableStream@a622be8
Element               : /subscribe-to-changeFlow/processors/8 @ voa-009-mule-authorisation-search-api:test.xml:60 (Final Transform Message)
Element XML           : <dw:transform-message doc:name="Final Transform Message" metadata:id="fa7e7663-50bb-4675-a8c8-e8cb311946a9">
                        <dw:input-payload mimeType="application/json"></dw:input-payload>
                        <dw:set-payload resource="classpath:dwl/owner-authorisation-enrichment.dwl"></dw:set-payload>
                        </dw:transform-message>
--------------------------------------------------------------------------------
Root Exception stack trace:
com.mulesoft.weave.reader.json.JsonReaderException: Unexpected end-of-input at index 0 (line 1, position 1), expected JSON Value
    at com.mulesoft.weave.reader.json.JsonTokenizer.fail(JsonTokenizer.scala:193)
    at com.mulesoft.weave.reader.json.JsonTokenizer.readValue(JsonTokenizer.scala:49)
    at com.mulesoft.weave.reader.json.JsonTokenizer.tokens(JsonTokenizer.scala:16)
    at com.mulesoft.weave.reader.json.JsonReader.root(JsonReader.scala:17)
    at com.mulesoft.weave.mule.reader.ReusableReader.root(ReusableReader.scala:12)
    at com.mulesoft.weave.engine.EvaluationContext$InternalExecutionContext$$anonfun$3.apply(EvaluationContext.scala:106)

流接收一个 ID 作为消息负载
进行 HTTP 调用以获取 ID 的一些 JSON
执行初始 Dataweave 转换以将 JSON 转换为新的 JSON 架构

为了从 JSON 中提取一些值，我使用 json:json-to-object-transformer 但我已将其放入针对 flowVar 的丰富器中，以允许转换为地图并保持原始有效负载完整

    <enricher target="#[flowVars.ownerAuthorisationJsonMap]"  doc:name="Copy to JSON map"> 
         <json:json-to-object-transformer returnClass="java.util.HashMap" doc:name="JSON to Object"/>
    </enricher>

然后我调用一个子流来获取一些进一步的日期，以便以后丰富。子流仅将浓缩器的结果存储在另一个 flowVar 中
第二个 Dataweave 转换器（发生异常）从第一个 Dataweave 转换器获取 JSON 模式输出，并使用子流查找的数据丰富一些缺失值（示例显示一个，但 Map 中会有很多）。这会引发异常。

但是，如果我用嵌套的 json-to-object-transformer

注释掉 enricher

         <enricher target="#[flowVars.ownerAuthorisationJsonMap]"  doc:name="Copy to JSON map"> 
             <json:json-to-object-transformer returnClass="java.util.HashMap" doc:name="JSON to Object"/>
        </enricher>

第二个数据编织工作正常（但我无法访问 JSON 中的数据）。似乎 json-to-object-transformer 让事情变得不安，但如果我调试有效负载类型和 java 对象 id 始终保持不变。

我只使用 json-to-object-transformer 来允许 MEL 访问 JSON，因为 JSON-path 已被弃用。

<?xml version="1.0" encoding="UTF-8"?>

<amqp:endpoint name="authorisationChangeQueueDef"
    queueName="authorisation-change-queue" queueDurable="true"
    exchangeName="authorisation-exchange" exchangeType="direct" exchangeDurable="true"
    connector-ref="AMQP_Connector"
    routingKey="authorisation-change-queue-routing-key" doc:name="authorisationChangeQueueDef">
       <properties>
            <spring:entry key="amqp-queue.x-dead-letter-exchange"    value="authorisation-exchange" />
            <spring:entry key="amqp-queue.x-dead-letter-routing-key" value="authorisation-change-queue-dead-routing-key" />
        </properties>
</amqp:endpoint>


<flow name="subscribe-to-changeFlow" processingStrategy="synchronous">

    <amqp:inbound-endpoint ref="authorisationChangeQueueDef" responseTimeout="10000" doc:name="authorisation-change-consumer" />

    <set-variable variableName="authorisationId" value="#[message.payloadAs(java.lang.String)]" doc:name="set authorisation id"/>

    <logger message="#['Authorisation Changed Event received. authorisation id: ' + flowVars.authorisationId]" level="INFO" doc:name="Logger"/>

    <http:request config-ref="Mule_Authorisation_Management_API_HTTP_Request_Configuration" path="#['/authorisation/' + flowVars.authorisationId]" method="GET" doc:name="HTTP REST call to Authorisation API"/>

    <dw:transform-message doc:name="Transform to basic owner-authorisation"  metadata:id="1bdd6f7e-6b4b-415b-819d-e88b9ff7f92b">
        <dw:input-payload mimeType="application/json"/>
        <dw:set-payload resource="dwl/owner-authorisation-skeleton.dwl"/>
    </dw:transform-message>

         <logger message="#['type before ' + message.dataType.type]" level="INFO" doc:name="Logger"/>

    <!-- Enricher targeting flowVar to keep original payload type intact -->            
    <enricher target="#[flowVars.ownerAuthorisationJsonMap]"  doc:name="Copy to JSON map"> 
        <json:json-to-object-transformer returnClass="java.util.HashMap" doc:name="JSON to Object"/>
    </enricher> 

         <logger message="#['type after: ' + message.dataType.type]" level="INFO" doc:name="Logger"/>

    <!--  subflow that uses flowVars.ownerAuthorisationJsonMap and MEL --> 
    <flow-ref name="enrich-Sub_Flow" doc:name="enrich-Sub_Flow"/>


    <dw:transform-message doc:name="Final Transform Message" metadata:id="fa7e7663-50bb-4675-a8c8-e8cb311946a9">
        <dw:input-payload mimeType="application/json"/>
        <dw:set-payload resource="classpath:dwl/owner-authorisation-enrichment.dwl"/>
    </dw:transform-message>


    <logger  level="INFO" doc:name="Logger all "/>


    <!-- Does a manual ACK to RabbitMQ, only do once we know all the processing is complete -->
    <amqp:acknowledge-message doc:name="AMQP-0-9 Acknowledge Message"/>


    <rollback-exception-strategy maxRedeliveryAttempts="3" doc:name="Rollback Exception Strategy">
        <logger message="#['REQUESTING RETRY OF REQUEST... delivery-tag : ' + message.inboundProperties['delivery-tag']]" level="INFO" doc:name="Logger"/>
        <!-- re-queue message for re-tries -->
        <amqp:reject-message requeue="true" doc:name="AMQP-0-9 Reject Message" />

        <on-redelivery-attempts-exceeded>
            <logger message="REDELIVERY EXHAUSTED" level="ERROR" doc:name="Logger"/>
            <amqp:reject-message doc:name="AMQP-0-9 Reject Message" />
        </on-redelivery-attempts-exceeded>           
    </rollback-exception-strategy>

</flow>

<sub-flow name="enrich-Sub_Flow">
    <set-variable variableName="organisationId" value="#[ownerAuthorisationJsonMap.organisationId]" doc:name="Extract Organisation Id" encoding="UTF-8" mimeType="application/json"/>

    <enricher target="#[flowVars.organisationName]" source="#[payload.organisationLatestDetail.organisationName]" doc:name="Message Enricher">  
        <processor-chain doc:name="Processor Chain">
            <flow-ref name="get-organisation-details-SubFlow" doc:name="get-organisation-details-Sub_Flow" />
           <json:json-to-object-transformer returnClass="java.util.HashMap" doc:name="JSON to Object"/>
           <logger level="INFO" message="Org Name :- Id:- #[payload.organisationLatestDetail.organisationName]" doc:name="Logger"/>
       </processor-chain>
    </enricher>
</sub-flow>

<sub-flow name="get-organisation-details-SubFlow">
    <http:request config-ref="VOA_Mule_Customer_Management_API_HTTP_Request_Configuration" path="#['/organisation?organisationId='+ flowVars.organisationId]" method="GET"
        doc:name="HTTP - Get Customer details">
    </http:request>
</sub-flow>

数据编织 1

%dw 1.0
%output application/json
---
{
    authorisationId: payload.id,
    uarn: payload.uarn,
    authorisedParties: payload.parties map ((party , indexOfParty) -> {
        organisationId: party.authorisedPartyOrganisationId,
        status: party.authorisedPartyStatus,
        startDate: party.startDate
    }),
    endDate: payload.endDate,
    organisationId: payload.authorisationOwnerOrganisationId,
    propertyLinkSubmissionId: payload.submissionId,
    startDate: payload.startDate,
    status: payload.authorisationStatus
}

Dataweave 2（这也将具有来自 Java Map 的进一步映射）

%dw 1.0
%output application/json
---
{
    authorisationId: payload.authorisationId,
    authorisedParties: payload.authorisedParties map ((authorisedParty , indexOfAuthorisedParty) -> {
        organisationId: authorisedParty.organisationId,
        startDate: authorisedParty.startDate,
        status: authorisedParty.status
    }),
    endDate: payload.endDate,
    organisationId: payload.organisationId,
    organisationName: flowVars.organisationName,
    propertyLinkSubmissionId: payload.propertyLinkSubmissionId,
    startDate: payload.startDate,
    status: payload.status,
    uarn: payload.uarn
}

日志

INFO  2017-08-10 15:05:52,787 [amqpReceiver.02] org.mule.api.processor.LoggerMessageProcessor: Authorisation Changed Event received. authorisation id: 1
INFO  2017-08-10 15:05:57,844 [amqpReceiver.02] org.mule.api.processor.LoggerMessageProcessor: type before class com.mulesoft.weave.reader.ByteArraySeekableStream
INFO  2017-08-10 15:06:02,058 [amqpReceiver.02] org.mule.api.processor.LoggerMessageProcessor: type after: class com.mulesoft.weave.reader.ByteArraySeekableStream
INFO  2017-08-10 15:06:13,309 [amqpReceiver.02] org.mule.api.processor.LoggerMessageProcessor: Org Name :- Id:- Org Name 3

【问题讨论】：

标签： mule mule-el

【解决方案1】：

该问题是由 json-to-object-transformer 消耗的 json 有效负载流引起的，并且没有被重置，尽管它位于消息丰富器中。请参阅此问题 - https://www.mulesoft.org/jira/browse/MULE-10623。

最简单的解决方案是在消息丰富器之后重置流：

...
<enricher target="#[flowVars.myMar]" doc:name="Message Enricher"> 
    <json:json-to-object-transformer doc:name="JSON to Object"/>
</enricher>
<expression-component doc:name="Expression">
    <![CDATA[payload.seek(0);]]>
</expression-component>
...

或在使用消息丰富器之前将有效负载转换为字符串：

...
<object-to-string-transformer doc:name="Object to String"/>
<enricher target="#[flowVars.myMar]"  doc:name="Message Enricher"> 
...

仅供参考，我用这个更简单的流程复制了这个问题：

<flow name="flow">

    <dw:transform-message doc:name="Transform Message">
        <dw:set-payload>
<![CDATA[%dw 1.0
%output application/json
---
{
    foo:0
}]]>
        </dw:set-payload>
    </dw:transform-message>
    <enricher target="#[flowVars.myVar]" doc:name="Message Enricher"> 
        <json:json-to-object-transformer doc:name="JSON to Object"/>
    </enricher>
    <dw:transform-message doc:name="Transform Message">
        <dw:set-payload>
<![CDATA[%dw 1.0
%output application/json
---
payload
]]> 
        </dw:set-payload>
    </dw:transform-message>
</flow>

【讨论】：

感谢您的及时答复，我将重置流。顺便说一句，最好避免使用 json:path 吗？因为我确实通过避免使用 json-to-object-transformer 而使用 JSON 路径来实现它。
JSON 路径已被弃用，并将在 Mule 4 中删除，因此最好避免使用它。 forums.mulesoft.com/questions/26717/…
感谢您的回答，但是我现在发现了另一个问题，如果我们在第二个 Dataweave 转换器中使用了 flowVar，那么有效负载会被转换为地图，请参阅我的问题 stackoverflow.com/questions/45689371/…