Skip to content

Commit d344d61

Browse files
michaelklishinmichaelklishin
authored
Merge pull request #915 from rabbitmq/backport-pr-910
Backport #910 to 6.x
2 parents 224134f + 644fb6c commit d344d61

File tree

9 files changed

+103
-161
lines changed

9 files changed

+103
-161
lines changed

‎projects/RabbitMQ.Client/client/events/AsyncEventingBasicConsumer.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -45,10 +45,10 @@ public override async Task HandleBasicConsumeOk(string consumerTag)
4545
}
4646

4747
///<summary>Fires the Received event.</summary>
48-
public override async Task HandleBasicDeliver(string consumerTag, ulong deliveryTag, bool redelivered, string exchange, string routingKey, IBasicProperties properties, ReadOnlyMemory<byte> body)
48+
public override Task HandleBasicDeliver(string consumerTag, ulong deliveryTag, bool redelivered, string exchange, string routingKey, IBasicProperties properties, ReadOnlyMemory<byte> body)
4949
{
50-
await base.HandleBasicDeliver(consumerTag, deliveryTag, redelivered, exchange, routingKey, properties, body).ConfigureAwait(false);
51-
await Received.InvokeAsync(this, new BasicDeliverEventArgs(consumerTag, deliveryTag, redelivered, exchange, routingKey, properties, body)).ConfigureAwait(false);
50+
// No need to call base, it's empty.
51+
return Received.InvokeAsync(this, new BasicDeliverEventArgs(consumerTag, deliveryTag, redelivered, exchange, routingKey, properties, body));
5252
}
5353

5454
///<summary>Fires the Shutdown event.</summary>

‎projects/RabbitMQ.Client/client/impl/AsyncConsumerDispatcher.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -66,7 +66,7 @@ public void HandleBasicCancel(IBasicConsumer consumer, string consumerTag)
6666
public void HandleModelShutdown(IBasicConsumer consumer, ShutdownEventArgs reason)
6767
{
6868
// the only case where we ignore the shutdown flag.
69-
Schedule(new ModelShutdown(consumer, reason));
69+
Schedule(new ModelShutdown(consumer, reason, _model));
7070
}
7171

7272
private void ScheduleUnlessShuttingDown<TWork>(TWork work)

‎projects/RabbitMQ.Client/client/impl/AsyncConsumerWorkService.cs‎

Lines changed: 35 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,10 @@
11
using System;
22
using System.Collections.Concurrent;
3+
using System.Collections.Generic;
34
using System.Threading;
45
using System.Threading.Channels;
56
using System.Threading.Tasks;
7+
using RabbitMQ.Client.Events;
68

79
namespace RabbitMQ.Client.Impl
810
{
@@ -79,29 +81,43 @@ public void Enqueue(Work work)
7981
_channel.Writer.TryWrite(work);
8082
}
8183

82-
async Task Loop()
84+
private async Task Loop()
8385
{
8486
while (await _channel.Reader.WaitToReadAsync().ConfigureAwait(false))
8587
{
8688
while (_channel.Reader.TryRead(out Work work))
8789
{
8890
try
8991
{
90-
Task task = work.Execute(_model);
92+
Task task = work.Execute();
9193
if (!task.IsCompleted)
9294
{
9395
await task.ConfigureAwait(false);
9496
}
9597
}
96-
catch(Exception)
98+
catch (Exception e)
9799
{
100+
if (!(_model is ModelBase modelBase))
101+
{
102+
return;
103+
}
98104

105+
var details = new Dictionary<string, object>
106+
{
107+
{ "consumer", work.Consumer },
108+
{ "context", work.Consumer }
109+
};
110+
modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
111+
}
112+
finally
113+
{
114+
work.PostExecute();
99115
}
100116
}
101117
}
102118
}
103119

104-
async Task LoopWithConcurrency(CancellationToken cancellationToken)
120+
private async Task LoopWithConcurrency(CancellationToken cancellationToken)
105121
{
106122
try
107123
{
@@ -125,22 +141,33 @@ async Task LoopWithConcurrency(CancellationToken cancellationToken)
125141
}
126142
}
127143

128-
static async Task HandleConcurrent(Work work, IModel model, SemaphoreSlim limiter)
144+
private static async Task HandleConcurrent(Work work, IModel model, SemaphoreSlim limiter)
129145
{
130146
try
131147
{
132-
Task task = work.Execute(model);
148+
Task task = work.Execute();
133149
if (!task.IsCompleted)
134150
{
135151
await task.ConfigureAwait(false);
136152
}
137153
}
138-
catch (Exception)
154+
catch (Exception e)
139155
{
140-
// ignored
156+
if (!(model is ModelBase modelBase))
157+
{
158+
return;
159+
}
160+
161+
var details = new Dictionary<string, object>
162+
{
163+
{ "consumer", work.Consumer },
164+
{ "context", work.Consumer }
165+
};
166+
modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
141167
}
142168
finally
143169
{
170+
work.PostExecute();
144171
limiter.Release();
145172
}
146173
}

