Two similar methods to process a queue of messages

Question

I have two functions with loops that behave nearly identically, but with some key differences. As a consequence, most (but not all!) of the code in the functions is repeated. I can accept this repetition, but I would prefer to have the repeated code in a single place, so that I do not need to remember to change both if a modification of the logic is required.

These methods are contained in an abstract base class, and they are intended to provide functionality to clients that consume messages from a queue. Clients provide their custom functions (as Action or Func) and the base class handles interacting with the message queues and executing the custom functions. Some clients want to parse data contained in a queue and do nothing else, but other clients want to parse data contained in a queue and push resulting data along a workflow into another queue.

A concrete example for the first case (where nothing is passed along a workflow) is a logging queue. Applications send messages containing logs to the logging queue. The client reads those messages and inserts the logs to a DB (or wherever). The workflow terminates at this point.

A concrete example for the second case is an ETL process where the workflow has been separated into multiple queues. There would be an extraction queue, a transformation queue, and a load queue. The clients of the extraction queue need to read messages, perform some data processing, and then write messages containing the processed data into the transformation queue -- where the clients of the transformation queue will perform a similar process.

Here is the class. The two functions in question are the overloads of WorkQueueUntilEmpty

/// <summary>
/// Basic functionality around a consumer that acts on message data
/// </summary>
public abstract class BaseWorker : BaseActor
{
    /// <summary>
    /// If set, the identifier of the queue to which data is written after this worker has finished working on it
    /// </summary>
    protected string NextQueue;

    /// <summary>
    /// Builds the base worker with the required dependencies
    /// </summary>
    /// <param name="factory"></param>
    protected BaseWorker(IConnectionProvider factory) : base(factory)
    {
    }

    /// <summary>
    /// Configures this worker
    /// </summary>
    /// <param name="configuration"></param>
    public virtual void ConfigureWorker(IWorkerConfiguration configuration)
    {
        NextQueue = configuration.NextQueue;
    }

    /// <summary>
    /// Sends a report that this worker has nothing to do
    /// </summary>
    protected void SendNoWorkReport()
    {
        try
        {
            var message = Message<WorkerReport>.WithData(new WorkerReport
            {
                SourceQueue = SourceQueue,
                WorkerId = Id,
                Status = WorkerStatus.NoWork
            });
            ExecuteOnQueue<WorkerReport>(ReportQueue, q => q.Write(message));
        }
        catch (Exception ex)
        {
            SendErrorReport(ex);
            ErrorAction(ex);
        }
    }

    /// <summary>
    /// Reads a message from the SourceQueue, executes the provided workMethod on each datum in the message, confirms the receipt, then reads the next message from the source queue
    /// until the queue has been emptied
    /// </summary>
    /// <typeparam name="TMessageDataType"></typeparam>
    /// <param name="workMethod"></param>
    protected void WorkQueueUntilEmpty<TMessageDataType>(Action<TMessageDataType> workMethod)
    {
        try
        {
            Running = true;
            var message = GetNextMessage<TMessageDataType>(SourceQueue);
            while (message != null)
            {
                // This overload of WorkBatch returns void
                WorkBatch(message.Data, workMethod);
                ExecuteOnQueue<TMessageDataType>(SourceQueue, q => q.ConfirmMessageReceipt(message));
                message = GetNextMessage<TMessageDataType>(SourceQueue);
            }
            SendNoWorkReport();
        }
        catch (Exception ex)
        {
            SendErrorReport(ex);
            throw;
        }
    }

