Postgresql transport #111

viktor-h · 2024-04-04T14:12:55Z

No description provided.

niclaslindstedt · 2024-04-15T07:29:33Z

knightbus-postgresql/src/KnightBus.PostgreSql.Management/PostgresManagementClient.cs

+            var propertiesOrdinal = reader.GetOrdinal("properties");
+            var isPropertiesNull = reader.IsDBNull(propertiesOrdinal);
+
+            var postgresMessage = new PostgresMessage<DictionaryMessage>
+            {
+                Id = reader.GetInt64(reader.GetOrdinal("message_id")),
+                ReadCount = reader.GetInt32(reader.GetOrdinal("read_count")),


You could move the GetOrdinal functions outwide the while loop since they won't change during the loop.

niclaslindstedt · 2024-04-15T07:48:05Z

knightbus/examples/KnightBus.Examples.PostgreSql/Program.cs

+        // var commands = new List<SamplePostgresMessage>();
+        // for (int i = 0; i < 10000; i++)
+        // {
+        //     commands.Add(new SamplePostgresMessage { MessageBody = $"{i}_{Guid.NewGuid()}" });
+        // }
+        // var chunks = commands.Chunk(1000);
+        // foreach (var chunk in chunks)
+        // {
+        //     await client.SendAsync(chunk);
+        // }
+
+        await client.SendAsync(new SamplePostgresMessage { MessageBody = Guid.NewGuid().ToString() });
+
+        //await client.SendAsync(new SamplePoisonPostgresMessage { MessageBody = $"error_{Guid.NewGuid()}" } );


Should this commented-out code be removed?

niclaslindstedt

Very nice :)

NiklasArbin · 2024-04-15T08:26:28Z

knightbus-postgresql/src/KnightBus.PostgreSql/PostgresChannelReceiver.cs

+    private readonly IMessageSerializer _serializer;
+    private readonly SemaphoreSlim _maxConcurrent;
+    private CancellationTokenSource _pumpDelayCancellationTokenSource = new();
+    private readonly TimeSpan _pollingSleepInterval = TimeSpan.FromSeconds(5);


Should this be configurable? 5 seconds might be too long in some scenarios

yeah I will have a go at it.

I would also like to explore a longer interval combined with LISTEN/NOTIFY. But that would still require polling since notifications are ephemeral.

NiklasArbin · 2024-04-15T08:27:09Z

knightbus-postgresql/src/KnightBus.PostgreSql/PostgresChannelReceiver.cs

+        _maxConcurrent = new SemaphoreSlim(settings.MaxConcurrentCalls);
+    }
+
+    public async Task StartAsync(CancellationToken cancellationToken)


Fix this warning. Return the task or Task.Completed

NiklasArbin · 2024-04-15T08:34:17Z

knightbus-postgresql/src/KnightBus.PostgreSql/NpgsqlDataReaderExtensions.cs