‎projects/RabbitMQ.Client/client/impl/BasicCancel.cs‎

Lines changed: 7 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -1,40 +1,21 @@
1-
using System;
2-
using System.Collections.Generic;
3-
using System.Threading.Tasks;
4-
5-
using RabbitMQ.Client.Events;
1+
using System.Threading.Tasks;
62

73
namespace RabbitMQ.Client.Impl
84
{
9-
sealed class BasicCancel : Work
5+
internal sealed class BasicCancel : Work
106
{
11-
readonly string _consumerTag;
7+
private readonly string _consumerTag;
8+
9+
public override string Context => "HandleBasicCancel";
1210

1311
public BasicCancel(IBasicConsumer consumer, string consumerTag) : base(consumer)
1412
{
1513
_consumerTag = consumerTag;
1614
}
1715

18-
protected override async Task Execute(IModel model, IAsyncBasicConsumer consumer)
16+
protected override Task Execute(IAsyncBasicConsumer consumer)
1917
{
20-
try
21-
{
22-
await consumer.HandleBasicCancel(_consumerTag).ConfigureAwait(false);
23-
}
24-
catch (Exception e)
25-
{
26-
if (!(model is ModelBase modelBase))
27-
{
28-
return;
29-
}
30-
31-
var details = new Dictionary<string, object>
32-
{
33-
{"consumer", consumer},
34-
{"context", "HandleBasicCancel"}
35-
};
36-
modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
37-
}
18+
return consumer.HandleBasicCancel(_consumerTag);
3819
}
3920
}
4021
}

‎projects/RabbitMQ.Client/client/impl/BasicCancelOk.cs‎

Lines changed: 7 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -1,40 +1,21 @@
1-
using System;
2-
using System.Collections.Generic;
3-
using System.Threading.Tasks;
4-
5-
using RabbitMQ.Client.Events;
1+
using System.Threading.Tasks;
62

73
namespace RabbitMQ.Client.Impl
84
{
9-
sealed class BasicCancelOk : Work
5+
internal sealed class BasicCancelOk : Work
106
{
11-
readonly string _consumerTag;
7+
private readonly string _consumerTag;
8+
9+
public override string Context => "HandleBasicCancelOk";
1210

1311
public BasicCancelOk(IBasicConsumer consumer, string consumerTag) : base(consumer)
1412
{
1513
_consumerTag = consumerTag;
1614
}
1715

18-
protected override async Task Execute(IModel model, IAsyncBasicConsumer consumer)
16+
protected override Task Execute(IAsyncBasicConsumer consumer)
1917
{
20-
try
21-
{
22-
await consumer.HandleBasicCancelOk(_consumerTag).ConfigureAwait(false);
23-
}
24-
catch (Exception e)
25-
{
26-
if (!(model is ModelBase modelBase))
27-
{
28-
return;
29-
}
30-
31-
var details = new Dictionary<string, object>()
32-
{
33-
{"consumer", consumer},
34-
{"context", "HandleBasicCancelOk"}
35-
};
36-
modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
37-
}
18+
return consumer.HandleBasicCancelOk(_consumerTag);
3819
}
3920
}
4021
}

‎projects/RabbitMQ.Client/client/impl/BasicConsumeOk.cs‎

Lines changed: 7 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -1,40 +1,21 @@
1-
using System;
2-
using System.Collections.Generic;
3-
using System.Threading.Tasks;
4-
5-
using RabbitMQ.Client.Events;
1+
using System.Threading.Tasks;
62

73
namespace RabbitMQ.Client.Impl
84
{
9-
sealed class BasicConsumeOk : Work
5+
internal sealed class BasicConsumeOk : Work
106
{
11-
readonly string _consumerTag;
7+
private readonly string _consumerTag;
8+
9+
public override string Context => "HandleBasicConsumeOk";
1210

1311
public BasicConsumeOk(IBasicConsumer consumer, string consumerTag) : base(consumer)
1412
{
1513
_consumerTag = consumerTag;
1614
}
1715

18-
protected override async Task Execute(IModel model, IAsyncBasicConsumer consumer)
16+
protected override Task Execute(IAsyncBasicConsumer consumer)
1917
{
20-
try
21-
{
22-
await consumer.HandleBasicConsumeOk(_consumerTag).ConfigureAwait(false);
23-
}
24-
catch (Exception e)
25-
{
26-
if (!(model is ModelBase modelBase))
27-
{
28-
return;
29-
}
30-
31-
var details = new Dictionary<string, object>()
32-
{
33-
{"consumer", consumer},
34-
{"context", "HandleBasicConsumeOk"}
35-
};
36-
modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
37-
}
18+
return consumer.HandleBasicConsumeOk(_consumerTag);
3819
}
3920
}
4021
}