    /// <summary>
    /// Reads a message from the SourceQueue, executes the provided function on the message and writes the resulting data to the NextQueue
    /// Repeats until the SourceQueue is empty
    /// </summary>
    /// <typeparam name="TMessageDataType"></typeparam>
    /// <typeparam name="TNextQueueMessageDataType"></typeparam>
    /// <param name="compositionFunction"></param>
    protected void WorkQueueUntilEmpty<TMessageDataType, TNextQueueMessageDataType>(Func<TMessageDataType, TNextQueueMessageDataType> compositionFunction)
    {
        try
        {
            Running = true;
            var message = GetNextMessage<TMessageDataType>(SourceQueue);
            while (message != null)
            {
                // This overload of WorkBatch returns an IEnumerable<TNextQueueMessageDataType>
                var nextQueueMessageData = WorkBatch(message.Data, compositionFunction);
                // This writes the results of the composition function to the next queue
                ExecuteOnQueue<TNextQueueMessageDataType>(NextQueue, q => q.Write(Message<TNextQueueMessageDataType>.WithData(nextQueueMessageData)));
                ExecuteOnQueue<TMessageDataType>(SourceQueue, q => q.ConfirmMessageReceipt(message));
                message = GetNextMessage<TMessageDataType>(SourceQueue);
            }
            SendNoWorkReport();
        }
        catch (Exception ex)
        {
            SendErrorReport(ex);
            throw;
        }
    }

    /// <summary>
    /// Sends a report that this worker has finished all of its available work, and has stopped.
    /// </summary>
    /// <param name="timeElapsed"></param>
    protected void SendWorkCompleteReport(TimeSpan timeElapsed)
    {
        try
        {
            var report = Message<WorkerReport>.WithData(new WorkerReport
            {
                SourceQueue = SourceQueue,
                WorkerId = Id,
                ElapsedTime = timeElapsed,
                Status = WorkerStatus.TaskComplete
            });
            ExecuteOnQueue<WorkerReport>(ReportQueue, q => q.Write(report));
        }
        catch (Exception ex)
        {
            SendErrorReport(ex);
            ErrorAction(ex);
        }
    }

    /// <summary>
    /// Executes the provided function on each item in the batch, yielding the result of the invocation. Sends a Work Complete message to the report queue when finished
    /// </summary>
    /// <typeparam name="TMessageDataType"></typeparam>
    /// <typeparam name="TNextQueueMessageDataType"></typeparam>
    /// <param name="data"></param>
    /// <param name="compositionFunction"></param>
    /// <returns></returns>
    protected IEnumerable<TNextQueueMessageDataType> WorkBatch<TMessageDataType, TNextQueueMessageDataType>(List<TMessageDataType> data, Func<TMessageDataType, TNextQueueMessageDataType> compositionFunction)
    {
        try
        {
            var start = DateTime.Now;
            var results = data.Select(compositionFunction).ToList();
            var elapsed = DateTime.Now - start;
            SendWorkCompleteReport(elapsed);
            return results;
        }
        catch (Exception ex)
        {
            SendErrorReport(ex);
            ErrorAction(ex);
            throw;
        }
    }

    /// <summary>
    /// Executes the provided action on each item in the batch, and sends a Work Complete message to the report queue when finished
    /// </summary>
    /// <typeparam name="T"></typeparam>
    /// <param name="batch"></param>
    /// <param name="workFunction"></param>
    protected void WorkBatch<T>(IEnumerable<T> batch, Action<T> workFunction)
    {
        try
        {
            var start = DateTime.Now;
            foreach (var datum in batch)
            {
                workFunction(datum);
            }
            var elapsed = DateTime.Now - start;
            SendWorkCompleteReport(elapsed);
        }
        catch (Exception ex)
        {
            SendErrorReport(ex);
            ErrorAction(ex);
            throw;
        }
    }
}

I am struggling with identifying a clean way to make these functions DRY, even though there are only three lines of distinct code between the two of them.

Answer 1

I'm not sure if this would work as intended, but what if instead of not returning anything, you'd return nothing?

Think null object pattern: return an object that behaves "neutral" and doesn't do anything except moving along with the flock like any other regular object would do.

// This overload of WorkBatch returns void
WorkBatch(message.Data, workMethod);

What if you make this WorkBatch return an empty IEnumerable<TNextQueueMessageDataType> instead?

You can then feed it to Message<TNextQueueMessageDataType>.WithData() and it wouldn't do anything, because it operates on an empty list.

Of course this assumes that an empty IEnumerable<TNextQueueMessageDataType> is nothing that WorkBatch(message.Data, compositionFunction) would regularly produce or that would cause Message<TNextQueueMessageDataType>.WithData() to have some functionality that's different from it not being executed.

I guessEnumerable.Empty<TNextQueueMessageDataType>(); would do the trick of returning the empty IEnumerable<TNextQueueMessageDataType>.