Skip to content

Commit f6ea15b

Browse files
zaruszzarusz
committed
Reduce relational transport duplication
Signed-off-by: Tomasz Maruszak <maruszaktomasz@gmail.com>
1 parent 47b0984 commit f6ea15b

35 files changed

Lines changed: 700 additions & 715 deletions

‎build/tasks.ps1‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@ $projects = @(
3232
"SlimMessageBus.Host.Redis",
3333
"SlimMessageBus.Host.Mqtt",
3434
"SlimMessageBus.Host.RabbitMQ",
35+
"SlimMessageBus.Host.Relational",
3536
"SlimMessageBus.Host.Sql",
3637
"SlimMessageBus.Host.Sql.Common",
3738
"SlimMessageBus.Host.PostgreSql",

‎src/SlimMessageBus.Host.PostgreSql/Config/PostgreSqlConsumerBuilderExtensions.cs‎

Lines changed: 4 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -5,38 +5,14 @@ public static class PostgreSqlConsumerBuilderExtensions
55
internal const string PropertySubscriptionName = "PostgreSql_SubscriptionName";
66

77
public static ConsumerBuilder<T> Queue<T>(this ConsumerBuilder<T> builder, string queue)
8-
{
9-
if (builder is null) throw new ArgumentNullException(nameof(builder));
10-
if (queue is null) throw new ArgumentNullException(nameof(queue));
11-
12-
builder.Path(queue);
13-
builder.ConsumerSettings.PathKind = PathKind.Queue;
14-
return builder;
15-
}
8+
=> RelationalBuilderExtensions.Queue(builder, queue);
169

1710
public static ConsumerBuilder<T> Topic<T>(this ConsumerBuilder<T> builder, string topic, string subscriptionName)
18-
{
19-
if (builder is null) throw new ArgumentNullException(nameof(builder));
20-
if (topic is null) throw new ArgumentNullException(nameof(topic));
21-
if (subscriptionName is null) throw new ArgumentNullException(nameof(subscriptionName));
22-
23-
builder.Topic(topic);
24-
builder.ConsumerSettings.PathKind = PathKind.Topic;
25-
builder.ConsumerSettings.Properties[PropertySubscriptionName] = subscriptionName;
26-
return builder;
27-
}
11+
=> RelationalBuilderExtensions.Topic(builder, topic, subscriptionName, PropertySubscriptionName);
2812

2913
public static ConsumerBuilder<T> Subscription<T>(this ConsumerBuilder<T> builder, string subscriptionName)
30-
{
31-
if (builder is null) throw new ArgumentNullException(nameof(builder));
32-
if (subscriptionName is null) throw new ArgumentNullException(nameof(subscriptionName));
33-
34-
builder.ConsumerSettings.Properties[PropertySubscriptionName] = subscriptionName;
35-
return builder;
36-
}
14+
=> RelationalBuilderExtensions.Subscription(builder, subscriptionName, PropertySubscriptionName);
3715

3816
internal static string GetSubscriptionName(this AbstractConsumerSettings settings)
39-
=> settings.Properties.TryGetValue(PropertySubscriptionName, out var value)
40-
? (string)value
41-
: settings.Path;
17+
=> RelationalBuilderExtensions.GetSubscriptionName(settings, PropertySubscriptionName);
4218
}

‎src/SlimMessageBus.Host.PostgreSql/Config/PostgreSqlHandlerBuilderExtensions.cs‎

