C# 8.0 - Async Streams
نویسنده: وحید نصیری
تاریخ: ۱۳۹۸/۰۳/۱۴ ۱۳:۱۰
آدرس: www.dntips.ir
| مطالب | ۳۶۹۴ |
| نویسندگان | ۲۷۶ |
| گروههای مطالب | ۱۰۲۴ |
| نقشههای راه | ۱۱۹ |
| دورهها | ۱۴ |
| اشتراکها | ۱۷۹۱۴ |
public struct Statement
{
public int Id { get; }
public string Description { get; }
public Statement(int id, string description) => (Id, Description) = (id, description);
public override string ToString() => Description;
}
public static async Task<IEnumerable<Statement>> GetStatements(bool error)
{
if (error)
{
throw new Exception("Oops, we messed up 😬");
}
await Task.Delay(1000); //Simulate waiting for data to come through.
yield return new Statement(1, "C# is cool!");
yield return new Statement(2, "C# orginally named COOL.");
yield return new Statement(3, "More examples...");
} The body of 'AsyncStreams.GetStatements(bool)' cannot be an iterator block because 'Task<IEnumerable<AsyncStreams.Statement>>' is not an iterator interface type (CS1624)
static async IAsyncEnumerable<Statement> GetStatements(bool error)
static async IAsyncEnumerable<Statement> GetStatementsAsync(bool error)
{
await foreach (var statement in GetStatements(error))
{
await Task.Delay(1000);
yield return statement;
}
} using System.Threading;
namespace System.Collections.Generic
{
public interface IAsyncEnumerable<out T>
{
IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default);
}
public interface IAsyncEnumerator<out T> : IAsyncDisposable
{
T Current { get; }
ValueTask<bool> MoveNextAsync();
}
}
namespace System
{
public interface IAsyncDisposable
{
ValueTask DisposeAsync();
}
} public class AwaitUsingTest : IAsyncDisposable
{
public async ValueTask DisposeAsync()
{
await Task.CompletedTask;
}
public void Dummy() { }
} async Task FooBar()
{
await using var test = new AwaitUsingTest();
test.Dummy();
} [HttpGet]
public IAsyncEnumerable<Product> Get()
=> productsRepository.GetAllProducts(); ‘AsyncEnumerableReader’ reached the configured maximum size of the buffer when enumerating a value of type ‘<type>’. This limit is in place to prevent infinite streams of ‘IAsyncEnumerable<>’ from continuing indefinitely. If this is not a programming mistake, consider ways to reduce the collection size, or consider manually converting ‘<type>’ into a list rather than increasing the limit.
The type ‘IAsyncEnumerable’ exists in both ‘System.Interactive.Async, Version=3.2.0.0, Culture=neutral, PublicKeyToken=94bc3704cddfc263’ and ‘System.Runtime, Version=4.2.2.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a’
<Target Name="AddAssemblyAliasToReactiveAsync" AfterTargets="ResolveAssemblyReferences">
<ItemGroup>
<ReferencePath Condition=" '%(FileName)' == 'System.Interactive.Async' ">
<Aliases>reactive</Aliases>
</ReferencePath>
</ItemGroup>
</Target> [HttpGet]
public IAsyncEnumerable<Product> Get()
=> productsRepository.GetAllProducts(); static async IAsyncEnumerable<int> RangeAsync(int start, int count)
{
for (int i = 0; i < count; i++)
{
await Task.Delay(i);
yield return start + i;
}
} [ApiController]
[Route("[controller]")]
public class CustomerController : ControllerBase
{
private readonly IDictionary<int, Customer> _customers;
private void FillCustomerFromMemory(int countOfCustomer)
{
for (int CustomerId = 1; CustomerId <= countOfCustomer; CustomerId++)
{
_customers.Add(key: CustomerId, new Customer($"name_{CustomerId}", CustomerId));
}
}
public CustomerController()
{
_customers = new Dictionary<int, Customer>();
FillCustomerFromMemory(countOfCustomer : 100);
}
[HttpGet]
public async Task<IEnumerable<Customer>> Get()
{
var output = new List<Customer>();
while (_customers.Any(_ => _.Key % 10 == 0))
{
var customer = _customers.First(_ => _.Key % 10 == 0);
output.Add(new Customer(customer.Value.Name, customer.Key));
await Task.Delay(500);
_customers.Remove(customer);
}
return output;
}
public class Customer
{
public int Id { get; private set; }
public string Name { get; private set; }
public Customer(string name, int id)
{
Name = name;
Id = id;
}
}
} [HttpGet]
public async IAsyncEnumerable<Customer> Get()
{
while (_customers.Any(_ => _.Key % 10 == 0))
{
var customer = _customers.First(_ => _.Key % 10 == 0);
yield return new Customer(customer.Value.Name, customer.Key);
_customers.Remove(customer);
await Task.Delay(500);
}
} const int TARGET = 80;
var _httpClient = new HttpClient();
using (var response = await _httpClient.GetAsync(
"https://localhost:7284/customer",
HttpCompletionOption.ResponseHeadersRead))
{
var stream = await response.Content.ReadAsStreamAsync();
var _jsonSerializerSettings = new JsonSerializerSettings();
var _serializer = Newtonsoft.Json.JsonSerializer.Create(_jsonSerializerSettings);
using TextReader textReader = new StreamReader(stream);
using JsonReader jsonReader = new JsonTextReader(textReader);
await using (stream.ConfigureAwait(false))
{
await jsonReader.ReadAsync().ConfigureAwait(false);
while (await jsonReader.ReadAsync().ConfigureAwait(false) &&
jsonReader.TokenType != JsonToken.EndArray)
{
Customer customer = _serializer!.Deserialize<Customer>(jsonReader);
if (customer.Id == TARGET)
{
Console.WriteLine(customer.Id + " : " + customer.Name);
break;
}
}
}
} [HttpGet]
public async IAsyncEnumerable<Customer> Get(CancellationToken cancellationToken)
{
while (!cancellationToken.IsCancellationRequested && _customers.Any(_ => _.Key % 10 == 0))
{
var customer = _customers.First(_ => _.Key % 10 == 0);
yield return new Customer(customer.Value.Name, customer.Key);
_customers.Remove(customer);
await Task.Delay(500,cancellationToken);
}
} const int TARGET = 80;
var _httpClient = new HttpClient();
var _cancelationTokenSource = new CancellationTokenSource();
using (var response = await _httpClient.GetAsync(
"https://localhost:7284/customer",
HttpCompletionOption.ResponseHeadersRead,
_cancelationTokenSource.Token))
{
var stream = await response.Content.ReadAsStreamAsync(_cancelationTokenSource.Token);
var _jsonSerializerSettings = new JsonSerializerSettings();
var _serializer = Newtonsoft.Json.JsonSerializer.Create(_jsonSerializerSettings);
using TextReader textReader = new StreamReader(stream);
using JsonReader jsonReader = new JsonTextReader(textReader);
await using (stream.ConfigureAwait(false))
{
await jsonReader.ReadAsync(_cancelationTokenSource.Token).ConfigureAwait(false);
while (await jsonReader.ReadAsync(_cancelationTokenSource.Token).ConfigureAwait(false) &&
jsonReader.TokenType != JsonToken.EndArray)
{
Customer customer = _serializer!.Deserialize<Customer>(jsonReader);
if (customer.Id == TARGET)
{
Console.WriteLine(customer.Id + " : " + customer.Name);
_cancelationTokenSource.Cancel();
break;
}
}
}
}