SQLiteInMemoryDB

SQLite In Memory DB

You may use in-memory SQLite databases inside of the same process. Here is an example.

Connection String

To specify an in memory database, use a Uri connection string and set the mode to memory, and the cache to shared.

ConnectionString = "FullUri=file:dbName.db3?mode=memory&cache=shared;Version=3;";

The Version=3 in the connection string is the System.Data.SQLite protocol version indicator, not a SQLite runtime version selector. It is required by the System.Data.SQLite.Core driver.

Keeping Connection Alive

SQLite will release the database if no open connections to the database are present. Because of this, you need to either:

1. Keep the creation context alive for as long as the producers and consumers are running.
2. Obtain the scope from the creation context and hold onto it while the producers and consumers are running. When processing is complete, dispose of the scope to close the last connection.

The below example uses option #2.

Sample Code

Note: This sample uses Serilog for logging. Any ILoggerFactory implementation works -- Serilog, NLog, the built-in Microsoft.Extensions.Logging console provider, etc.

public class SimpleMessage
{
	public string Message { get; set; }
}

using System;
using System.Threading.Tasks;
using DotNetWorkQueue;
using DotNetWorkQueue.Transport.SQLite.Basic;
using Serilog;
namespace SQLiteMemoryQueue
{
    class Program
    {
        private static readonly string ConnectionString = "FullUri=file:dbName.db3?mode=memory&cache=shared;Version=3;";
        private static readonly string QueueName = "testing";
        static void Main(string[] args)
        {
            RunQueue();
        }

        private static async Task RunQueue()
        {
            var log = new LoggerConfiguration()
            .WriteTo.ColoredConsole(outputTemplate: "{Timestamp:HH:mm} [{Level}] ({Name:l}) {Message}{NewLine}{Exception}")
            .CreateLogger();

            Log.Logger = log;

            using (var scope = CreateQueue())
            {
                using (var queueContainerSend = new QueueContainer<SqLiteMessageQueueInit>())
                {
                    var queueConnection = new QueueConnection(QueueName, ConnectionString);
                    using (var sendQueue = queueContainerSend.CreateProducer<SimpleMessage>(queueConnection))
                    {
                        using (var queueContainerConsumer = new QueueContainer<SqLiteMessageQueueInit>())
                        {
                            using (var queue = queueContainerConsumer.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);

                                //process user input
                                var keepRunning = true;

                                while (keepRunning)
                                {
                                    Console.WriteLine(@"a) Send 1 jobs
	b) Send 10 jobs
	c) Send 100 jobs
	q) Quit");
                                    var key = char.ToLower(Console.ReadKey(true).KeyChar);

                                    switch (key)
                                    {
                                        case 'a':
                                            await Send(1, sendQueue);
                                            break;
                                        case 'b':
                                            await Send(10, sendQueue);
                                            break;
                                        case 'c':
                                            await Send(100, sendQueue);
                                            break;
                                        case 'q':
                                            Console.WriteLine("Quitting");
                                            keepRunning = false;
                                            break;
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }

        private static async Task Send(int messageCount, IProducerQueue<SimpleMessage> queue)
        {
            for (var i = 0; i < messageCount; i++)
            {
                var result = await queue.SendAsync(new SimpleMessage {Message = "Hello World"});
                if (result.SendingException != null)
                {
                    Console.WriteLine(result.SendingException);
                }
            }
        }
        private static void HandleMessages(IReceivedMessage<SimpleMessage> message, IWorkerNotification notifications)
        {
            notifications.Log.LogInformation($"Processing message {message.MessageId.Id.Value}");
        }

        private static ICreationScope CreateQueue()
        {
            using (var createQueueContainer = new QueueCreationContainer<SqLiteMessageQueueInit>())
            {
                var queueConnection = new QueueConnection(QueueName, ConnectionString);
                using (var createQueue = createQueueContainer.GetQueueCreation<SqLiteMessageQueueCreation>(queueConnection))
                {
                    createQueue.CreateQueue();
                    return createQueue.Scope; //hold onto the scope to prevent the queue from being deleted
                }
            }
        }
    }
}