‎projects/RabbitMQ.Client/client/impl/BasicDeliver.cs‎

Lines changed: 23 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -1,22 +1,21 @@
11
using System;
22
using System.Buffers;
3-
using System.Collections.Generic;
43
using System.Runtime.InteropServices;
54
using System.Threading.Tasks;
65

7-
using RabbitMQ.Client.Events;
8-
96
namespace RabbitMQ.Client.Impl
107
{
11-
sealed class BasicDeliver : Work
8+
internal sealed class BasicDeliver : Work
129
{
13-
readonly string _consumerTag;
14-
readonly ulong _deliveryTag;
15-
readonly bool _redelivered;
16-
readonly string _exchange;
17-
readonly string _routingKey;
18-
readonly IBasicProperties _basicProperties;
19-
readonly ReadOnlyMemory<byte> _body;
10+
private readonly string _consumerTag;
11+
private readonly ulong _deliveryTag;
12+
private readonly bool _redelivered;
13+
private readonly string _exchange;
14+
private readonly string _routingKey;
15+
private readonly IBasicProperties _basicProperties;
16+
private readonly ReadOnlyMemory<byte> _body;
17+
18+
public override string Context => "HandleBasicDeliver";
2019

2120
public BasicDeliver(IBasicConsumer consumer,
2221
string consumerTag,
@@ -36,38 +35,22 @@ public BasicDeliver(IBasicConsumer consumer,
3635
_body = body;
3736
}
3837

39-
protected override async Task Execute(IModel model, IAsyncBasicConsumer consumer)
38+
protected override Task Execute(IAsyncBasicConsumer consumer)
4039
{
41-
try
42-
{
43-
await consumer.HandleBasicDeliver(_consumerTag,
44-
_deliveryTag,
45-
_redelivered,
46-
_exchange,
47-
_routingKey,
48-
_basicProperties,
49-
_body).ConfigureAwait(false);
50-
}
51-
catch (Exception e)
52-
{
53-
if (!(model is ModelBase modelBase))
54-
{
55-
return;
56-
}
40+
return consumer.HandleBasicDeliver(_consumerTag,
41+
_deliveryTag,
42+
_redelivered,
43+
_exchange,
44+
_routingKey,
45+
_basicProperties,
46+
_body);
47+
}
5748

58-
var details = new Dictionary<string, object>()
59-
{
60-
{"consumer", consumer},
61-
{"context", "HandleBasicDeliver"}
62-
};
63-
modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
64-
}
65-
finally
49+
public override void PostExecute()
50+
{
51+
if (MemoryMarshal.TryGetArray(_body, out ArraySegment<byte> segment))
6652
{
67-
if (MemoryMarshal.TryGetArray(_body, out ArraySegment<byte> segment))
68-
{
69-
ArrayPool<byte>.Shared.Return(segment.Array);
70-
}
53+
ArrayPool<byte>.Shared.Return(segment.Array);
7154
}
7255
}
7356
}

‎projects/RabbitMQ.Client/client/impl/ModelShutdown.cs‎

Lines changed: 9 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -1,40 +1,23 @@
1-
using System;
2-
using System.Collections.Generic;
31
using System.Threading.Tasks;
42

5-
using RabbitMQ.Client.Events;
6-
73
namespace RabbitMQ.Client.Impl
84
{
9-
sealed class ModelShutdown : Work
5+
internal sealed class ModelShutdown : Work
106
{
11-
readonly ShutdownEventArgs _reason;
7+
private readonly ShutdownEventArgs _reason;
8+
private readonly IModel _model;
9+
10+
public override string Context => "HandleModelShutdown";
1211

13-
public ModelShutdown(IBasicConsumer consumer, ShutdownEventArgs reason) : base(consumer)
12+
public ModelShutdown(IBasicConsumer consumer, ShutdownEventArgs reason, IModel model) : base(consumer)
1413
{
1514
_reason = reason;
15+
_model = model;
1616
}
1717

18-
protected override async Task Execute(IModel model, IAsyncBasicConsumer consumer)
18+
protected override Task Execute(IAsyncBasicConsumer consumer)
1919
{
20-
try
21-
{
22-
await consumer.HandleModelShutdown(model, _reason).ConfigureAwait(false);
23-
}
24-
catch (Exception e)
25-
{
26-
if (!(model is ModelBase modelBase))
27-
{
28-
return;
29-
}
30-
31-
var details = new Dictionary<string, object>()
32-
{
33-
{ "consumer", consumer },
34-
{ "context", "HandleModelShutdown" }
35-
};
36-
modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
37-
}
20+
return consumer.HandleModelShutdown(_model, _reason);
3821
}
3922
}
4023
}

0 commit comments

Comments
 (0)