<commit_before>﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.IO;
using LitJson;

namespace Kayak.Framework
{
    public static partial class Extensions
    {
        public static IObservable<Unit> WriteJsonResponse(this object o, IKayakContext context, TypedJsonMapper mapper)
        {
            var buffer = new MemoryStream();
            var writer = new StreamWriter(buffer);

            bool minified = context.GetJsonOutputMinified();

            var response = context.Response;

            response.Headers["Content-Type"] = minified ? "application/json; charset=utf-8" : "text/plain; charset=utf-8";

            mapper.Write(o, new JsonWriter(writer) { Validate = false, PrettyPrint = !minified });
            writer.Flush();

            var bufferBytes = buffer.ToArray();

            context.Response.Headers["Content-Length"] = bufferBytes.Length.ToString();
            return context.Response.Body.WriteAsync(bufferBytes, 0, bufferBytes.Length);
        }
    }

    public class JsonHandler : IInvocationResultHandler
    {
        TypedJsonMapper mapper;

        public JsonHandler(TypedJsonMapper mapper)
        {
            this.mapper = mapper;
        }

        public IObservable<Unit> HandleResult(IKayakContext context, InvocationInfo info, object result)
        {
            return result.WriteJsonResponse(context, mapper);
        }
    }

    public class JsonBinder : IInvocationArgumentBinder
    {
        TypedJsonMapper mapper;

        public JsonBinder(TypedJsonMapper mapper)
        {
            this.mapper = mapper;
        }

        public void BindArgumentsFromHeaders(IKayakContext context, InvocationInfo info)
        {
            return;
        }

        public IObservable<Unit> BindArgumentsFromBody(IKayakContext context, InvocationInfo info)
        {
            if (!RequestBodyAttribute.IsDefinedOnParameters(info.Method))
                return null;

            return BindArgumentsFromBodyInternal(context, info).AsCoroutine<Unit>();
        }

        IEnumerable<object> BindArgumentsFromBodyInternal(IKayakContext context, InvocationInfo info)
        {
            var parameters = info.Method.GetParameters().Where(p => RequestBodyAttribute.IsDefinedOn(p));

            if (parameters.Count() == 0 || context.Request.Body == null)
                yield break;

            MemoryStream buffer = new MemoryStream();
            yield return BufferRequestBody(context, buffer).AsCoroutine<Unit>();

            var reader = new JsonReader(new StreamReader(buffer));

            if (parameters.Count() > 1)
                reader.Read(); // read array start

            foreach (var param in parameters)
                info.Arguments[param.Position] = mapper.Read(param.ParameterType, reader);

            if (parameters.Count() > 1)
                reader.Read(); // read array end
        }

        IEnumerable<object> BufferRequestBody(IKayakContext context, MemoryStream stream)
        {
            int bytesRead = 0;
            byte[] buffer = new byte[1024];

            while (true)
            {
                yield return context.Request.Body.ReadAsync(buffer, 0, buffer.Length).Do(n => bytesRead = n);

                if (bytesRead == 0)
                    break;

                stream.Write(buffer, 0, bytesRead);
            }

            buffer = null;
            stream.Position = 0;
        }
    }

    public static partial class Extensions
    {
        static readonly string MinifiedOutputKey = "JsonSerializerPrettyPrint";

        public static void SetJsonOutputMinified(this IKayakContext context, bool value)
        {
            context.Items[MinifiedOutputKey] = value;
        }

        public static bool GetJsonOutputMinified(this IKayakContext context)
        {
            return
                context.Items.ContainsKey(MinifiedOutputKey) &&
                (bool)Convert.ChangeType(context.Items[MinifiedOutputKey], typeof(bool));
        }

        public static void AddJsonSupport(this KayakInvocationBehavior behavior)
        {
            var mapper = new TypedJsonMapper();
            mapper.AddDefaultOutputConversions();
            mapper.AddDefaultInputConversions();

            behavior.AddJsonSupport(mapper);
        }
        public static void AddJsonSupport(this KayakInvocationBehavior behavior, TypedJsonMapper mapper)
        {
            behavior.Binders.Add(new JsonBinder(mapper));
            behavior.ResultHandlers.Add(new JsonHandler(mapper));
        }
    }
}
<commit_msg>Fixed JsonSupport buffering of request body.<commit_after>﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.IO;
using LitJson;

