Как обогатить полезную нагрузку с объектом из MongoDB (верблюд-MongoDB)

Question

Я пытаюсь вытащить объект из MongoDb и добавить его к моей текущей полезной нагрузке и сохранить его в другой базе данных:

@Override 
public void configure() throws Exception 
{ 
    from(kafkaEndpoint) 
      .convertBodyTo(DBObject.class) 
      .enrich("mongodb:mongoDb?database=myDbName1&collection=UserColl&operation=findOneByQuery", 
        (original, external) -> { 
         DBObject originalBody = original.getIn().getBody(DBObject.class); 
         DBObject externalBody = external.getIn().getBody(DBObject.class); 

         Map<String, DBObject> map = new HashMap<String, DBObject>(); 
         map.put("original", originalBody); 
         map.put("external", externalBody); 

         original.getIn().setBody(map); 
         return original; 
        }) 
      .to("mongodb:mongoDb?database=myDbName2&collection=UserColl&operation=insert"); 
} 

Проблема что обогащают извлечь запрос из In.body, который держит мой оригинальный объект ...

Так как я могу передать запрос ({ «entity.id»: «»}) к обогащают (mongoldb: ...) и сохранить оригинальный объект, чтобы объединить его с результатами?

Спасибо.

Answer 1

@Override 
    public void configure() throws Exception 
    { 
     from(kafkaEndpoint) 
       .convertBodyTo(DBObject.class) 
       .enrich("direct:findOneByQuery",  // <------- 
         (original, external) -> { 
          DBObject originalBody = original.getIn().getBody(DBObject.class); 
          DBObject externalBody = external.getIn().getBody(DBObject.class); 

          Map<String, DBObject> map = new HashMap<String, DBObject>(); 
          map.put("original", originalBody); 
          map.put("external", externalBody); 

          original.getIn().setBody(map); 
          return original; 
         }) 
       .to("mongodb:mongoDb?database=myDbName2&collection=UserColl&operation=insert"); 

    } 

    from("direct:findOneByQuery") 
      .process(new Processor() 
      { 
       @Override 
       public void process(Exchange exchange) throws Exception 
       { 
        DBObject body = exchange.getIn().getBody(DBObject.class); 
        DBObject query = BasicDBObjectBuilder.start() 
          .append("entity._id", body.get("_id")) 
          .get(); 

        exchange.getIn().setBody(query); 
       } 
      }) 
      .to("mongodb:mongoDb?database=myDbName1&collection=UserColl&operation=findOneByQuery"); 

//