4
\$\begingroup\$

Currently I'm writing something like a messaging or logging framework (I'm not really sure how you name something like that). There are different servers which have to be notified if something happens.

The servers mainly store the data e. g. for statistics. One server is an ELK stack which receives the data via a REST api.
But every server expects a slightly different model, so the events have to be transformed in different ways, e. g. by adding an application key or formatting the timestamp.

A windows desktop application, which runs on multiple devies, creates the events, which have to be transmitted to the servers.

The following code is part of the client (windows desktop application).

I have a facade which accepts a event and passes it to all targets.

public class EventController : IEventSender
    {
        public EventController(params IEventTarget[] targets)
        {
            Targets = targets;
        }

        public IEventTarget[] Targets { get; }

        public void SendEvent(IEvent e)
        {
            foreach (var target in Targets)
            {
                target.SendEvent(e);
            }
        }
    }

Each target has to handle the event on its own. But the events have to be send to the servers in the right order (for that I created a queue with System.Collections.Concurrent.BlockingCollection and System.Collections.Concurrent.ConcurrentQueue). For the beginning (and later for debugging and testing purposes) I created a target which writes the events to a file. Currently 3 different targets (+ FileTarget) are intended. Each targe should implement its own queue and background task (I want to prevent duplicated code by inheritance).

public class FileTarget : IEventTarget
    {
        private static readonly ILogger Log = LoggerFactory.GetLogger(typeof(FileTarget));  // logger interface, not of importance 
        private static readonly object FileLock = new object();  // lock to protect multiple file access

        private readonly BlockingCollection<IEvent> _queue =
            new BlockingCollection<IEvent>(new ConcurrentQueue<IEvent>());  // queue of elements to proceed

        private bool _backgroundActionShouldRun = true;  // token to exit background action

        public FileTarget(string fileName)
        {
            FileName = fileName;

            StartBackgroundAction();
        }

        public string FileName { get; }

        public void SendEvent(IEvent e)
        {
            _queue.Add(e);
        }

        ~FileTarget()
        {
            _backgroundActionShouldRun = false;
        }

        private void StartBackgroundAction()
        {
            Task.Run((Action) BackgroundAction);
        }

        private void BackgroundAction()
        {
            // im not sure if this task will be completed if the instance get destroyed
            while (_backgroundActionShouldRun) { 
                try
                {
                    var e = _queue.Take();
                    var transformedEvent = TransformEvent(e);
                    WriteEventToFile(transformedEvent);
                }
                catch (Exception ex)
                {
                    Log.Error("Unable to take an object from queue.", ex);
                }
            }
        }

        private TransformedEvent TransformEvent(IEvent e)
        {
            // transform the event and return it
        }

        private void WriteEventToFile(TransformedEvent e)
        {
            try
            {
                // im not sure if this is thread save, if it is i can remove the lock
                lock (FileLock)
                {
                    // System.IO.File.AppendAllLines requires a collection, I'm not happy with this
                    string[] contents = {e.ToString()};
                    System.IO.File.AppendAllLines(FileName, contents );
                }
            }
            catch (Exception ex)
            {
                Log.Error($"Failed to write message to file {FileName}.", ex);
            }
        }
    }

I have multiple questions on how to improve that code

  • Is there a better way to queue the events?
  • Is this thread safe (as long as no other process/thread tries to access the file)?
  • I'm afraid that in some cases (or maybe in most?) the background task will never be completed.
  • Should I move the queue and the background process to the facade?
  • I have oriented myself by some logging frameworks. Should I take advantage of one like NLog with a custom target?
  • Do you have some other suggestions to improve the code?
\$\endgroup\$
  • \$\begingroup\$ Each target has to handle the event on its own. But the events have to be send to the servers in the right order does that mean that the queue should be in the controller or only this one event handler, or should all event handlers implement a queue? \$\endgroup\$ – dfhwze Sep 12 at 6:05
  • \$\begingroup\$ I wanted to say that the events have to be augmented/transformed in different ways for each server, e. g. adding a application-key or formatting the timestamp. Because some interactions trigger multiple events the order should be preserved.Currently every handler should implement its own queue (thats why I saked if I should move the queue, so the handler has only to handle the event) \$\endgroup\$ – D. Weber Sep 12 at 6:16
  • \$\begingroup\$ I would add this information to the question (events have to be transformed for each server). \$\endgroup\$ – dfhwze Sep 12 at 6:21
  • 2
    \$\begingroup\$ @dfhwze done, thx for your help :) \$\endgroup\$ – D. Weber Sep 12 at 6:26
6
\$\begingroup\$

Alternatives

