1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
|
//------------------------------------------------------------
// Copyright (c) Microsoft Corporation. All rights reserved.
//------------------------------------------------------------
namespace System.ServiceModel.Channels
{
using System.Collections.Generic;
using System.Runtime;
using System.ServiceModel.Diagnostics.Application;
sealed class Msmq3PoisonHandler : IPoisonHandlingStrategy
{
const int maxTrackedMessages = 256;
MsmqReceiveHelper receiver;
SortedList<long, int> trackedMessages;
object thisLock = new object();
internal Msmq3PoisonHandler(MsmqReceiveHelper receiver)
{
this.receiver = receiver;
this.trackedMessages = new SortedList<long, int>(maxTrackedMessages);
}
public bool CheckAndHandlePoisonMessage(MsmqMessageProperty messageProperty)
{
long lookupId = messageProperty.LookupId;
int seen;
lock (thisLock)
{
seen = this.UpdateSeenCount(lookupId);
if (seen > (receiver.MsmqReceiveParameters.ReceiveRetryCount + 1) && receiver.MsmqReceiveParameters.ReceiveRetryCount != Int32.MaxValue)
{
if (TD.ReceiveRetryCountReachedIsEnabled())
{
TD.ReceiveRetryCountReached(messageProperty.MessageId);
}
FinalDisposition(messageProperty);
this.trackedMessages.Remove(lookupId);
return true;
}
}
messageProperty.AbortCount = seen - 1;
return false;
}
public void FinalDisposition(MsmqMessageProperty messageProperty)
{
switch (receiver.MsmqReceiveParameters.ReceiveErrorHandling)
{
case ReceiveErrorHandling.Drop:
//Unlocking message here since the message is locked under internal transaction and
//cannot be unlocked by aborting ambient transaction
MsmqDefaultLockingQueue queue = this.receiver.Queue as MsmqDefaultLockingQueue;
if ((queue != null) && this.receiver.Transactional)
queue.UnlockMessage(messageProperty.LookupId, TimeSpan.Zero);
this.receiver.DropOrRejectReceivedMessage(messageProperty, false);
break;
case ReceiveErrorHandling.Fault:
MsmqReceiveHelper.TryAbortTransactionCurrent();
if (null != this.receiver.ChannelListener)
this.receiver.ChannelListener.FaultListener();
if (null != this.receiver.Channel)
this.receiver.Channel.FaultChannel();
break;
default:
Fx.Assert("System.ServiceModel.Channels.Msmq3PoisonHandler.FinalDisposition(): (unexpected ReceiveErrorHandling)");
break;
}
}
int UpdateSeenCount(long lookupId)
{
int value;
if (this.trackedMessages.TryGetValue(lookupId, out value))
{
++value;
this.trackedMessages[lookupId] = value;
return value;
}
else
{
if (maxTrackedMessages == this.trackedMessages.Count)
{
this.trackedMessages.RemoveAt(0);
}
this.trackedMessages.Add(lookupId, 1);
return 1;
}
}
public void Open()
{ }
public void Dispose()
{ }
}
}
|
