ISessionClient.AcceptMessageSessionAsync 中的 operationTimeout 实际上是做什么的?
What does the operationTimeout in ISessionClient.AcceptMessageSessionAsync actually do?
Context:我有一些代码正在为特定会话创建消息会话,使用
ISessionClient.Task<IMessageSession> AcceptMessageSessionAsync(string sessionId, TimeSpan operationTimeout);
问题:AcceptMessageSessionAsync中的operationTimeout有什么作用?我尝试将其设置为一分钟,但一分钟后,什么也没有发生。这个超时是否只是设置了一个 属性 我需要检查自己? SessionLockLostException 不应该触发吗?
代码示例:
var session = await sessionClient.AcceptMessageSessionAsync(0, TimeSpan.FromMinutes(1));
var gotSession = true;
if (gotSession)
{
while (!session.IsClosedOrClosing)
{
try
{
Message message = await session.ReceiveAsync(TimeSpan.FromMinutes(2));
if (message != null)
{
await session.CompleteAsync(message.SystemProperties.LockToken);
}
else
{
await session.CloseAsync();
}
}
}
}
AcceptMessageSessionAsync 中的 OperationTimeout 是调用应等待获取下一个会话的时间量。
您可以在此处找到 AcceptMessageSessionAsync 方法的完整实现
/// <summary>
/// Gets a particular session object identified by <paramref name="sessionId"/> that can be used to receive messages for that sessionId.
/// </summary>
/// <param name="sessionId">The sessionId present in all its messages.</param>
/// <param name="operationTimeout">Amount of time for which the call should wait to fetch the next session.</param>
/// <remarks>All plugins registered on <see cref="SessionClient"/> will be applied to each <see cref="MessageSession"/> that is accepted.
/// Individual sessions can further register additional plugins.</remarks>
public async Task<IMessageSession> AcceptMessageSessionAsync(string sessionId, TimeSpan operationTimeout)
{
this.ThrowIfClosed();
MessagingEventSource.Log.AmqpSessionClientAcceptMessageSessionStart(
this.ClientId,
this.EntityPath,
this.ReceiveMode,
this.PrefetchCount,
sessionId);
bool isDiagnosticSourceEnabled = ServiceBusDiagnosticSource.IsEnabled();
Activity activity = isDiagnosticSourceEnabled ? this.diagnosticSource.AcceptMessageSessionStart(sessionId) : null;
Task acceptMessageSessionTask = null;
var session = new MessageSession(
this.EntityPath,
this.EntityType,
this.ReceiveMode,
this.ServiceBusConnection,
this.CbsTokenProvider,
this.RetryPolicy,
this.PrefetchCount,
sessionId,
true);
try
{
acceptMessageSessionTask = this.RetryPolicy.RunOperation(
() => session.GetSessionReceiverLinkAsync(operationTimeout),
operationTimeout);
await acceptMessageSessionTask.ConfigureAwait(false);
}
catch (Exception exception)
{
if (isDiagnosticSourceEnabled)
{
this.diagnosticSource.ReportException(exception);
}
MessagingEventSource.Log.AmqpSessionClientAcceptMessageSessionException(
this.ClientId,
this.EntityPath,
exception);
await session.CloseAsync().ConfigureAwait(false);
throw AmqpExceptionHelper.GetClientException(exception);
}
finally
{
this.diagnosticSource.AcceptMessageSessionStop(activity, session.SessionId, acceptMessageSessionTask?.Status);
}
MessagingEventSource.Log.AmqpSessionClientAcceptMessageSessionStop(
this.ClientId,
this.EntityPath,
session.SessionIdInternal);
session.UpdateClientId(ClientEntity.GenerateClientId(nameof(MessageSession), $"{this.EntityPath}_{session.SessionId}"));
// Register plugins on the message session.
foreach (var serviceBusPlugin in this.RegisteredPlugins)
{
session.RegisterPlugin(serviceBusPlugin);
}
return session;
}
您可以在下面找到完整的示例 link
希望对您有所帮助。
Context:我有一些代码正在为特定会话创建消息会话,使用
ISessionClient.Task<IMessageSession> AcceptMessageSessionAsync(string sessionId, TimeSpan operationTimeout);
问题:AcceptMessageSessionAsync中的operationTimeout有什么作用?我尝试将其设置为一分钟,但一分钟后,什么也没有发生。这个超时是否只是设置了一个 属性 我需要检查自己? SessionLockLostException 不应该触发吗?
代码示例:
var session = await sessionClient.AcceptMessageSessionAsync(0, TimeSpan.FromMinutes(1));
var gotSession = true;
if (gotSession)
{
while (!session.IsClosedOrClosing)
{
try
{
Message message = await session.ReceiveAsync(TimeSpan.FromMinutes(2));
if (message != null)
{
await session.CompleteAsync(message.SystemProperties.LockToken);
}
else
{
await session.CloseAsync();
}
}
}
}
AcceptMessageSessionAsync 中的 OperationTimeout 是调用应等待获取下一个会话的时间量。
您可以在此处找到 AcceptMessageSessionAsync 方法的完整实现
/// <summary>
/// Gets a particular session object identified by <paramref name="sessionId"/> that can be used to receive messages for that sessionId.
/// </summary>
/// <param name="sessionId">The sessionId present in all its messages.</param>
/// <param name="operationTimeout">Amount of time for which the call should wait to fetch the next session.</param>
/// <remarks>All plugins registered on <see cref="SessionClient"/> will be applied to each <see cref="MessageSession"/> that is accepted.
/// Individual sessions can further register additional plugins.</remarks>
public async Task<IMessageSession> AcceptMessageSessionAsync(string sessionId, TimeSpan operationTimeout)
{
this.ThrowIfClosed();
MessagingEventSource.Log.AmqpSessionClientAcceptMessageSessionStart(
this.ClientId,
this.EntityPath,
this.ReceiveMode,
this.PrefetchCount,
sessionId);
bool isDiagnosticSourceEnabled = ServiceBusDiagnosticSource.IsEnabled();
Activity activity = isDiagnosticSourceEnabled ? this.diagnosticSource.AcceptMessageSessionStart(sessionId) : null;
Task acceptMessageSessionTask = null;
var session = new MessageSession(
this.EntityPath,
this.EntityType,
this.ReceiveMode,
this.ServiceBusConnection,
this.CbsTokenProvider,
this.RetryPolicy,
this.PrefetchCount,
sessionId,
true);
try
{
acceptMessageSessionTask = this.RetryPolicy.RunOperation(
() => session.GetSessionReceiverLinkAsync(operationTimeout),
operationTimeout);
await acceptMessageSessionTask.ConfigureAwait(false);
}
catch (Exception exception)
{
if (isDiagnosticSourceEnabled)
{
this.diagnosticSource.ReportException(exception);
}
MessagingEventSource.Log.AmqpSessionClientAcceptMessageSessionException(
this.ClientId,
this.EntityPath,
exception);
await session.CloseAsync().ConfigureAwait(false);
throw AmqpExceptionHelper.GetClientException(exception);
}
finally
{
this.diagnosticSource.AcceptMessageSessionStop(activity, session.SessionId, acceptMessageSessionTask?.Status);
}
MessagingEventSource.Log.AmqpSessionClientAcceptMessageSessionStop(
this.ClientId,
this.EntityPath,
session.SessionIdInternal);
session.UpdateClientId(ClientEntity.GenerateClientId(nameof(MessageSession), $"{this.EntityPath}_{session.SessionId}"));
// Register plugins on the message session.
foreach (var serviceBusPlugin in this.RegisteredPlugins)
{
session.RegisterPlugin(serviceBusPlugin);
}
return session;
}
您可以在下面找到完整的示例 link
希望对您有所帮助。