Apache Camel + JBoss Fuse - CXFRS 组件和 bean 组件之间的路由 - IOException 流已关闭

Question

我编写了一个 CXF REST 服务，它返回配置如下的 application/json：

前景生成器：

@POST
@Consumes(MediaType.APPLICATION_JSON)
@Produces(MediaType.APPLICATION_JSON)
@Path("/consume")
public Prospect convertProspectToJson(Prospect prospect,@QueryParam("customerType") String customerType){

    //DTO logic goes here
    System.out.println("Prospect obj received.."+prospect.getProspectName());

    prospect.setCustomerType(customerType);

    return prospect;

}

此服务在部署到 JBoss Fuse 时有效。

现在我有一个骆驼路线构建器，如下所示：

public class ServiceRouter extends RouteBuilder {

/* (non-Javadoc)
 * @see org.apache.camel.builder.RouteBuilder#configure()
 */
@Override
public void configure() throws Exception {

    from("jetty://http://localhost:8182/route/prospect?bindingStyle=SimpleConsumer")
    //The Raw type to String conversion is done here. Refer the class for details
    //Processor converts the Message from byte[] to String
    .process(new ResponseToStringProcessor())
    .unmarshal("jsonDataformat") //jsonDataformat - bean in blueprint XML for converting to requested object type
    .to("cxfrs://http://localhost:8181/cxf/prospect/consume")
    /*1) The above line has the BodyType as Cxf's ResponseImpl instead of Prospect Object
    * 2) The below bean accepts type only as Prospect object and hence I get exception */
    .to("bean:prospectService?method=doSomething");

}

ProspectService：

public class ProspectService {
    public void doSomething(@Body Prospect prospect){

    System.out.println("Prospect Service got the request.."+prospect.getCustomerType());

    }
}

我在尝试路由到定义的 bean 时遇到以下异常：

org.apache.camel.InvalidPayloadException：没有可用的类型：com.karthik.bo.Prospect 但有值：org.apache.cxf.jaxrs.impl.ResponseImpl@54c96476 类型

如何解决此错误？我不知道如何将 ResponseImpl 转换为我的 POJO。来的痕迹显示：

BodyType:org.apache.cxf.jaxrs.impl.ResponseImpl, Body:{"prospect":{"customerType":"VIP","prospectId":999,"prospectName":"karthik"} }

我尝试将 ResponseImpl 转换为 String（将包含 JSON 有效负载），但我总是以 IOException 流结束。 - 以下转换代码：

public void convert(Exchange exchange){

InputStream is =  (InputStream)exchange.getIn().getBody(ResponseImpl.class).getEntity();

String s="";
try {
    s = org.apache.commons.io.IOUtils.toString(is);
    is.close();
} catch (IOException e) {
    e.printStackTrace();
}
System.out.println("in convertor..."+s);

}

---

经过多次失败尝试（在cmets中提到），最终的解决方案是编写一个FallBackCoverter如下 - 这并没有抛出Stream closed IOException，无法解释给自己。

@FallbackConverter
    public static <T> T convertTo(Class<T> type, Exchange exchange, Object value, TypeConverterRegistry registry){

        System.out.println("Checking fall backconverter");
        if (ResponseImpl.class.isAssignableFrom(value.getClass())) {
            TypeConverter tc = registry.lookup(type, ResponseImpl.class);
            if (tc == null && type.getName().equals("com.karthik.bo.Prospect")) {
                Prospect prospect=new Prospect();
                try(InputStream is =  (InputStream)((ResponseImpl)value).getEntity()) {
                    if(is!=null){
                        String s = org.apache.commons.io.IOUtils.toString(is);
                        if(s!=null && s.length()>0){
                            ObjectMapper mapper = new ObjectMapper();
                            prospect = mapper.readValue(s, Prospect.class);

                        }
                        is.close();
                    }
                } catch (IOException e) {
                    System.out.println("Exception occured"+e.getMessage());
                }
                return type.cast(prospect);
            }
        }

        return (T) Void.TYPE;
    }

我的路由器被修改为：

from("jetty://http://localhost:8182/route/prospect?bindingStyle=SimpleConsumer")
        //The Raw type to String conversion is done here. Refer the class for details
        .process(new ResponseToStringProcessor())
        .unmarshal("jsonDataformat")
        .to("cxfrs://http://localhost:8181/cxf/prospect/consume")
        .convertBodyTo(Prospect.class)
        .to("bean:prospectService?method=doSomething")
        .marshal("jsonDataformat");

---

如果不确认这是否是唯一的方法，我无法将此标记为解决方案。我的解决方案是在我在 camel-cxf 中发现一个旧的 JIRA 问题后实施的

https://issues.apache.org/jira/browse/CAMEL-3208

为此，我不得不编写一个 FallBackConvertor。

Answer 1

看起来您的呼叫可能没有通过服务类的反射被接听。我不确定您的转换类到底要做什么，但是您是否尝试过在客户端/服务器的末尾添加一个提供程序来处理 Json 到 POJO 的映射？让我发布一个对我有用的简单示例（我在蓝图中做了这个，但它应该同样适用于 DSL）。请原谅代码的状态，我还在做大量的测试，以找出哪些有效，哪些无效。

这里我有一个 cxf rs 客户端和服务器：

<cxf:rsClient id="rsClient" address="${CXFclient}" serviceClass="com.limelight.esb.servicedb.services.ServiceDBExternalRestService"
    loggingFeatureEnabled="true" loggingSizeLimit="500">
    <cxf:providers>
        <bean class="com.fasterxml.jackson.jaxrs.json.JacksonJsonProvider" />
    </cxf:providers>
</cxf:rsClient>

<cxf:rsServer id="rsServer" address="${CXFserver}${serverService}" serviceClass="com.limelight.esb.servicedb.services.ServiceDBRestService"
    loggingFeatureEnabled="true" loggingSizeLimit="500">
    <cxf:providers>
        <bean class="com.fasterxml.jackson.jaxrs.json.JacksonJsonProvider" />
    </cxf:providers>
</cxf:rsServer>

我的路线如下所示：