There are plenty of robust options for message queueing in C# that handle multithreading gracefully as well as scaling to truly massive sizes.

The microsoft solution is MSMQ.

There are plenty of free or commercial alternatives. ActiveMQ RabbitMQ are a couple I have experience with. Both are useful with solid .NET libraries supporting them. Generally, I would use one of the existing MQ options. They are backed by persistence engines which store the queue data in databases so you can do useful things like replay blocks of messages to debug issues, and bulk process messages to test function under heavy load.

Review

The file writing sections are locked for exclusivity, which looks like the only place you are in resource contention so it seems threadsafe (you should always test it thoroughly of course :-) ).

If you are worried that some tasks will not complete, you can fire an async task with a cancellation token set to expire after some timeout (don't forget to set the task to throw on cancellation if you need notification).

Where you put the code for the facade and the queue is up to you, I'd leave them separate as much as possible, but that decision may not work in your case. Code it however you are comfortable.

I love NLog. It is incredibly powerful and flexible. I would use it for logging but again, it is up to you. If it seems to fit your needs, use it.

I hope some of this is useful.

\$\endgroup\$
  • \$\begingroup\$ There are plenty of robust options for message queueing in C# - there are, indeed but unfortunatelly they are not always easy to setup or even supported by a company so more often than not, you need your own workaround. \$\endgroup\$ – t3chb0t Sep 12 at 6:52
  • \$\begingroup\$ When requiring cross-tier reliability and distributed transactions, it's always going to be tough, third party library or rolling out your own API. \$\endgroup\$ – dfhwze Sep 12 at 7:11
  • \$\begingroup\$ Thanks for your answer. In my opinion a proper MQ like RabbitMQ is too large for our scenario. But your review is helpful, I will take a look at cancelation tokens to handle the hbackground task. (I would upvote your answer, but unfortunately i have not enough rep) \$\endgroup\$ – D. Weber Sep 12 at 7:35
2
\$\begingroup\$

You started correct, however perhaps you are over complicating things. All you need to do is create an entry where you can log your structures and a method that fills a ConcurrentQueue with the items you'd like to log, this will do just fine.

Then in your class you add a timer that dequeues items from the buffer and store them as and how you like. that's all you need to do. let me show you a simple example:

namespace MyApplication.Logging
{
    using System;
    using System.IO;
    using System.Text;

    class Logger<T>:IDisposable
    {
        private readonly System.Collections.Concurrent.ConcurrentQueue<T> queue;
        private readonly System.Threading.Timer timer;
        private bool isLogging = false;//default is false so you can remove assigning it if you like
        private readonly FileInfo file;

        /// <summary>
        /// Initializes a new instance of the <see cref="Logger{T}"/> class.
        /// </summary>
        /// <param name="logFile">The log file.</param>
        /// <exception cref="System.ArgumentNullException">logFile must be provided</exception>
        /// <exception cref="System.ApplicationException">Will throw error if logging directory could not be created</exception>
        public Logger(FileInfo logFile)
        {
            if (logFile is null)
                throw new ArgumentNullException(nameof(logFile));

            this.file = logFile;

            if (!file.Directory.Exists)
            {
                try
                {
                    file.Directory.Create();
                }
                catch (Exception e)
                {
                    throw new ApplicationException("Could not create log directory, look at inner exception for details", e);
                }
            }

            //create buffer
            queue = new System.Collections.Concurrent.ConcurrentQueue<T>();
            //create timer to save items in buffer, start in 60 seconds, then process every 10 seconds
            timer = new System.Threading.Timer(DequeueLog, null, TimeSpan.FromSeconds(60), TimeSpan.FromSeconds(10));

        }

        /// <summary>
        /// Dequeues the log and saves it to disk.
        /// </summary>
        /// <param name="_">The state object is ignored.</param>
        private void DequeueLog(object _)
        {
            //avoid duplicate processing
            if (isLogging || queue.Count == 0)
                return;

            isLogging = true;
            try
            {
                var sb = new StringBuilder();
                while (queue.TryDequeue(out T item))
                {
                    sb.AppendLine(System.Text.Json.JsonSerializer.Serialize(item));
                }
                File.WriteAllText(file.FullName, sb.ToString());
                file.Refresh();
            }
            finally {
                isLogging = false;
            }
        }

        /// <summary>
        /// Logs the specified item.
        /// </summary>
        /// <param name="item">The item.</param>
        public void Log(T item)=> queue.Enqueue(item);

        /// <summary>
        /// Gets the log size in MB.
        /// </summary>
        public double LogSizeInMB => file.Exists ? file.Length / 1_048_576:0;

        #region IDisposable Support
        private bool disposedValue = false; // To detect redundant calls

        protected virtual void Dispose(bool disposing)
        {
            if (!disposedValue)
            {
                if (disposing)
                {
                    timer.Dispose();
                    if (queue.Count > 0)
                        DequeueLog(null);
                }


                disposedValue = true;
            }
        }

        // TODO: override a finalizer only if Dispose(bool disposing) above has code to free unmanaged resources.
        // ~Logger()
        // {
        //   // Do not change this code. Put cleanup code in Dispose(bool disposing) above.
        //   Dispose(false);
        // }

        // This code added to correctly implement the disposable pattern.
        public void Dispose()
        {
            // Do not change this code. Put cleanup code in Dispose(bool disposing) above.
            Dispose(true);
            // TODO: uncomment the following line if the finalizer is overridden above.
            // GC.SuppressFinalize(this);
        }
        #endregion


    }
}

I have used no 3rd party lib's only the latest .net as of this writing.

Important is to not have double processing of the queue if you are writing to disk as a disk file will not like this, in a database this would work but having a single "pump" usually is best to avoid locking & blocking. The variable isLogging takes care of that.

\$\endgroup\$
2
\$\begingroup\$

EventController

Transformation

the events have to be augmented/transformed in different ways

The code provided in the question does not show event transformation. However, if targets are allowed to change the event data to fit the needs of the that specific target, you should provide a copy of the event to each target.


Dispatching

There are different servers which have to be notified if something happens.

Given the code provided in the question it doesn't look like multiple servers get notified. All we see is a controller dispatching an event received from any caller to the registered targets (in-process or external .. not clear). I can't tell whether each server has such controller running, or whether a master server receives the event from a UI and dispatches the events further to other servers.

EventController.SendEvent(IEvent e) // Does this code run on a master server that dispatches 
                                    // events to other servers, or does each server have its
                                    // own controller?

Regardless where the code runs, if the requirement is that each target should get notified independently, I would make event publication more robust.

public void SendEvent(IEvent e)
{
    foreach (var target in Targets)
    {
        // - what if a target gets registered during publication of an event?
        // - what if a target throws an exception?
        // - what if a target takes ages to process the event?
        target.SendEvent(e);
    }
}

Consider changing the following aspects:

  • use a mutex to make event publication and target registration mutually exclusive operations (unless all targets are only set once at constructor, but that doesn't provide much configuration options at runtime)
  • catch and handle errors by individual targets without exiting the loop early
  • dispatch the event asynchronously to each target

FileTarget

You've provided a static file lock, meaning you allow multiple instances to write to the same file.

private static readonly object FileLock = new object();

But what's the benefit of having an instance queue if several concurrent queues (each instance its own queue) are used to write to the same file?

private readonly BlockingCollection<IEvent> _queue = ..

And how sure can you be that no other process is accessing the file simultaniously?

System.IO.File.AppendAllLines..

This is already a more robust approach that allows shared access to the file:

new FileStream(FileName, FileMode.Append, FileAccess.Write, FileShare.Read);

As suggested in the other answers, use a CancellationToken rather than a bool _backgroundActionShouldRun and implement the Dispose Pattern.

I don't know how many targets will write to a file. But think about how many shared threadpool threads will not be able to participate in the server because they are exclusively working for a target instance.

private void StartBackgroundAction()
{
    Task.Run((Action) BackgroundAction); // This task's worker thread is dedicated to you :s
}
\$\endgroup\$
  • 1
    \$\begingroup\$ Thx for your answer. I'll try to response to all your points. First Transformation: you're right, in the code is no example of transformation. I tried to keep the code as clean as possible, but I'll add it to the example. The event itself don't get transformed, it is used to create a new object from another type, suited for the needs of the server. \$\endgroup\$ – D. Weber Sep 12 at 18:48
  • 1
    \$\begingroup\$ Dispatching: I though it it will be enough if I only post one possible target. Another target should send the data e. g. to an ELK stack which only stores the data. So every server handels the transmitted values in a different way. The whole code I have posted is part of a client application which runs on multiple devies. The server only accumulate the data. (Another point I will add to my question). I'll apply your sugestions to make the event publication more robust, thx. \$\endgroup\$ – D. Weber Sep 12 at 18:48
  • 1
    \$\begingroup\$ FileTarget: Currently only 3 targets are intended which only one instance but all your suggestions sound great. I'll try to apply them. \$\endgroup\$ – D. Weber Sep 12 at 18:49
  • 1
    \$\begingroup\$ Glad to be of assistance. If you apply these changes, make sure not to edit the current question, but perhaps write a self-answer or follow-up question after you have sufficient feedback on this question. \$\endgroup\$ – dfhwze Sep 12 at 18:50

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.