Invoke Action when message is accepted by LinkTo

Question

Is there a way to invoke an action when a message is accepted by a ITargetBlock via the LinkTo method?

I'm essentially looking for a hook for the moment when the message responsibility is transferred between one block and another.

The action would not be performed on the target but on the source and only after the message has been accepted.

Will I need to write a custom block and override the LinkTo method?

Answer 1

There is no such hook in the built-in blocks. Could you explain why exactly do you want to do this? I don't see why anything like this would be necessary.

And you could do this by writing a custom block, but I don't think the LinkTo() method is relevant to this, this functionality would have to be in whatever method calls the target's OfferMessage() and also in ConsumeMessage().

Answer 2

Because I want to signal to a non-block resource that the message has been consumed from by custom block. An example might be some kind of audit functionality. The ability to say "this message was accepted by this block at this time" without relying on the target to provide that functionality.

Answer 3

In that case, I think the best option is to write your own version of LinkTo() that inserts another block between the two blocks. The new block would rely any method call between the two blocks and call a delegate when a message is accepted. Something like:

public static class DataflowExtensions
{
	public static IDisposable LinkTo<T>(this ISourceBlock<T> source, ITargetBlock<T> target, Action<T> onAccepted)
	{
		var onAcceptedBlock = new OnAcceptedBlock<T>(source, target, onAccepted);
		return source.LinkTo(onAcceptedBlock);
	}
}

internal class OnAcceptedBlock<T> : IPropagatorBlock<T, T>
{
	private readonly ISourceBlock<T> m_sourceBlock;
	private readonly ITargetBlock<T> m_targetBlock;
	private readonly Action<T> m_onAccepted;

	public OnAcceptedBlock(ISourceBlock<T> sourceBlock, ITargetBlock<T> targetBlock, Action<T> onAccepted)
	{
		m_sourceBlock = sourceBlock;
		m_targetBlock = targetBlock;
		m_onAccepted = onAccepted;
	}

	public DataflowMessageStatus OfferMessage(DataflowMessageHeader messageHeader, T messageValue, ISourceBlock<T> source, bool consumeToAccept)
	{
		var result = m_targetBlock.OfferMessage(messageHeader, messageValue, this, consumeToAccept);
		if (!consumeToAccept && result == DataflowMessageStatus.Accepted)
			m_onAccepted(messageValue);
		return result;
	}

	public void Complete()
	{
		throw new NotImplementedException();
	}

	public void Fault(Exception exception)
	{
		throw new NotImplementedException();
	}

	public Task Completion { get { throw new NotImplementedException(); } }

	public IDisposable LinkTo(ITargetBlock<T> target, DataflowLinkOptions linkOptions)
	{
		throw new NotImplementedException();
	}

	public T ConsumeMessage(DataflowMessageHeader messageHeader, ITargetBlock<T> target, out bool messageConsumed)
	{
		var result = m_sourceBlock.ConsumeMessage(messageHeader, this, out messageConsumed);
		if (messageConsumed)
			m_onAccepted(result);
		return result;
	}

	public bool ReserveMessage(DataflowMessageHeader messageHeader, ITargetBlock<T> target)
	{
		return m_sourceBlock.ReserveMessage(messageHeader, this);
	}

	public void ReleaseReservation(DataflowMessageHeader messageHeader, ITargetBlock<T> target)
	{
		m_sourceBlock.ReleaseReservation(messageHeader, this);
	}
}

Answer 4

Woah, thanks for the quick response and the code snippet. I'll have a look at this later.