HttpClientTransport.cs

Go to the documentation of this file.

using System;
using System.IO;
using System.Net;
using System.Net.Http;
using System.Net.Http.Headers;
using System.Threading;
using System.Threading.Tasks;

namespace kinetica;

    internal sealed class HttpClientTransport : IHttpTransport, IDisposable
    {
        private readonly HttpClient _client;
        private readonly bool _ownsClient;

        public HttpClientTransport(
            TimeSpan timeout,
            TimeSpan? pooledConnectionLifetime = null,
            TimeSpan? pooledConnectionIdleTimeout = null)
        {
            var handler = new SocketsHttpHandler
            {
                // Connection pooling with DNS refresh
                PooledConnectionLifetime    = pooledConnectionLifetime ?? TimeSpan.FromMinutes(2),
                PooledConnectionIdleTimeout = pooledConnectionIdleTimeout ?? TimeSpan.FromMinutes(2),

                // Kinetica-specific optimizations
                AutomaticDecompression      = DecompressionMethods.None,  // Kinetica uses Snappy
                UseCookies                  = false,                       // Not needed for API calls
                AllowAutoRedirect           = false,                       // API endpoints don't redirect
            };

            _client = new HttpClient(handler, disposeHandler: true)
            {
                Timeout = timeout,
            };
            _ownsClient = true;
        }

        internal HttpClientTransport(HttpClient client)
        {
            _client = client;
            _ownsClient = false;
        }

        public byte[] Post(
            string url,
            byte[] body,
            string contentType,
            string? authorization,
            CancellationToken cancellationToken)
        {
            using var request = BuildRequest(url, body, contentType, authorization);
            using var response = _client.Send(request, cancellationToken);
            return ReadOrThrow(response, cancellationToken);
        }

        public async Task<byte[]> PostAsync(
            string url,
            byte[] body,
            string contentType,
            string? authorization,
            CancellationToken cancellationToken)
        {
            using var request = BuildRequest(url, body, contentType, authorization);
            using var response = await _client
                .SendAsync(request, HttpCompletionOption.ResponseHeadersRead, cancellationToken)
                .ConfigureAwait(false);

            var bytes = await response.Content
                .ReadAsByteArrayAsync(cancellationToken)
                .ConfigureAwait(false);

            if (response.IsSuccessStatusCode)
                return bytes;

            throw new KineticaTransportException((int)response.StatusCode, bytes);
        }

        private static HttpRequestMessage BuildRequest(
            string url,
            byte[] body,
            string contentType,
            string? authorization)
        {
            var request = new HttpRequestMessage(HttpMethod.Post, url)
            {
                Content = new ByteArrayContent(body),
                // Force HTTP/1.1 to preserve existing behaviour — Kinetica servers
                // typically do not support HTTP/2 and mis-negotiation would break
                // the connection silently.
                Version = HttpVersion.Version11,
            };
            request.Content.Headers.ContentType  = MediaTypeHeaderValue.Parse(contentType);
            request.Content.Headers.ContentLength = body.Length;

            if (!string.IsNullOrEmpty(authorization))
            {
                var space = authorization.IndexOf(' ');
                if (space > 0)
                {
                    request.Headers.Authorization = new AuthenticationHeaderValue(
                        authorization[..space],
                        authorization[(space + 1)..]);
                }
                else
                {
                    // Handle authorization without scheme (legacy compatibility)
                    request.Headers.Add("Authorization", authorization);
                }
            }

            return request;
        }

        private static byte[] ReadOrThrow(
            HttpResponseMessage response,
            CancellationToken cancellationToken)
        {
            using var stream = response.Content.ReadAsStream(cancellationToken);
            using var buffer = new MemoryStream();
            stream.CopyTo(buffer);
            var bytes = buffer.ToArray();

            if (response.IsSuccessStatusCode)
                return bytes;

            // The Kinetica server encodes error bodies with the same Avro/JSON
            // envelope as success responses. Hand the raw bytes back to the
            // caller via KineticaTransportException so that SubmitRequestToUrlInternal
            // can decode the server's error message.
            throw new KineticaTransportException((int)response.StatusCode, bytes);
        }

        public void Dispose()
        {
            if (_ownsClient)
                _client.Dispose();
        }
    }

    internal sealed class KineticaTransportException : Exception
    {
        public int StatusCode { get; }

        public byte[] Body { get; }

        public KineticaTransportException(int statusCode, byte[] body)
            : base($"Kinetica server returned HTTP {statusCode}.")
        {
            StatusCode = statusCode;
            Body       = body;
        }
    }