Lines changed: 4 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -3,46 +3,14 @@ namespace SlimMessageBus.Host.PostgreSql;
33
public static class PostgreSqlHandlerBuilderExtensions
44
{
55
public static HandlerBuilder<TRequest, TResponse> Queue<TRequest, TResponse>(this HandlerBuilder<TRequest, TResponse> builder, string queue)
6-
{
7-
if (builder is null) throw new ArgumentNullException(nameof(builder));
8-
if (queue is null) throw new ArgumentNullException(nameof(queue));
9-
10-
builder.Path(queue);
11-
builder.ConsumerSettings.PathKind = PathKind.Queue;
12-
return builder;
13-
}
6+
=> RelationalBuilderExtensions.Queue(builder, queue);
147

158
public static HandlerBuilder<TRequest> Queue<TRequest>(this HandlerBuilder<TRequest> builder, string queue)
16-
{
17-
if (builder is null) throw new ArgumentNullException(nameof(builder));
18-
if (queue is null) throw new ArgumentNullException(nameof(queue));
19-
20-
builder.Path(queue);
21-
builder.ConsumerSettings.PathKind = PathKind.Queue;
22-
return builder;
23-
}
9+
=> RelationalBuilderExtensions.Queue(builder, queue);
2410

2511
public static HandlerBuilder<TRequest, TResponse> Topic<TRequest, TResponse>(this HandlerBuilder<TRequest, TResponse> builder, string topic, string subscriptionName)
26-
{
27-
if (builder is null) throw new ArgumentNullException(nameof(builder));
28-
if (topic is null) throw new ArgumentNullException(nameof(topic));
29-
if (subscriptionName is null) throw new ArgumentNullException(nameof(subscriptionName));
30-
31-
builder.Path(topic);
32-
builder.ConsumerSettings.PathKind = PathKind.Topic;
33-
builder.ConsumerSettings.Properties[PostgreSqlConsumerBuilderExtensions.PropertySubscriptionName] = subscriptionName;
34-
return builder;
35-
}
12+
=> RelationalBuilderExtensions.Topic(builder, topic, subscriptionName, PostgreSqlConsumerBuilderExtensions.PropertySubscriptionName);
3613

3714
public static HandlerBuilder<TRequest> Topic<TRequest>(this HandlerBuilder<TRequest> builder, string topic, string subscriptionName)
38-
{
39-
if (builder is null) throw new ArgumentNullException(nameof(builder));
40-
if (topic is null) throw new ArgumentNullException(nameof(topic));
41-
if (subscriptionName is null) throw new ArgumentNullException(nameof(subscriptionName));
42-
43-
builder.Path(topic);
44-
builder.ConsumerSettings.PathKind = PathKind.Topic;
45-
builder.ConsumerSettings.Properties[PostgreSqlConsumerBuilderExtensions.PropertySubscriptionName] = subscriptionName;
46-
return builder;
47-
}
15+
=> RelationalBuilderExtensions.Topic(builder, topic, subscriptionName, PostgreSqlConsumerBuilderExtensions.PropertySubscriptionName);
4816
}

‎src/SlimMessageBus.Host.PostgreSql/Config/PostgreSqlProducerBuilderExtensions.cs‎

Lines changed: 3 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -3,28 +3,11 @@ namespace SlimMessageBus.Host.PostgreSql;
33
public static class PostgreSqlProducerBuilderExtensions
44
{
55
public static ProducerBuilder<T> DefaultQueue<T>(this ProducerBuilder<T> producerBuilder, string queue)
6-
{
7-
if (producerBuilder is null) throw new ArgumentNullException(nameof(producerBuilder));
8-
if (queue is null) throw new ArgumentNullException(nameof(queue));
9-
10-
producerBuilder.DefaultTopic(queue);
11-
producerBuilder.ToQueue();
12-
return producerBuilder;
13-
}
6+
=> RelationalBuilderExtensions.DefaultQueue(producerBuilder, queue);
147

158
public static ProducerBuilder<T> ToTopic<T>(this ProducerBuilder<T> producerBuilder)
16-
{
17-
if (producerBuilder is null) throw new ArgumentNullException(nameof(producerBuilder));
18-
19-
producerBuilder.Settings.PathKind = PathKind.Topic;
20-
return producerBuilder;
21-
}
9+
=> RelationalBuilderExtensions.ToTopic(producerBuilder);
2210

2311
public static ProducerBuilder<T> ToQueue<T>(this ProducerBuilder<T> producerBuilder)
24-
{
25-
if (producerBuilder is null) throw new ArgumentNullException(nameof(producerBuilder));
26-
27-
producerBuilder.Settings.PathKind = PathKind.Queue;
28-
return producerBuilder;
29-
}
12+
=> RelationalBuilderExtensions.ToQueue(producerBuilder);
3013
}

‎src/SlimMessageBus.Host.PostgreSql/Config/PostgreSqlRequestResponseBuilderExtensions.cs‎

Lines changed: 2 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -3,24 +3,8 @@ namespace SlimMessageBus.Host.PostgreSql;
33
public static class PostgreSqlRequestResponseBuilderExtensions
44
{
55
public static RequestResponseBuilder ReplyToQueue(this RequestResponseBuilder builder, string queue)
6-
{
7-
if (builder is null) throw new ArgumentNullException(nameof(builder));
8-
if (queue is null) throw new ArgumentNullException(nameof(queue));
9-
10-
builder.Settings.Path = queue;
11-
builder.Settings.PathKind = PathKind.Queue;
12-
return builder;
13-
}
6+
=> RelationalBuilderExtensions.ReplyToQueue(builder, queue);
147

158
public static RequestResponseBuilder ReplyToTopic(this RequestResponseBuilder builder, string topic, string subscriptionName)
16-
{
17-
if (builder is null) throw new ArgumentNullException(nameof(builder));
18-
if (topic is null) throw new ArgumentNullException(nameof(topic));
19-
if (subscriptionName is null) throw new ArgumentNullException(nameof(subscriptionName));
20-
21-
builder.Settings.Path = topic;
22-
builder.Settings.PathKind = PathKind.Topic;
23-
builder.Settings.Properties[PostgreSqlConsumerBuilderExtensions.PropertySubscriptionName] = subscriptionName;
24-
return builder;
25-
}
9+
=> RelationalBuilderExtensions.ReplyToTopic(builder, topic, subscriptionName, PostgreSqlConsumerBuilderExtensions.PropertySubscriptionName);
2610
}