namespace Kayak.Framework
{
    public static partial class Extensions
    {
        public static IObservable<Unit> WriteJsonResponse(this object o, IKayakContext context, TypedJsonMapper mapper)
        {
            var buffer = new MemoryStream();
            var writer = new StreamWriter(buffer);

            bool minified = context.GetJsonOutputMinified();

            var response = context.Response;

            response.Headers["Content-Type"] = minified ? "application/json; charset=utf-8" : "text/plain; charset=utf-8";

            mapper.Write(o, new JsonWriter(writer) { Validate = false, PrettyPrint = !minified });
            writer.Flush();

            var bufferBytes = buffer.ToArray();

            context.Response.Headers["Content-Length"] = bufferBytes.Length.ToString();
            return context.Response.Body.WriteAsync(bufferBytes, 0, bufferBytes.Length);
        }
    }

    public class JsonHandler : IInvocationResultHandler
    {
        TypedJsonMapper mapper;

        public JsonHandler(TypedJsonMapper mapper)
        {
            this.mapper = mapper;
        }

        public IObservable<Unit> HandleResult(IKayakContext context, InvocationInfo info, object result)
        {
            return result.WriteJsonResponse(context, mapper);
        }
    }

    public class JsonBinder : IInvocationArgumentBinder
    {
        TypedJsonMapper mapper;

        public JsonBinder(TypedJsonMapper mapper)
        {
            this.mapper = mapper;
        }

        public void BindArgumentsFromHeaders(IKayakContext context, InvocationInfo info)
        {
            return;
        }

        public IObservable<Unit> BindArgumentsFromBody(IKayakContext context, InvocationInfo info)
        {
            if (!RequestBodyAttribute.IsDefinedOnParameters(info.Method))
                return null;

            return BindArgumentsFromBodyInternal(context, info).AsCoroutine<Unit>();
        }

        IEnumerable<object> BindArgumentsFromBodyInternal(IKayakContext context, InvocationInfo info)
        {
            var parameters = info.Method.GetParameters().Where(p => RequestBodyAttribute.IsDefinedOn(p));

            if (parameters.Count() == 0 || context.Request.Body == null)
                yield break;

            MemoryStream buffer = new MemoryStream();
            yield return BufferRequestBody(context, buffer).AsCoroutine<Unit>();

            var reader = new JsonReader(new StreamReader(buffer));

            if (parameters.Count() > 1)
                reader.Read(); // read array start

            foreach (var param in parameters)
                info.Arguments[param.Position] = mapper.Read(param.ParameterType, reader);

            if (parameters.Count() > 1)
                reader.Read(); // read array end
        }

        IEnumerable<object> BufferRequestBody(IKayakContext context, MemoryStream stream)
        {
            int bytesRead = 0;
            byte[] buffer = new byte[1024];

            int contentLength = context.Request.Headers.GetContentLength();

            while (true)
            {
                yield return context.Request.Body.ReadAsync(buffer, 0, buffer.Length).Do(n => bytesRead = n);

                stream.Write(buffer, 0, bytesRead);

                if (stream.Length == contentLength)
                    break;
            }

            buffer = null;
            stream.Position = 0;
        }
    }

    public static partial class Extensions
    {
        static readonly string MinifiedOutputKey = "JsonSerializerPrettyPrint";

        public static void SetJsonOutputMinified(this IKayakContext context, bool value)
        {
            context.Items[MinifiedOutputKey] = value;
        }

        public static bool GetJsonOutputMinified(this IKayakContext context)
        {
            return
                context.Items.ContainsKey(MinifiedOutputKey) &&
                (bool)Convert.ChangeType(context.Items[MinifiedOutputKey], typeof(bool));
        }

        public static void AddJsonSupport(this KayakInvocationBehavior behavior)
        {
            var mapper = new TypedJsonMapper();
            mapper.AddDefaultOutputConversions();
            mapper.AddDefaultInputConversions();

            behavior.AddJsonSupport(mapper);
        }
        public static void AddJsonSupport(this KayakInvocationBehavior behavior, TypedJsonMapper mapper)
        {
            behavior.Binders.Add(new JsonBinder(mapper));
            behavior.ResultHandlers.Add(new JsonHandler(mapper));
        }
    }
}

{'fraction_non_alphanumeric': 0.08282975619920817, 'fraction_numerical': 0.0030214628047509897, 'mean_word_length': 1.9978138663335416, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 25, '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': '1529797', 'n_tokens_mistral': 2509, 'n_tokens_neox': 2318, 'n_words': 522}