Я хотел бы получить доступ к очередям служебной шины и темам из Workflows с некоторыми конкретными действиями.
Я не смог найти ничего подходящего для этого сценария (эта статья MSDN и эта статья Романа Kiss) являются ближайшими.
Я хотел бы создать пользовательскую активность, которая использует QueueClient для асинхронного получения брокерских сообщений с использованием метода BeginReceive, реализованного с использованием шаблона async/await (см. мой вопрос об этом).
Прежде всего, я хотел бы спросить, есть ли у меня какие-либо причины, почему я предпочитаю предлагаемый подход (адаптированный WCF) вместо моего желаемого (используя QueueClient).
Тогда, я был бы признателен за помощь в его разработке настойчивым образом.
Update:
Это то, что я пробовал до сих пор:
public class AsyncReceiveBrokeredMessage : AsyncCodeActivity<BrokeredMessage>
{
[RequiredArgument]
public InArgument<string> ConnectionString { get; set; }
[RequiredArgument]
public InArgument<string> Path { get; set; }
protected sealed override IAsyncResult BeginExecute(AsyncCodeActivityContext context, AsyncCallback callback, object state)
{
var connectionString = this.ConnectionString.Get(context);
var path = this.Path.Get(context);
var queueClient = QueueClient.CreateFromConnectionString(connectionString, path);
var cts = new CancellationTokenSource();
context.UserState = new ReceiveState
{
CancellationTokenSource = cts,
QueueClient = queueClient
};
var task = ExecuteAsync(context, cts.Token);
var tcs = new TaskCompletionSource<BrokeredMessage>(state);
task.ContinueWith(
t =>
{
if (t.IsFaulted)
{
tcs.TrySetException(t.Exception.InnerExceptions);
}
else if (t.IsCanceled)
{
tcs.TrySetCanceled();
}
else
{
tcs.TrySetResult(t.Result);
}
if (callback != null)
{
callback(tcs.Task);
}
});
return tcs.Task;
}
protected sealed override BrokeredMessage EndExecute(AsyncCodeActivityContext context, IAsyncResult result)
{
var task = (Task<BrokeredMessage>)result;
try
{
return task.Result;
}
catch (OperationCanceledException)
{
if (context.IsCancellationRequested)
{
context.MarkCanceled();
}
else
{
throw;
}
return null; // or throw?
}
catch (AggregateException exception)
{
if (exception.InnerException is OperationCanceledException)
{
if (context.IsCancellationRequested)
{
context.MarkCanceled();
}
else
{
throw;
}
return null; // or throw?
}
ExceptionDispatchInfo.Capture(exception.InnerException).Throw();
throw;
}
}
protected override void Cancel(AsyncCodeActivityContext context)
{
var state = (ReceiveState)context.UserState;
state.CancellationTokenSource.Cancel();
}
private async Task<BrokeredMessage> ExecuteAsync(
AsyncCodeActivityContext context, CancellationToken cancellationToken)
{
var receiveState = context.UserState as ReceiveState;
var receiveTask = Task<BrokeredMessage>.Factory.FromAsync(
receiveState.QueueClient.BeginReceive, receiveState.QueueClient.EndReceive, null);
var completionTask = receiveTask.ContinueWith(
t =>
{
BrokeredMessage result;
if (t.IsCanceled)
{
context.MarkCanceled();
result = null;
}
else if (t.IsFaulted)
{
result = null;
}
else
{
t.Result.Complete();
result = t.Result;
}
receiveState.QueueClient.Close();
return result;
},
cancellationToken);
return await completionTask;
}
private class ReceiveState
{
public CancellationTokenSource CancellationTokenSource { get; set; }
public QueueClient QueueClient { get; set; }
}
}
И протестировал этот путь (используя локальную служебную шину Windows Server):
var connectionString = new Variable<string>
{
Default = connectionStringValue
};
var path = new Variable<string>
{
Default = pathValue
};
var test = new While
{
Body =
new Pick
{
Branches =
{
new PickBranch
{
Trigger =
new AsyncReceiveBrokeredMessage
{
ConnectionString = new InArgument<string>(connectionString),
Path = new InArgument<string>(path)
},
Action =
new WriteLine
{
Text =
"Received message"
}
},
new PickBranch
{
Trigger =
new Delay
{
Duration = TimeSpan.FromSeconds(10)
},
Action =
new WriteLine
{
Text =
"Timeout!"
}
}
}
},
Condition = true,
Variables = { connectionString, path }
};
WorkflowInvoker.Invoke(test);
Я получаю сообщения, как ожидалось, если я их постоянно отправляю. Проблемы возникают после первого таймаута, потому что тогда я больше не получаю никакого сообщения. Любое разъяснение оценивается.
