SQLiteTransport

SQLite Transport

The SQLite transport uses SQLite databases to store messages. It supports the following features

• Delayed messages
• Message expiration
• Message priority
• Routing
• History tracking
• User-defined metadata columns
• Status-based or FIFO processing
• WAL mode for improved concurrent throughput

The SQLite transport requires that the queue be created before usage. This can be done manually, or via the queue creation classes.

Creating queues

• Queue Creation Options

//Create the queue if it doesn't exist
var queueName = "testing";
var connectionString = @"Data Source=c:\queue.db;Version=3;";
var queueConnection = new QueueConnection(queueName, connectionString);
using (var createQueueContainer = new QueueCreationContainer<SqLiteMessageQueueInit>())
{
	using (var createQueue = createQueueContainer.GetQueueCreation<SqLiteMessageQueueCreation>(queueConnection))
    {
    	if (!createQueue.QueueExists)
        {
        	createQueue.CreateQueue();
        }
    }
 }

Usage

• SQLite Options

[Producer]

using (var queueContainer = new QueueContainer<SqLiteMessageQueueInit>())
{
	using (var queue = queueContainer.CreateProducer<SimpleMessage>(queueConnection))
    {
    	queue.Send(new SimpleMessage { Message = "Hello World" });
    }
}

[Consumer]

using (var queueContainer = new QueueContainer<SqLiteMessageQueueInit>())
{
	using (var queue = queueContainer.CreateConsumer(queueConnection))
    {
    	var notifications = new ConsumerQueueNotifications(
            (n) => Console.WriteLine($"Error: {n.Error}"),
            (n) => Console.WriteLine($"Receive error: {n.Error}"),
            (n) => Console.WriteLine($"Moved to error queue: {n.MessageId}"),
            (n) => Console.WriteLine($"Poison message: {n.MessageId}"),
            (n) => Console.WriteLine($"Rollback: {n.MessageId}"),
            (n) => Console.WriteLine($"Completed: {n.MessageId}"));
        queue.Start<SimpleMessage>(HandleMessages, notifications);
        Console.WriteLine("Processing messages - press any key to stop");
        Console.ReadKey((true));
    }
}

private void HandleMessages(IReceivedMessage<SimpleMessage> message, IWorkerNotification notifications)
{
	notifications.Log.LogDebug($"Processing Message {message.Body.Message}");
}

For more consumer patterns, see ConsumerMethod and ConsumerAsync.

If history tracking is enabled, see MessageHistory for retention and query options.

For in-memory databases, see SQLiteInMemoryDB.

Full samples

See the SQLite samples in the DotNetWorkQueue.Samples repository.