I want to use DataflowEx to encapsulate this test. I am not sure that my design is correct as I have tried a lot to do it my self but I failed to make the Dataflow to complete as in my test and is not clear how I should post the server, IP combination from the outside.
class HttpClientHandlerFactory : Dataflow<string, HttpClientHandler> {
private readonly TransformBlock<string, WebProxy> _webproxyDataflow;
private readonly TransformBlock<WebProxy, HttpClientHandler> _httpClientHandlerDataflow;
public HttpClientHandlerFactory(IHttpClientHandlerFactoryData handlerFactoryData) : this(DataflowOptions.Default, handlerFactoryData) {
}
public HttpClientHandlerFactory(DataflowOptions dataflowOptions, IHttpClientHandlerFactoryData handlerFactoryData) : base(dataflowOptions){
var dataflowBlockOptions = dataflowOptions.ToExecutionBlockOption();
_webproxyDataflow = new TransformBlock<string, WebProxy>(async s => {
new WebProxy(s);
}, dataflowBlockOptions);
_httpClientHandlerDataflow = new TransformBlock<WebProxy, HttpClientHandler>(proxy => new HttpClientHandler(){Proxy = proxy}, dataflowBlockOptions);
var dataflowLinkOptions = new DataflowLinkOptions() { PropagateCompletion = true };
_webproxyDataflow.LinkTo(_httpClientHandlerDataflow, dataflowLinkOptions, proxy => proxy != null);
var nullTarget = DataflowBlock.NullTarget<WebProxy>();
_webproxyDataflow.LinkTo(nullTarget, dataflowLinkOptions, proxy => proxy == null);
RegisterChild(_webproxyDataflow);
RegisterChild(_httpClientHandlerDataflow);
}
public override ITargetBlock<string> InputBlock => _webproxyDataflow;
public override ISourceBlock<HttpClientHandler> OutputBlock => _httpClientHandlerDataflow;
}
class FactoryData: IHttpClientHandlerFactoryData {
}
[Fact]
public async void MethodName(){
var httpClientHandlers = new BufferBlock<HttpClientHandler>();
var httpClientHandlerFactory = new HttpClientHandlerFactory(new FactoryData());
httpClientHandlerFactory.LinkTo(httpClientHandlers.ToDataflow());
var baseAddresses = new BufferBlock<Uri>();
var transformBlock = new TransformBlock<Tuple<HttpClientHandler, Uri>,HttpClient>(
data => {
WriteLine($"{data.Item2.ToString()}-{Environment.CurrentManagedThreadId}-{((WebProxy) data.Item1.Proxy).Address}");
return new HttpClient(data.Item1){BaseAddress = data.Item2};
});
var joinBlock = new JoinBlock<HttpClientHandler,Uri>();
httpClientHandlers.LinkTo(joinBlock.Target1);
baseAddresses.LinkTo(joinBlock.Target2);
joinBlock.LinkTo(transformBlock,new DataflowLinkOptions(){PropagateCompletion = true});
baseAddresses.Post(new Uri("http://www.serverA.com"));
baseAddresses.Post(new Uri("http://www.ServerB.com"));
httpClientHandlerFactory.Post("127.0.0.1");
httpClientHandlerFactory.Post("127.0.0.2");
joinBlock.Complete();
await transformBlock.Completion;
}
Update
Reading carefully JSteward comment about the missing NullTarget block I realized that my raw block method was incorrect. So here is the correct version.
[Fact]
public async Task MethodName() {
var dataflowLinkOptions = new DataflowLinkOptions() { PropagateCompletion = true };
var httpClientHandlerFactory = new HttpClientHandlerFactory(new HttpClientHandlerFactoryData());
var baseAddresses = new BufferBlock<Uri>();
var transformBlock = new TransformBlock<Tuple<HttpClientHandler, Uri>, HttpClient>(
data => {
WriteLine($"{data.Item2.ToString()}-{Environment.CurrentManagedThreadId}-{((WebProxy)data.Item1.Proxy).Address}");
var httpClient = new HttpClient(data.Item1) { BaseAddress = data.Item2 };
return httpClient;
});
var joinBlock = new JoinBlock<HttpClientHandler, Uri>();
httpClientHandlerFactory.OutputBlock.LinkTo(joinBlock.Target1,dataflowLinkOptions);
baseAddresses.LinkTo(joinBlock.Target2,dataflowLinkOptions);
joinBlock.LinkTo(transformBlock, dataflowLinkOptions);
baseAddresses.Post(new Uri("http://www.serverA.com"));
baseAddresses.Post(new Uri("http://www.ServerB.com"));
httpClientHandlerFactory.Post("127.0.0.1");
httpClientHandlerFactory.Post("127.0.0.2");
httpClientHandlerFactory.Complete();
baseAddresses.Complete();
transformBlock.LinkTo(DataflowBlock.NullTarget<HttpClient>(), dataflowLinkOptions);
await transformBlock.Completion;
}
It completes with this output.
02:51:41.3365|http://www.servera.com/-11-http://127.0.0.1/ 02:51:41.3365|http://www.serverb.com/-11-http://127.0.0.2/
Now to migrate to DataflowEx I declared the next class
class HttpClientFactory:Dataflow<string,HttpClient>{
private readonly HttpClientHandlerFactory _httpClientHandlerFactory;
private readonly TransformBlock<Tuple<HttpClientHandler, Uri>, HttpClient> _transformBlock;
private JoinBlock<HttpClientHandler, Uri> _joinBlock;
private BufferBlock<Uri> _baseAddresses;
public HttpClientFactory(IHttpClientFactoryData httpClientHandlerFactoryData) : this(DataflowOptions.Default, httpClientHandlerFactoryData) {
}
public HttpClientFactory(DataflowOptions dataflowOptions, IHttpClientFactoryData handlerFactoryData) : base(dataflowOptions) {
var dataflowLinkOptions = new DataflowLinkOptions() { PropagateCompletion = true };
_httpClientHandlerFactory = new HttpClientHandlerFactory(new HttpClientHandlerFactoryData());
_baseAddresses = new BufferBlock<Uri>();
_transformBlock = new TransformBlock<Tuple<HttpClientHandler, Uri>, HttpClient>(
data => {
handlerFactoryData.Logger.WriteLine($"{data.Item2.ToString()}-{Environment.CurrentManagedThreadId}-{((WebProxy)data.Item1.Proxy).Address}");
return new HttpClient(data.Item1) { BaseAddress = data.Item2 };
});
_joinBlock = new JoinBlock<HttpClientHandler, Uri>();
_httpClientHandlerFactory.OutputBlock.LinkTo(_joinBlock.Target1,dataflowLinkOptions);
_baseAddresses.LinkTo(_joinBlock.Target2,dataflowLinkOptions);
_joinBlock.LinkTo(_transformBlock, dataflowLinkOptions);
RegisterChild(_transformBlock);
RegisterChild(_httpClientHandlerFactory);
RegisterChild(_baseAddresses);
foreach (var baseAddress in handlerFactoryData.BaseAddresses){
_baseAddresses.Post(baseAddress);
}
}
public override ITargetBlock<string> InputBlock => _httpClientHandlerFactory.InputBlock;
public override ISourceBlock<HttpClient> OutputBlock => _transformBlock;
}
and wrote a test that it completes ok with the expected output
[Fact]
public async Task MethodName2(){
var clientFactoryData = new HttpClientFactoryData(new[]{new Uri("http://www.serverA.com")}){Logger = this};
var httpClientFactory = new HttpClientFactory(DataflowOptions.Verbose,clientFactoryData);
httpClientFactory.InputBlock.Post("127.0.0.1");
httpClientFactory.LinkLeftToNull();
httpClientFactory.Complete();
await httpClientFactory.OutputBlock.Completion;
}
02:55:44.8536|[HttpClientFactory1] has 0 todo items (in:0, out:0) at this moment. 02:55:45.0039|[HttpClientFactory1]->[HttpClientHandlerFactory1] completed 02:55:45.0039|http://www.servera.com/-11-http://127.0.0.1/
However in in the above test I used the raw Complete method and awaited on the OutputBlock and not on the DataFlow. The next test that uses the builtin Dataflow method does not complete.
[Fact]
public async Task MethodName2(){
var clientFactoryData = new HttpClientFactoryData(new[]{new Uri("http://www.serverA.com")}){Logger = this};
var httpClientFactory = new HttpClientFactory(DataflowOptions.Verbose,clientFactoryData);
httpClientFactory.InputBlock.Post("127.0.0.1");
httpClientFactory.LinkLeftToNull();
await httpClientFactory.SignalAndWaitForCompletionAsync();
}
Overriding the Complete method to complete the additional BufferBlock makes the test pass.
public override void Complete(){
base.Complete();
_baseAddresses.Complete();
}