Wie die Nutzlast mit einem Objekt von MongoDB (Kamel mongodb) bereichern

Question

Ich versuche Objekt von MongoDB zu ziehen und es zu meiner aktuellen Nutzlast ADD und es in einer anderen Datenbank speichern:

@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"); 
} 

Das Problem dass enrich die Abfrage aus dem In.body holen, die mein ursprüngliches Objekt hält ...

So wie kann ich Abfrage übergeben ({ "entity.id": ""}) zu Enrich (mongoldb: ...) und Originalobjekt erhalten, um es mit Ergebnissen zu verbinden?

Danke.

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"); 

//