    <route id="servicedb.rest.company.get.route">
        <from uri="{{servicedb.rest.company.get}}" />
        <log message="Rest getCompany called." />
        <to uri="{{servicedb.company.get}}" />
        <removeHeader headerName="Content-Length" />
    </route>

最后是我的服务类：

@Path("/servicedb/")
public class ServiceDBRestService {
    @GET
    @Path("company/{id}")
    @Produces("application/json")
    public ServiceDBCompanyResponse getCompanyRest(@PathParam("id") String id) {
        //just an interface
        return null;
    }
}

这也可能与预期响应对象的标头有关。为此，您可以使用自定义预处理器对其进行修改，然后在调用之前将其添加到路由中。

public class ServiceDBProcessors implements Processor {

@Override
public void process(Exchange exchng) throws Exception {        
    exchng.getIn().setHeader("CamelCxfRsResponseClass", your.class.here.class);
}

} 现在，我偶然发现这个项目的原因是我在通过肥皂打电话时确实遇到了这个问题，我不得不玩很多，直到我在这里找到你的解决方案。如果上述方法没有帮助，我至少可以提供一个稍微好一点的答案，消除一些转换。我发现您可以在返回后放置另一个处理程序来进行转换，如果您的输入参数是字符串，则应该通过反射来获取它。我不得不写一个像下面这样的类：

    public ServiceDBCompany convertCompany(String companyResponseBody) {
        ServiceDBCompany newCompany = new ServiceDBCompany();
        if(companyResponseBody!=null && companyResponseBody.length()>0){
            ObjectMapper mapper = new ObjectMapper();
            try {
                ServiceDBCompanyResponse company = mapper.readValue(companyResponseBody, ServiceDBCompanyResponse.class);
                newCompany.setCompanyId(company.getCompanyId());
                newCompany.setName(company.getCompanyName());
            } catch(IOException e) {
                System.out.println("Exception reading body, returning null " + ServiceDBCompany.class);
            }
        }
        return newCompany;
}

希望这至少能给你一些可以尝试的解决方案的想法！