<commit_before>﻿using System;
using System.IO;
using System.Net.Http;
using System.Net.Http.Headers;
using System.Threading.Tasks;

namespace OneCog.Io.Spark
{
    public interface IRestClient
    {
        Task<Stream> Get(Uri uri);

        Task<Stream> Post(Uri uri, HttpContent content);
    }

    public class RestClient : IRestClient
    {
        private readonly HttpClient _client;

        public RestClient(string accessToken)
        {
            _client = new HttpClient();
            _client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", accessToken);
        }

        public Task<Stream> Get(Uri uri)
        {
            return _client.GetStreamAsync(uri);
        }

        public async Task<Stream> Post(Uri uri, HttpContent content)
        {
            HttpResponseMessage response = await _client.PostAsync(uri, content);
            
            response.EnsureSuccessStatusCode();

            return await response.Content.ReadAsStreamAsync();
        }
    }
}<commit_msg>Now schedules requests to rest endpoint so that we don't make multiple requests simultaneously<commit_after>﻿using System;
using System.IO;
using System.Net.Http;
using System.Net.Http.Headers;
using System.Reactive.Concurrency;
using System.Reactive.Linq;
using System.Reactive.Threading.Tasks;
using System.Threading.Tasks;

namespace OneCog.Io.Spark
{
    public interface IRestClient
    {
        Task<Stream> Get(Uri uri);

        Task<Stream> Post(Uri uri, HttpContent content);
    }

    public class RestClient : IRestClient
    {
        private readonly HttpClient _client;
        private readonly IScheduler _requestScheduler;
        private readonly IScheduler _responseScheduler;

        public RestClient(string accessToken, IScheduler requestScheduler = null, IScheduler responseScheduler = null)
        {
            _client = new HttpClient();
            _client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", accessToken);

            _requestScheduler = requestScheduler ?? new EventLoopScheduler();
            _responseScheduler = responseScheduler ?? TaskPoolScheduler.Default;
        }

        public Task<Stream> Get(Uri uri)
        {
            return Observable
                .StartAsync(() => _client.GetStreamAsync(uri))
                .SubscribeOn(_requestScheduler)
                .ObserveOn(_responseScheduler)
                .ToTask();
        }

        public Task<Stream> Post(Uri uri, HttpContent content)
        {
            return Observable
                .StartAsync(
                    async () =>
                    {
                        HttpResponseMessage response = await _client.PostAsync(uri, content);

                        response.EnsureSuccessStatusCode();

                        return await response.Content.ReadAsStreamAsync();
                    }
                )
                .SubscribeOn(_requestScheduler)
                .ObserveOn(_responseScheduler)
                .ToTask();
        }
    }
}
{'fraction_non_alphanumeric': 0.07742567788304476, 'fraction_numerical': 0.0, 'mean_word_length': 2.1405128205128205, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5196095', 'n_tokens_mistral': 768, 'n_tokens_neox': 745, 'n_words': 177}