SubscriptionTask.cs

using System;
using System.Threading;
using System.Threading.Tasks;
using Waher.Events;
 
namespace TAG.Simulator.MQ.Tasks
{
    internal class SubscriptionTask : MqTask
    {
        private readonly ManualResetEvent cancel;
        private readonly TaskCompletionSource<bool> stopped;
        private readonly MqMessageEventHandler callback;
        private readonly string queue;
        private readonly object state;
 
        public SubscriptionTask(MqClient Client, string Queue, ManualResetEvent Cancel, TaskCompletionSource<bool> Stopped,
            MqMessageEventHandler Callback, object State)
            : base(Client)
        {
            this.queue = Queue;
            this.cancel = Cancel;
            this.stopped = Stopped;
            this.callback = Callback;
            this.state = State;
        }
 
        public override void Dispose()
        {
            this.stopped?.TrySetResult(true);
            base.Dispose();
        }
 
        public override bool DoWork()
        {
            if (this.cancel?.WaitOne(0) ?? false)
            {
                this.Client.Information("Cancelling subscription to messages from " + this.queue);
                this.stopped?.TrySetResult(true);
                return false;
            }
            else
            {
                string Message = this.Client.GetOne(this.queue, 1000);
 
                if (!(Message is null))
                {
                    try
                    {
                        this.callback(this.Client, new MqMessageEventArgs(this.Client, Message, this.state));
                    }
                    catch (Exception ex)
                    {
                        Log.Exception(ex);
                    }
                }
 
                return true;
            }
        }
    }
}