Come registrare il messaggio non riuscito in masstransit?

Question

Sto cercando una buona soluzione per registrare il messaggio non riuscito, subito dopo il superamento del limite di tentativi, senza avere un accordo con la coda degli errori. Quello che ho trovato finora:

• posso ereditare da InMemoryInboundMessageTracker e sovrascrivere IsRetryLimitExceeded, ma a questo punto non ci sono informazioni su messaggio stesso, tranne id.
• posso realizzare IInboundMessageInterceptor e ottenere IConsumeContext in Pre/PostDispatch, ma a questo punto non ci sono informazioni sul successo/fallimento.

Così come una soluzione, posso ottenere IConsumeContext in preDispatch metterlo in una sorta di una cache quindi ottenere fuori di una cache in IsRetryLimitExceeded quando limite di tentativi di superamento.

metodi sono chiamati a tale ordine: IsRetryLimitExceeded -> preDispatch -> PostDispatch

Quindi non riesco a trovare un buon posto per rimuovere messaggio elaborato con successo da una cache.

Ovviamente posso usare una cache con dimensioni ridotte ma questa intera soluzione sembra essere strana.

Sarebbe gradita qualsiasi riflessione in merito.

Answer 1

È possibile implementare e configurare il proprio Message Carry Tracking sul bus, in modo che i messaggi non riusciti siano passati attraverso l'implementazione. È possibile delegare al tracker di tentativi predefinito e intercettare solo gli eventi in modo da poter agire su di essi, oppure è possibile implementare il proprio tracciamento dei tentativi se necessario.

MessageTrackerFactory è il delegato per la configurazione, penso che l'interfaccia sia nelle vicinanze.

Answer 2

ho ednded con questa soluzione:

class MessageInterceptor: IInboundMessageInterceptor 
{ 
    public void PreDispatch(IConsumeContext context) 
    { 
     MessageTracker.Register(context); 
    } 

    public void PostDispatch(IConsumeContext context) 
    {} 
} 

class MessageTracker: InMemoryInboundMessageTracker 
{ 
    readonly Logger logger; 

    static readonly ConcurrentDictionary<string, IConsumeContext> DispatchingCache = new ConcurrentDictionary<string, IConsumeContext>(); 

    public MessageTracker(int retryLimit, Logger logger) 
     : base(retryLimit) 
    { 
     this.logger = logger; 
    } 

    public static void Register(IConsumeContext context) 
    { 
     DispatchingCache.GetOrAdd(context.MessageId, context); 
    } 

    public override void MessageWasReceivedSuccessfully(string id) 
    { 
     base.MessageWasReceivedSuccessfully(id); 

     IConsumeContext value; 
     DispatchingCache.TryRemove(id, out value); 
    } 

    public override bool IsRetryLimitExceeded(string id, out Exception retryException, out IEnumerable<Action> faultActions) 
    { 
     var result = base.IsRetryLimitExceeded(id, out retryException, out faultActions); 

     IConsumeContext failed; 
     if (!result || !DispatchingCache.TryRemove(id, out failed)) 
      return result; 

     // --> log failed IConsumeContext with exception 

     return true; 
    } 
} 

E per collegare tali classi nelle

 serviceBus = ServiceBusFactory.New(config => 
     { 
      ... 
      config.AddBusConfigurator(new PostCreateBusBuilderConfigurator(sb => 
      { 
       var interceptorConfig = new InboundMessageInterceptorConfigurator(sb.InboundPipeline); 
       interceptorConfig.Create(new MessageInterceptor()); 
      })); 

      config.SetDefaultInboundMessageTrackerFactory(retryLimit => new MessageTracker(retryLimit, LogManager.GetCurrentClassLogger())); 
     });