When an object is inserted into the queue with a delay, the operation executes immediately about dotnetworkqueue HOT 16 CLOSED

if (!(queueCreator.QueueExists))
{
	Logger.Debug($"Queue {queueName} doesn't exist, creating it");
	queueCreator.Options.EnableDelayedProcessing = true;
	queueCreator.Options.EnableHeartBeat = true;
	queueCreator.Options.EnableMessageExpiration = true;
	queueCreator.Options.EnableStatus = true;
	queueCreator.Options.EnableStatusTable = true;
	var result = queueCreator.CreateQueue();
	Logger.Debug(result.Status.ToString());
}

var container = new QueueContainer<SqLiteMessageQueueInit>();
var schedulerContainer = new SchedulerContainer();
var scheduler = schedulerContainer.CreateTaskScheduler();
taskFactory = schedulerContainer.CreateTaskFactory(scheduler);
taskFactory.Scheduler.Configuration.MaximumThreads = Config.MessageQueue.SchedulerMaxThreads;
taskFactory.Scheduler.Start();

AsyncMessageQueue<T> queue;

using (var queueCreator = queueCreationcontainer.GetQueueCreation<SqLiteMessageQueueCreation>(queueName, connectionString))
{
	Logger.Debug($"Checking to see if queue {queueName} exists");
	if (!(queueCreator.QueueExists))
	{
		Logger.Debug($"Queue {queueName} doesn't exist, creating it");
		queueCreator.Options.EnableDelayedProcessing = true;
		queueCreator.Options.EnableHeartBeat = true;
		queueCreator.Options.EnableMessageExpiration = true;
		queueCreator.Options.EnableStatus = true;
		queueCreator.Options.EnableStatusTable = true;
		var result = queueCreator.CreateQueue();
		Logger.Debug(result.Status.ToString());
	}

	queue = new AsyncMessageQueue<T>
	{
		producer = container.CreateProducer<T>(queueName, connectionString),
		consumer = container.CreateConsumerQueueScheduler(queueName, connectionString, taskFactory)
	};

	queue.producer.Configuration.TimeConfiguration.RefreshTime = TimeSpan.FromMinutes(Config.MessageQueue.RefreshTime);
	queue.consumer.Configuration.TimeConfiguration.RefreshTime = TimeSpan.FromMinutes(Config.MessageQueue.RefreshTime);
	queue.consumer.Configuration.Worker.TimeToWaitForWorkersToCancel = TimeSpan.FromSeconds(10);
	queue.consumer.Configuration.Worker.WorkerCount = Config.MessageQueue.WorkerMaxThreads;
}

public class AsyncMessageQueue<T> where T : class
{
	public IProducerQueue<T> producer { get; set; }
	public IConsumerQueueScheduler consumer { get; set; }

	public void Start(Action<T> messageHandler)
	{
		this.consumer.Start<T>((m, n) => messageHandler(m.Body));
	}

	public void Stop()
	{
		producer.Dispose();
		consumer.Dispose();
	}

	public void Send(T message, AdditionalMessageData data = null)
	{
		this.producer.Send(message, data);
	}

	public async Task SendAsync(T message, AdditionalMessageData data = null)
	{
		await this.producer.SendAsync(message, data);
	}
}

queue.Start(async (someObject) => someObjectHandler(someObject));

var data = new AdditionalMessageData();
data.SetDelay(TimeSpan.FromSeconds(120));
queue.SendAsync(someObject, data);

async Task someObjectHandler(SomeObject someObject)
{
    doStuff(someObject);
}