Skip to content

Commit 47b0984

Browse files
zaruszzarusz
committed
Address SonarCloud findings
Signed-off-by: Tomasz Maruszak <maruszaktomasz@gmail.com>
1 parent 220ef24 commit 47b0984

13 files changed

Lines changed: 187 additions & 106 deletions

File tree

‎src/SlimMessageBus.Host.PostgreSql/ObjectToInferredTypesConverter.cs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ namespace SlimMessageBus.Host.PostgreSql;
55

66
public class ObjectToInferredTypesConverter : JsonConverter<object>
77
{
8-
public override object? Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options) =>
8+
public override object Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options) =>
99
reader.TokenType switch
1010
{
1111
JsonTokenType.True => true,
@@ -14,7 +14,7 @@ JsonTokenType.Number when reader.TryGetInt32(out var i) => i,
1414
JsonTokenType.Number when reader.TryGetInt64(out var l) => l,
1515
JsonTokenType.Number => reader.GetDouble(),
1616
JsonTokenType.String when reader.TryGetDateTime(out var datetime) => datetime,
17-
JsonTokenType.String => reader.GetString(),
17+
JsonTokenType.String => reader.GetString()!,
1818
_ => JsonDocument.ParseValue(ref reader).RootElement.Clone()
1919
};
2020

‎src/SlimMessageBus.Host.PostgreSql/PostgreSqlConsumer.cs‎

Lines changed: 50 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@ protected override async Task OnStop()
4949
}
5050
catch (OperationCanceledException) when (CancellationToken.IsCancellationRequested)
5151
{
52+
Logger.LogDebug("PostgreSQL consumer for path {Path} stopped", Path);
5253
}
5354
_task = null;
5455
}
@@ -62,41 +63,7 @@ private async Task Run()
6263
{
6364
try
6465
{
65-
var scope = _serviceProvider.CreateScope();
66-
try
67-
{
68-
var repository = scope.ServiceProvider.GetRequiredService<IPostgreSqlRepository>();
69-
var messages = await repository.LockAndSelect(Path, _pathKind, _subscriptionName, _instanceId, _providerSettings.PollBatchSize, _providerSettings.LockDuration, CancellationToken).ConfigureAwait(false);
70-
71-
if (messages.Count == 0)
72-
{
73-
if (idle.Elapsed >= _providerSettings.PollDelay)
74-
{
75-
await Task.Delay(_providerSettings.PollDelay, CancellationToken).ConfigureAwait(false);
76-
}
77-
continue;
78-
}
79-
80-
idle.Restart();
81-
82-
foreach (var message in messages)
83-
{
84-
var result = await _messageProcessor.ProcessMessage(message, message.Headers, cancellationToken: CancellationToken).ConfigureAwait(false);
85-
if (result.Exception == null)
86-
{
87-
await repository.Complete([message], CancellationToken).ConfigureAwait(false);
88-
}
89-
else
90-
{
91-
Logger.LogError(result.Exception, "Error occurred while processing PostgreSQL message {MessageId} on {Path}", message.Id, Path);
92-
await repository.Fail([message], _providerSettings.MaxDeliveryAttempts, CancellationToken).ConfigureAwait(false);
93-
}
94-
}
95-
}
96-
finally
97-
{
98-
await scope.DisposeAsyncScope().ConfigureAwait(false);
99-
}
66+
await PollOnce(idle).ConfigureAwait(false);
10067
}
10168
catch (OperationCanceledException) when (CancellationToken.IsCancellationRequested)
10269
{
@@ -116,4 +83,52 @@ private async Task Run()
11683
}
11784
}
11885
}
86+
87+
private async Task PollOnce(Stopwatch idle)
88+
{
89+
var scope = _serviceProvider.CreateScope();
90+
try
91+
{
92+
var repository = scope.ServiceProvider.GetRequiredService<IPostgreSqlRepository>();
93+
var messages = await repository.LockAndSelect(Path, _pathKind, _subscriptionName, _instanceId, _providerSettings.PollBatchSize, _providerSettings.LockDuration, CancellationToken).ConfigureAwait(false);
94+
95+
if (messages.Count == 0)
96+
{
97+
await DelayWhenIdle(idle).ConfigureAwait(false);
98+
return;
99+
}
100+
101+
idle.Restart();
102+
await ProcessMessages(repository, messages).ConfigureAwait(false);
103+
}
104+
finally
105+
{
106+
await scope.DisposeAsyncScope().ConfigureAwait(false);
107+
}
108+
}
109+
110+
private async Task DelayWhenIdle(Stopwatch idle)
111+
{
112+
if (idle.Elapsed >= _providerSettings.PollDelay)
113+
{
114+
await Task.Delay(_providerSettings.PollDelay, CancellationToken).ConfigureAwait(false);
115+
}
116+
}
117+
118+
private async Task ProcessMessages(IPostgreSqlRepository repository, IReadOnlyCollection<PostgreSqlTransportMessage> messages)
119+
{
120+
foreach (var message in messages)
121+
{
122+
var result = await _messageProcessor.ProcessMessage(message, message.Headers, cancellationToken: CancellationToken).ConfigureAwait(false);
123+
if (result.Exception == null)
124+
{
125+
await repository.Complete([message], CancellationToken).ConfigureAwait(false);
126+
}
127+
else
128+
{
129+
Logger.LogError(result.Exception, "Error occurred while processing PostgreSQL message {MessageId} on {Path}", message.Id, Path);
130+
await repository.Fail([message], _providerSettings.MaxDeliveryAttempts, CancellationToken).ConfigureAwait(false);
131+
}
132+
}
133+
}
119134
}

