Add sync support to retriable stream

[pakrym]

Also, adds validating stream that makes sure appropriate methods are being called on the response stream.

Fix bugs found using validating stream.

Breaking change:

Sync request factory is now required.

Before:

response.Value.Content = RetriableStream.Create(
    response.GetRawResponse(),
    async startOffset =>
        (await this.StartDownloadAsync(
            range,
            accessConditions,
            rangeGetContentHash,
            startOffset,
            cancellation)
            .ConfigureAwait(false))
        .GetRawResponse(),
    // TODO: For now we're using the default ResponseClassifier
    // on BlobConnectionOptions so we'll do the same here
    new ResponseClassifier(),
    Constants.MaxReliabilityRetries);

After:

response.Value.Content = RetriableStream.Create(
    response.GetRawResponse(),
        startOffset =>
        (this.StartDownloadAsync(
                range,
                accessConditions,
                rangeGetContentHash,
                startOffset,
                cancellation)
            .ConfigureAwait(false).GetAwaiter().GetResult())
        .GetRawResponse(),
    async startOffset =>
        (await this.StartDownloadAsync(
            range,
            accessConditions,
            rangeGetContentHash,
            startOffset,
            cancellation)
            .ConfigureAwait(false))
        .GetRawResponse(),
    // TODO: For now we're using the default ResponseClassifier
    // on BlobConnectionOptions so we'll do the same here
    new ResponseClassifier(),
    Constants.MaxReliabilityRetries);

[KrzysztofCwalina]

Why do we need these sync code paths? We are inside an async method.

[pakrym]

It's a pattern we use where we write a method that takes a bool async parameter and becomes fully sync or async depending on the value.

This is a shorter example from BearerTokenAuthenticationPolicy implementation:

        public override Task ProcessAsync(HttpPipelineMessage message, ReadOnlyMemory<HttpPipelinePolicy> pipeline)
        {
            return ProcessAsync(message, pipeline, true);
        }

        public override void Process(HttpPipelineMessage message, ReadOnlyMemory<HttpPipelinePolicy> pipeline)
        {
            ProcessAsync(message, pipeline, false).EnsureCompleted();
        }

        public async Task ProcessAsync(HttpPipelineMessage message, ReadOnlyMemory<HttpPipelinePolicy> pipeline, bool async)
        {
>>>>        string token = async ?
                    await _credential.GetTokenAsync(_scopes, message.CancellationToken).ConfigureAwait(false) :
                    _credential.GetToken(_scopes, message.CancellationToken);

            if (token != _currentToken)
            {
                // Avoid per request allocations
                _currentToken = token;
                _headerValue = "Bearer " + token;
            }

            message.Request.SetHeader(HttpHeader.Names.Authorization, _headerValue);

            if (async)
            {
>>>>            await ProcessNextAsync(message, pipeline);
            }
            else
            {
>>>>            ProcessNext(message, pipeline);
            }
        }