PostGreTransport

The PostGre transport uses SQL tables to store messages. It supports the following features:

• Delayed messages
• Message expiration
• Message priority
• Routing
• History tracking
• User-defined metadata columns
• Transactional, status-based, or FIFO processing

You must create the queue before you use the PostGre transport. You can create it manually or with the queue creation classes.

• Queue Creation Options

//Create the queue if it doesn't exist
var queueName = "testing";
var connectionString = "Server=V-SQL;Port=5432;Database=IntegrationTesting;Integrated Security=true;";
var queueConnection = new QueueConnection(queueName, connectionString);
using (var createQueueContainer = new QueueCreationContainer<PostgreSqlMessageQueueInit>())
{
	using (var createQueue = createQueueContainer.GetQueueCreation<PostgreSqlMessageQueueCreation>(queueConnection))
	{
		if (!createQueue.QueueExists)
		{
			createQueue.CreateQueue();
		}
	}
}

• PostGre Options

[Producer]

var queueConnection = new QueueConnection(queueName, connectionString);
using (var queueContainer = new QueueContainer<PostgreSqlMessageQueueInit>())
{
	using (var queue = queueContainer.CreateProducer<SimpleMessage>(queueConnection))
    {
		queue.Send(new SimpleMessage {Message = "Hello World"});
    }
}

[Consumer]

var queueConnection = new QueueConnection(queueName, connectionString);
using (var queueContainer = new QueueContainer<PostgreSqlMessageQueueInit>())
{
	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.

To enqueue a queue message inside your own business transaction (transactional outbox pattern), see Outbox Pattern.

Npgsql can prepare statements automatically. This feature is off by default in Npgsql, and this transport does not turn it on. But the send path issues the same handful of statements again and again, and automatic preparation targets exactly that pattern. To enable it, change the connection string. You do not need to change code:

Host=localhost;Database=MyDb;Username=user;Password=pass;Max Auto Prepare=20;Auto Prepare Min Usages=2;

Measured on this transport (net10, LAN PostgreSQL, 256-byte payload), across two runs:

Allocation is unchanged. The benefit is on the batch path only. A single send showed a 6% gain in one run and no gain in the other. That result is noise, not a small win. If you send one message at a time, this setting is not for you.

⚠️ Know the trade-off before enabling it. Prepared statements live on the physical connection, and the pool hands that connection out again. As a result, when a table is dropped or recreated, every prepared statement that references it becomes invalid. This library creates and drops queues as a normal operation, so this case is not hypothetical. An integration test covers a create/drop/recreate cycle, and the test does not fail. But Npgsql exposes no public counter for auto-prepared statements. So the test shows that the scenario works, but it does not prove that the test exercised the invalidation path.

If your application creates and drops queues at runtime, measure your own workload before turning this on. If your queues are long-lived and you send in batches, it is close to free.

See the PostgreSQL samples in the DotNetWorkQueue.Samples repository.