+
+public static class NpgsqlDataReaderExtensions
+{
+    public static async Task<List<PostgresMessage<T>>> ReadDeadLetterRows<T>(


IAsyncEnumerable instead?

This will copy arrays multiple times in memory of not either sized properly or handled as IEnumerable

NiklasArbin · 2024-04-15T08:37:16Z

knightbus-postgresql/src/KnightBus.PostgreSql/PostgresBus.cs

+            var mBody = _serializer.Serialize(messagesList[i]);
+            command.Parameters.Add(new NpgsqlParameter { Value = mBody, NpgsqlDbType = NpgsqlDbType.Jsonb });
+        }
+        var stringValues = values.ToString().TrimEnd(',');


Optimization: do not add ',' if last row instead of modifying the entire string every time

Yes let's get rid of that. Especially better for large batch inserts when the cmd text can get big.

NiklasArbin · 2024-04-15T08:37:59Z

knightbus-postgresql/src/KnightBus.PostgreSql/PostgresBus.cs

+
+namespace KnightBus.PostgreSql;
+
+public interface IPostgresBus


Add CancellationToken support!

NiklasArbin · 2024-04-15T08:41:51Z

knightbus-postgresql/src/KnightBus.PostgreSql/PostgresQueueClient.cs

+    }
+
+    public async Task<List<PostgresMessage<T>>> GetMessagesAsync(int count, int visibilityTimeout)
+    {


CancellationToken?

hmm yes GetMessagesAsync should have support for cancellation. But not CompleteAsync, AbandonByErrorAsync and DeadLetterMessageAsync if we are in the midst of processing a message during shutdown. But no new ones should be fetched.

NiklasArbin · 2024-04-15T08:42:57Z

knightbus-postgresql/src/KnightBus.PostgreSql/QueueInitializer.cs

+
+public static class QueueInitializer
+{
+    public static async Task InitQueue(PostgresQueueName queueName, NpgsqlDataSource npgsqlDataSource)


How will you manage to update to a newer version? Should schema version be included somewhere?

Need to think a bit about how this should be handled. I have deliberately left it out because I might have to invent some migration support that works for a lot of tables sharing the same schema.

NiklasArbin

Impressive work!

viktor-h · 2024-04-15T11:31:56Z

Impressive work!

thanks!

mbaneryd · 2024-04-15T11:44:09Z

knightbus-postgresql/src/KnightBus.PostgreSql/QueueInitializer.cs

+        await using var transaction = await connection.BeginTransactionAsync();
+
+        await using var createSchema = new NpgsqlCommand(@$"
+ CREATE SCHEMA IF NOT EXISTS {SchemaName};


Do we have an idea on how to handle schema updates in the future, if needed?

Even though it only uses a small set of tables we might end up in a situation where we want to add functionality. For example a priority column to order by?

I can think of two ways at the moment. Run migrations for each queue on first application start YOLO style or have some bundled CLI tool that does all the queue creations and migrations with a connection string.

NiklasArbin · 2024-04-16T12:13:55Z

knightbus-postgresql/src/KnightBus.PostgreSql/PostgresChannelReceiver.cs

+            foreach (var postgresMessage in messages)
+            {
+                await _maxConcurrent.WaitAsync(cancellationToken).ConfigureAwait(false);
+                var timeoutToken = new CancellationTokenSource(Settings.MessageLockTimeout);


This will not work properly, the timer starts ticking when you get the messages, but you start the timer for the token after waiting for the semaphore.

472ad53 not perfect but should align better with message fetch time.

hmm this won't work. The timeoutToken will get disposed if declared once above message loop.

This reverts commit 8460079.

…d on string

viktor-h force-pushed the postgresql-transport branch from 2630738 to 7bacfdc Compare April 14, 2024 14:42

viktor-h marked this pull request as ready for review April 14, 2024 21:20

viktor-h requested a review from NiklasArbin April 15, 2024 07:02

niclaslindstedt reviewed Apr 15, 2024

View reviewed changes

niclaslindstedt approved these changes Apr 15, 2024

View reviewed changes

NiklasArbin reviewed Apr 15, 2024

View reviewed changes

NiklasArbin requested changes Apr 15, 2024

View reviewed changes

NiklasArbin approved these changes Apr 15, 2024

View reviewed changes

mbaneryd reviewed Apr 15, 2024

View reviewed changes

NiklasArbin reviewed Apr 16, 2024

View reviewed changes

viktor-h force-pushed the postgresql-transport branch 3 times, most recently from a043efa to 9cba7dd Compare April 16, 2024 21:17

Viktor Hartenberger added 15 commits April 17, 2024 12:23

wip

cb99307

wiring

78130e1

Add example

c8a9724

Add scheduling support

e2fec04

Auto init, add constants, remove warnings and general clean up

690afdb

Add note about future investigation

244ac62

Add typed parameter for visibility timeout

e57ff08

Add support for delete queue

0a9b655

Add deadletter to example

55c260c

Prepare thy statements

40e21e0

Remove sql cheat sheet

fb559b6

review performance

387f333

encapsulate queue name validation

88e4454

Use StringBuilder

311b911

Management client and tests

f3dce37

Viktor Hartenberger added 23 commits April 17, 2024 12:23

Move queue init to management client

84b15e7

Dispose connections when receiving stop signal

61f573b

Implement MessageStateHandlerTests for postgres

27609ed

Add peek dlq support for postgres queue client

e2a8f99

Create schema if not created during queue init

035d5f2

test clean up

197a517

Extract managment code

5add7f4

Extract queue client query text

2d5fd05

Revert "Extract queue client query text"

12008dd

This reverts commit 8460079.

Edit csproj for packages

c804166

Add logos

58d52a3

Call getOrdinal outside loops and extract message row read

6ded313

Remove unused example code

8602398

Fix warning

2bb5f31

Add cancellation token support for bus

ed9f3df

Do not add trailing comma on last iteration instead of calling TrimEn…

2ca139e

…d on string

CancellationToken support for GetMessagesAsync

d88ebcd

Save 2 bytes

4faf4bc

Configurable polling sleep interval

3620af0

Fix calculated timeout token

f9ffdca

IAsyncEnumerable

f6062f0

MS serializer

3487d98

remove warnings

2ff5154

viktor-h force-pushed the postgresql-transport branch from 189a9c8 to 2ff5154 Compare April 17, 2024 10:23

Viktor Hartenberger added 2 commits April 17, 2024 13:04

Implement generic message pump

820f3fb

Fix warning

1dbe1e6

viktor-h merged commit 3c686c6 into master Apr 17, 2024
3 checks passed

viktor-h deleted the postgresql-transport branch April 17, 2024 11:26

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Postgresql transport #111

Postgresql transport #111

viktor-h commented Apr 4, 2024

niclaslindstedt Apr 15, 2024 •

edited

Loading

viktor-h Apr 15, 2024

niclaslindstedt Apr 15, 2024

niclaslindstedt left a comment

NiklasArbin Apr 15, 2024

viktor-h Apr 15, 2024

NiklasArbin Apr 15, 2024

NiklasArbin Apr 15, 2024

NiklasArbin Apr 16, 2024

NiklasArbin Apr 15, 2024

viktor-h Apr 15, 2024

NiklasArbin Apr 15, 2024

NiklasArbin Apr 15, 2024

viktor-h Apr 15, 2024

NiklasArbin Apr 15, 2024

viktor-h Apr 15, 2024

NiklasArbin left a comment

viktor-h commented Apr 15, 2024

mbaneryd Apr 15, 2024

viktor-h Apr 15, 2024

NiklasArbin Apr 16, 2024

viktor-h Apr 16, 2024

viktor-h Apr 16, 2024


		namespace KnightBus.PostgreSql;

		public interface IPostgresBus

Postgresql transport #111

Postgresql transport #111

Conversation

viktor-h commented Apr 4, 2024

niclaslindstedt Apr 15, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

niclaslindstedt left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

NiklasArbin left a comment

Choose a reason for hiding this comment

viktor-h commented Apr 15, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

niclaslindstedt Apr 15, 2024 •

edited

Loading