Abrufen eines Ereignisses für den Aggregator-Nachrichtengruppenablauf mit Spring Integration DSL?

Question

Ich habe einen Aggregator über die Java-DSL in einem Spring-Integration-Flow konfiguriert, und ich möchte eine Ausnahme auslösen, die bei einem Gruppen-Timeout auf den globalen Fehlerkanal geht.

Der Verwerfungskanal ist nicht gut für mich, da die Verwerfungsnachrichten auf Gruppenebene und nicht für die gesamte Gruppe liegen.

Ich habe versucht, das Anwendungsereignis Publisher wie folgt verwendet wird, aber der Verlag Objekt scheint nicht aufgerufen werden:

  .aggregate(new Consumer<AggregatorSpec>() { 
       @Override 
       public void accept(AggregatorSpec aggregatorSpec) { 
        try { 
         aggregatorSpec 
         .outputProcessor(groupPublishStrategy()) 
         .correlationStrategy(groupPublishStrategy()) 
         .releaseStrategy(groupPublishStrategy()) 
         .groupTimeout(groupTimeout) 
         .get().getT2().setApplicationEventPublisher(myGroupExpirationPublisher()); 
        } 
        catch (Exception e) { 
         e.printStackTrace(); 
        } 
       } 
      }) 

Gibt es eine empfohlene Weg Benachrichtigung für diesen Anwendungsfall zu bekommen? Irgendwelche Ideen, warum das oben genannte nicht zu funktionieren scheint?

Ich nehme an, ich könnte die AggregatorSpec Klasse erweitern, um den Message-Handler so zu konfigurieren, wie ich will, aber ich wollte sehen, ob ich das mit SI-Klassen machen könnte.

Answer 1

Haben Sie gerade getestet und ApplicationEventPublisher ist ordnungsgemäß in der Bean-Initialisierungsphase gefüllt.

Die expireGroup(Object correlationKey, MessageGroup group) ist groß genug, um es hier zu demonstrieren, aber Sie können seinen Code auf GitHub finden. Also, MessageGroupExpiredEvent wird immer veröffentlicht, wenn wir diese Methode erreichen. Natürlich, wenn discardMessage(message); keine Ausnahme auslöst.

OTOH der eprireGroup() erreichbar ist nur in diesem Fall:

if (this.releaseStrategy.canRelease(groupNow)) { 
    completeGroup(correlationKey, groupNow); 
} 
else { 
    expireGroup(correlationKey, groupNow); 
} 

Also, bitte, seien Sie sicher, dass Ihre groupPublishStrategy() richtige Logik hat und nicht true zurückkehrt, wenn Ihre Gruppe noch nicht abgeschlossen ist.

Nun, es wäre wirklich besser, wenn Sie AbstractCorrelatingMessageHandler für Ihren Anwendungsfall debuggen. Wenn Sie sicher sind, dass Ihre Gruppe während einiger groupTimeout nicht abgeschlossen ist, ist die forceComplete(MessageGroup group) ein guter Ort für Sie, um das Debuggen zu starten.

Andernfalls teilen Sie bitte DEBUG Protokolle für die Kategorie org.springframework.integration, wenn Sie denken, dass ein Ereignis ausgegeben werden muss.