‎src/SlimMessageBus.Host.PostgreSql/PostgreSqlHelper.cs‎

Lines changed: 32 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,7 @@
11
namespace SlimMessageBus.Host.PostgreSql;
22

3-
using System.Text.RegularExpressions;
4-
53
public static partial class PostgreSqlHelper
64
{
7-
private static readonly Regex SafeIdentifierRegex = new("^[A-Za-z_][A-Za-z0-9_]*$", RegexOptions.CultureInvariant);
8-
95
private static readonly HashSet<int> TransientErrorNumbers =
106
[
117
40001,
@@ -54,14 +50,33 @@ public static Task RetryIfTransientError(ILogger logger, RetrySettings retrySett
5450

5551
public static string QuoteIdentifier(string identifier, string name)
5652
{
57-
if (string.IsNullOrWhiteSpace(identifier) || !SafeIdentifierRegex.IsMatch(identifier))
53+
if (!IsSafeIdentifier(identifier))
5854
{
5955
throw new ArgumentException("PostgreSQL identifiers can only contain letters, numbers, and underscores, and cannot start with a number.", name);
6056
}
6157

6258
return $"\"{identifier}\"";
6359
}
6460

61+
private static bool IsSafeIdentifier(string identifier)
62+
{
63+
if (string.IsNullOrWhiteSpace(identifier) || !(char.IsLetter(identifier[0]) || identifier[0] == '_'))
64+
{
65+
return false;
66+
}
67+
68+
for (var i = 1; i < identifier.Length; i++)
69+
{
70+
var c = identifier[i];
71+
if (!(char.IsLetterOrDigit(c) || c == '_'))
72+
{
73+
return false;
74+
}
75+
}
76+
77+
return true;
78+
}
79+
6580
[LoggerMessage(
6681
EventId = 0,
6782
Level = LogLevel.Information,
@@ -80,9 +95,19 @@ public static string QuoteIdentifier(string identifier, string name)
8095
partial class PostgreSqlHelper
8196
{
8297
private static partial void LogError(ILogger logger, int retryCount, int retryNumber)
83-
=> logger.LogInformation("PostgreSQL error encountered. Will begin attempt number {RetryNumber} of {RetryCount} max...", retryNumber, retryCount);
98+
{
99+
if (logger.IsEnabled(LogLevel.Information))
100+
{
101+
logger.LogInformation("PostgreSQL error encountered. Will begin attempt number {RetryNumber} of {RetryCount} max...", retryNumber, retryCount);
102+
}
103+
}
84104

85105
private static partial void LogWillRetry(ILogger logger, int errorCode, NpgsqlException e)
86-
=> logger.LogDebug(e, "PostgreSQL error occurred {ErrorCode}. Will retry operation", errorCode);
106+
{
107+
if (logger.IsEnabled(LogLevel.Debug))
108+
{
109+
logger.LogDebug(e, "PostgreSQL error occurred {ErrorCode}. Will retry operation", errorCode);
110+
}
111+
}
87112
}
88113
#endif

‎src/SlimMessageBus.Host.PostgreSql/PostgreSqlRepository.cs‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,8 @@ namespace SlimMessageBus.Host.PostgreSql;
22

33
public class PostgreSqlRepository : IPostgreSqlRepository
44
{
5+
private const string SubscriptionNameParameter = "subscription_name";
6+
57
private static readonly JsonSerializerOptions _jsonOptions = new()
68
{
79
Converters = { new ObjectToInferredTypesConverter() }
@@ -35,7 +37,7 @@ public async Task UpsertSubscription(string path, string subscriptionName, Cance
3537
await ExecuteNonQuery(_settings.OperationRetry, _template.UpsertSubscription, cmd =>
3638
{
3739
cmd.Parameters.Add("path", NpgsqlDbType.Text).Value = path;
38-
cmd.Parameters.Add("subscription_name", NpgsqlDbType.Text).Value = subscriptionName;
40+
cmd.Parameters.Add(SubscriptionNameParameter, NpgsqlDbType.Text).Value = subscriptionName;
3941
}, cancellationToken).ConfigureAwait(false);
4042
}
4143

@@ -75,7 +77,7 @@ await ExecuteScalarAsync(_settings.OperationRetry, insertSql, cmd =>
7577
}
7678
cmd.Parameters.Add("path", NpgsqlDbType.Text).Value = message.Path;
7779
cmd.Parameters.Add("path_kind", NpgsqlDbType.Smallint).Value = (short)message.PathKind;
78-
cmd.Parameters.Add("subscription_name", NpgsqlDbType.Text).Value = (object?)message.SubscriptionName ?? DBNull.Value;
80+
cmd.Parameters.Add(SubscriptionNameParameter, NpgsqlDbType.Text).Value = (object?)message.SubscriptionName ?? DBNull.Value;
7981
cmd.Parameters.Add("message_type", NpgsqlDbType.Text).Value = message.MessageType;
8082
cmd.Parameters.Add("message_payload", NpgsqlDbType.Bytea).Value = message.MessagePayload;
8183
cmd.Parameters.Add("headers", NpgsqlDbType.Jsonb).Value = message.Headers != null ? JsonSerializer.Serialize(message.Headers, _jsonOptions) : DBNull.Value;
@@ -90,7 +92,7 @@ public async Task<IReadOnlyCollection<PostgreSqlTransportMessage>> LockAndSelect
9092
cmd.CommandText = _template.LockAndSelect;
9193
cmd.Parameters.Add("path", NpgsqlDbType.Text).Value = path;
9294
cmd.Parameters.Add("path_kind", NpgsqlDbType.Smallint).Value = (short)pathKind;
93-
cmd.Parameters.Add("subscription_name", NpgsqlDbType.Text).Value = (object?)subscriptionName ?? DBNull.Value;
95+
cmd.Parameters.Add(SubscriptionNameParameter, NpgsqlDbType.Text).Value = (object?)subscriptionName ?? DBNull.Value;
9496
cmd.Parameters.Add("instance_id", NpgsqlDbType.Text).Value = instanceId;
9597
cmd.Parameters.Add("batch_size", NpgsqlDbType.Integer).Value = batchSize;
9698
cmd.Parameters.Add("lock_duration", NpgsqlDbType.Interval).Value = lockDuration;

‎src/SlimMessageBus.Host.PostgreSql/PostgreSqlTopologyService.cs‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -100,11 +100,7 @@ await PostgreSqlHelper.RetryIfTransientError(_logger, _settings.SchemaCreationRe
100100
await cmd.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
101101
}
102102

103-
#if NETSTANDARD2_0
104-
transaction.Commit();
105-
#else
106103
await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
107-
#endif
108104
}, cancellationToken).ConfigureAwait(false);
109105
}
110106
}

‎src/SlimMessageBus.Host.PostgreSql/PostgreSqlTransportMessage.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -3,10 +3,10 @@ namespace SlimMessageBus.Host.PostgreSql;
33
public class PostgreSqlTransportMessage
44
{
55
public Guid Id { get; set; }
6-
public string? MessageType { get; set; }
7-
public byte[]? MessagePayload { get; set; }
6+
public string MessageType { get; set; } = null!;
7+
public byte[] MessagePayload { get; set; } = null!;
88
public Dictionary<string, object>? Headers { get; set; }
9-
public string? Path { get; set; }
9+
public string Path { get; set; } = null!;
1010
public PathKind PathKind { get; set; }
1111
public string? SubscriptionName { get; set; }
1212
}

‎src/SlimMessageBus.Host.Sql/SqlConsumer.cs‎

Lines changed: 50 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@ protected override async Task OnStop()
4949
}
5050
catch (OperationCanceledException) when (CancellationToken.IsCancellationRequested)
5151
{
52+
Logger.LogDebug("SQL consumer for path {Path} stopped", Path);
5253
}
5354
_task = null;
5455
}
@@ -62,41 +63,7 @@ private async Task Run()
6263
{
6364
try
6465
{
65-
var scope = _serviceProvider.CreateScope();
66-
try
67-
{
68-
var repository = scope.ServiceProvider.GetRequiredService<ISqlRepository>();
69-
var messages = await repository.LockAndSelect(Path, _pathKind, _subscriptionName, _instanceId, _providerSettings.PollBatchSize, _providerSettings.LockDuration, CancellationToken).ConfigureAwait(false);
70-
71-
if (messages.Count == 0)
72-
{
73-
if (idle.Elapsed >= _providerSettings.PollDelay)
74-
{
75-
await Task.Delay(_providerSettings.PollDelay, CancellationToken).ConfigureAwait(false);
76-
}
77-
continue;
78-
}
79-
80-
idle.Restart();
81-
82-
foreach (var message in messages)
83-
{
84-
var result = await _messageProcessor.ProcessMessage(message, message.Headers, cancellationToken: CancellationToken).ConfigureAwait(false);
85-
if (result.Exception == null)
86-
{
87-
await repository.Complete([message], CancellationToken).ConfigureAwait(false);
88-
}
89-
else
90-
{
91-
Logger.LogError(result.Exception, "Error occurred while processing SQL message {MessageId} on {Path}", message.Id, Path);
92-
await repository.Fail([message], _providerSettings.MaxDeliveryAttempts, CancellationToken).ConfigureAwait(false);
93-
}
94-
}
95-
}
96-
finally
97-
{
98-
await scope.DisposeAsyncScope().ConfigureAwait(false);
99-
}
66+
await PollOnce(idle).ConfigureAwait(false);
10067
}
10168
catch (OperationCanceledException) when (CancellationToken.IsCancellationRequested)
10269
{
@@ -116,4 +83,52 @@ private async Task Run()
11683
}
11784
}
11885
}
86+
87+
private async Task PollOnce(Stopwatch idle)
88+
{
89+
var scope = _serviceProvider.CreateScope();
90+
try
91+
{
92+
var repository = scope.ServiceProvider.GetRequiredService<ISqlRepository>();
93+
var messages = await repository.LockAndSelect(Path, _pathKind, _subscriptionName, _instanceId, _providerSettings.PollBatchSize, _providerSettings.LockDuration, CancellationToken).ConfigureAwait(false);
94+
95+
if (messages.Count == 0)
96+
{
97+
await DelayWhenIdle(idle).ConfigureAwait(false);
98+
return;
99+
}
100+
101+
idle.Restart();
102+
await ProcessMessages(repository, messages).ConfigureAwait(false);
103+
}
104+
finally
105+
{
106+
await scope.DisposeAsyncScope().ConfigureAwait(false);
107+
}
108+
}
109+
110+
private async Task DelayWhenIdle(Stopwatch idle)
111+
{
112+
if (idle.Elapsed >= _providerSettings.PollDelay)
113+
{
114+
await Task.Delay(_providerSettings.PollDelay, CancellationToken).ConfigureAwait(false);
115+
}
116+
}
117+
118+
private async Task ProcessMessages(ISqlRepository repository, IReadOnlyCollection<SqlTransportMessage> messages)
119+
{
120+
foreach (var message in messages)
121+
{
122+
var result = await _messageProcessor.ProcessMessage(message, message.Headers, cancellationToken: CancellationToken).ConfigureAwait(false);
123+
if (result.Exception == null)
124+
{
125+
await repository.Complete([message], CancellationToken).ConfigureAwait(false);
126+
}
127+
else
128+
{
129+
Logger.LogError(result.Exception, "Error occurred while processing SQL message {MessageId} on {Path}", message.Id, Path);
130+
await repository.Fail([message], _providerSettings.MaxDeliveryAttempts, CancellationToken).ConfigureAwait(false);
131+
}
132+
}
133+
}
119134
}

0 commit comments

Comments
 (0)