Client.cs

Go to the documentation of this file.

using System.Globalization;
using Fauna.Core;
using Fauna.Exceptions;
using Fauna.Linq;
using Fauna.Mapping;
using Fauna.Serialization;
using Fauna.Types;
using Stream = System.IO.Stream;
 
namespace Fauna;
 
public class Client : BaseClient, IDisposable
{
    private const string QueryUriPath = "/query/1";
    private const string StreamUriPath = "/stream/1";
    private const string FeedUriPath = "/feed/1";
 
    private readonly Configuration _config;
    private readonly IConnection _connection;
 
    private readonly MappingContext _defaultCtx = new();
    private readonly Dictionary<Type, DataContext> _dbCtxs = new();
 
    private bool _disposed;
 
    internal override MappingContext MappingCtx { get => _defaultCtx; }
 
    public readonly IStatsCollector? StatsCollector;
 
    public long LastSeenTxn { get; private set; }
 
    public Client() : this(new Configuration()) { }
 
    public Client(string secret) :
        this(new Configuration(secret))
    {
    }
 
    public Client(Configuration config)
    {
        config.Validate();
        _config = config;
        StatsCollector = config.StatsCollector;
        _connection = new Connection(config);
    }
 
    public DB DataContext<DB>() where DB : DataContext
    {
        var dbCtxType = typeof(DB);
        DataContext? ctx;
        lock (_dbCtxs)
        {
            if (!_dbCtxs.TryGetValue(dbCtxType, out ctx))
            {
                var builder = new DataContextBuilder<DB>();
                ctx = builder.Build(this);
                _dbCtxs[dbCtxType] = ctx;
            }
        }
 
        return (DB)ctx;
    }
 
    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
    }
 
    ~Client()
    {
        Dispose(false);
    }
 
    internal override async Task<QuerySuccess<T>> QueryAsyncInternal<T>(
        Query query,
        ISerializer<T> serializer,
        MappingContext ctx,
        QueryOptions? queryOptions,
        CancellationToken cancel)
    {
        if (query == null)
        {
            throw new ArgumentNullException(nameof(query));
        }
 
        var finalOptions = QueryOptions.GetFinalQueryOptions(_config.DefaultQueryOptions, queryOptions);
        var headers = GetRequestHeaders(finalOptions);
 
        using var stream = new MemoryStream();
        Serialize(stream, query, ctx);
 
        using var httpResponse = await _connection.DoPostAsync(
            QueryUriPath,
            stream,
            headers,
            GetRequestTimeoutWithBuffer(finalOptions.QueryTimeout),
            cancel);
        var body = await httpResponse.Content.ReadAsStringAsync(cancel);
        var res = QueryResponse.GetFromResponseBody<T>(ctx, serializer, httpResponse.StatusCode, body);
        switch (res)
        {
            case QuerySuccess<T> success:
                LastSeenTxn = res.LastSeenTxn;
                StatsCollector?.Add(res.Stats);
                return success;
            case QueryFailure failure:
                StatsCollector?.Add(res.Stats);
                throw ExceptionHandler.FromQueryFailure(ctx, failure);
            default:
                throw ExceptionHandler.FromRawResponse(body, httpResponse);
        }
    }
 
    internal override async IAsyncEnumerator<Event<T>> SubscribeStreamInternal<T>(
        Types.EventSource eventSource,
        MappingContext ctx,
        CancellationToken cancel = default)
    {
        var finalOptions = QueryOptions.GetFinalQueryOptions(_config.DefaultQueryOptions, null);
        var headers = GetRequestHeaders(finalOptions);
 
        await foreach (var evt in _connection.OpenStream<T>(
                           StreamUriPath,
                           eventSource,
                           headers,
                           ctx,
                           cancel))
        {
            LastSeenTxn = evt.TxnTime;
            eventSource.Options.Cursor = evt.Cursor;
 
            StatsCollector?.Add(evt.Stats);
            yield return evt;
        }
    }
    internal override async IAsyncEnumerator<FeedPage<T>> SubscribeFeedInternal<T>(
        Types.EventSource eventSource,
        MappingContext ctx,
        CancellationToken cancel = default)
    {
        cancel.ThrowIfCancellationRequested();
 
        var finalOptions = _config.DefaultQueryOptions;
        var headers = GetRequestHeaders(finalOptions);
 
        while (!cancel.IsCancellationRequested)
        {
            var feedData = new MemoryStream();
            eventSource.Serialize(feedData);
 
            using var httpResponse = await _connection.DoPostAsync(
                FeedUriPath,
                feedData,
                headers,
                GetRequestTimeoutWithBuffer(finalOptions.QueryTimeout),
                cancel);
            string body = await httpResponse.Content.ReadAsStringAsync(cancel);
 
            var res = FeedPage<T>.From(body, ctx);
            eventSource.Options.Cursor = res.Cursor;
 
            StatsCollector?.Add(res.Stats);
            yield return res;
            if (!res.HasNext)
            {
                break;
            }
        }
    }
 
    private void Serialize(Stream stream, Query query, MappingContext ctx)
    {
        using var writer = new Utf8FaunaWriter(stream);
        writer.WriteStartObject();
        writer.WriteFieldName("query");
        query.Serialize(ctx, writer);
        writer.WriteEndObject();
        writer.Flush();
    }
 
    private Dictionary<string, string> GetRequestHeaders(QueryOptions queryOptions)
    {
        var headers = new Dictionary<string, string>
        {
            { Headers.Authorization, $"Bearer {_config.Secret}"},
            { Headers.Format, "tagged" },
            { Headers.Driver, "C#" }
        };
 
        if (LastSeenTxn > long.MinValue)
        {
            headers.Add(Headers.LastTxnTs, LastSeenTxn.ToString());
        }
 
        if (queryOptions.QueryTimeout != TimeSpan.Zero)
        {
            headers.Add(
                Headers.QueryTimeoutMs,
                queryOptions.QueryTimeout.TotalMilliseconds.ToString(CultureInfo.InvariantCulture));
        }
 
        if (queryOptions.QueryTags != null)
        {
            headers.Add(Headers.QueryTags, EncodeQueryTags(queryOptions.QueryTags));
        }
 
        if (!string.IsNullOrEmpty(queryOptions.TraceParent))
        {
            headers.Add(Headers.TraceParent, queryOptions.TraceParent);
        }
 
        if (queryOptions.Linearized != null)
        {
            headers.Add(Headers.Linearized, queryOptions.Linearized.ToString()!);
        }
 
        if (queryOptions.TypeCheck != null)
        {
            headers.Add(Headers.TypeCheck, queryOptions.TypeCheck.ToString()!);
        }
 
        return headers;
    }
 
    private TimeSpan GetRequestTimeoutWithBuffer(TimeSpan queryTimeout)
    {
        return queryTimeout.Add(_config.ClientBufferTimeout);
    }
 
    private static string EncodeQueryTags(Dictionary<string, string> tags)
    {
        return string.Join(",", tags.Select(entry => entry.Key + "=" + entry.Value));
    }
 
    private void Dispose(bool disposing)
    {
        if (_disposed) return;
 
        if (disposing)
        {
            _connection.Dispose();
            GC.SuppressFinalize(this);
        }
        _disposed = true;
    }
}