‎src/SlimMessageBus.Host.PostgreSql/Configuration/PostgreSqlMessageBusSettings.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
namespace SlimMessageBus.Host.PostgreSql;
22

3-
public class PostgreSqlMessageBusSettings
3+
public class PostgreSqlMessageBusSettings : IRelationalMessageBusSettings
44
{
55
public string? ConnectionString { get; set; }
66
public string DatabaseSchemaName { get; set; } = "public";

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

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

3-
public interface IPostgreSqlRepository
3+
public interface IPostgreSqlRepository : IRelationalRepository<PostgreSqlTransportMessage>
44
{
55
NpgsqlConnection Connection { get; }
66
Task EnsureConnection(CancellationToken cancellationToken);
7-
Task UpsertSubscription(string path, string subscriptionName, CancellationToken cancellationToken);
8-
Task<IReadOnlyCollection<string>> GetSubscriptions(string path, CancellationToken cancellationToken);
9-
Task Insert(PostgreSqlTransportMessage message, CancellationToken cancellationToken);
10-
Task<IReadOnlyCollection<PostgreSqlTransportMessage>> LockAndSelect(string path, PathKind pathKind, string? subscriptionName, string instanceId, int batchSize, TimeSpan lockDuration, CancellationToken cancellationToken);
11-
Task Complete(IReadOnlyCollection<PostgreSqlTransportMessage> messages, CancellationToken cancellationToken);
12-
Task Fail(IReadOnlyCollection<PostgreSqlTransportMessage> messages, int maxDeliveryAttempts, CancellationToken cancellationToken);
137
}

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

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

3-
using System.Diagnostics;
4-
5-
public class PostgreSqlConsumer : AbstractConsumer
3+
public class PostgreSqlConsumer : RelationalConsumerBase<PostgreSqlTransportMessage, IPostgreSqlRepository>
64
{
7-
private readonly IServiceProvider _serviceProvider;
8-
private readonly PostgreSqlMessageBusSettings _providerSettings;
9-
private readonly IMessageProcessor<PostgreSqlTransportMessage> _messageProcessor;
10-
private readonly PathKind _pathKind;
11-
private readonly string? _subscriptionName;
12-
private readonly string _instanceId;
13-
private Task? _task;
14-
155
public PostgreSqlConsumer(
166
ILogger<PostgreSqlConsumer> logger,
177
IEnumerable<AbstractConsumerSettings> consumerSettings,
@@ -23,112 +13,7 @@ public PostgreSqlConsumer(
2313
PathKind pathKind,
2414
string? subscriptionName,
2515
string instanceId)
26-
: base(logger, consumerSettings, path, interceptors)
27-
{
28-
_serviceProvider = serviceProvider;
29-
_providerSettings = providerSettings;
30-
_messageProcessor = messageProcessor;
31-
_pathKind = pathKind;
32-
_subscriptionName = subscriptionName;
33-
_instanceId = instanceId;
34-
}
35-
36-
protected override Task OnStart()
37-
{
38-
_task = Run();
39-
return Task.CompletedTask;
40-
}
41-
42-
protected override async Task OnStop()
43-
{
44-
if (_task != null)
45-
{
46-
try
47-
{
48-
await _task.ConfigureAwait(false);
49-
}
50-
catch (OperationCanceledException) when (CancellationToken.IsCancellationRequested)
51-
{
52-
Logger.LogDebug("PostgreSQL consumer for path {Path} stopped", Path);
53-
}
54-
_task = null;
55-
}
56-
}
57-
58-
private async Task Run()
59-
{
60-
var idle = Stopwatch.StartNew();
61-
62-
while (!CancellationToken.IsCancellationRequested)
63-
{
64-
try
65-
{
66-
await PollOnce(idle).ConfigureAwait(false);
67-
}
68-
catch (OperationCanceledException) when (CancellationToken.IsCancellationRequested)
69-
{
70-
return;
71-
}
72-
catch (Exception e) when (!CancellationToken.IsCancellationRequested)
73-
{
74-
Logger.LogWarning(e, "Error occurred while polling PostgreSQL path {Path}", Path);
75-
try
76-
{
77-
await Task.Delay(_providerSettings.PollDelay, CancellationToken).ConfigureAwait(false);
78-
}
79-
catch (OperationCanceledException) when (CancellationToken.IsCancellationRequested)
80-
{
81-
return;
82-
}
83-
}
84-
}
85-
}
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)
16+
: base(logger, consumerSettings, interceptors, serviceProvider, providerSettings, messageProcessor, path, pathKind, subscriptionName, instanceId, "PostgreSQL")
11917
{
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-
}
13318
}
13419
}

0 commit comments

Comments
 (0)