|
| 1 | +// Licensed to Elasticsearch B.V under one or more agreements. |
| 2 | +// Elasticsearch B.V licenses this file to you under the Apache 2.0 License. |
| 3 | +// See the LICENSE file in the project root for more information |
| 4 | + |
| 5 | +using System.Buffers; |
| 6 | +using System.Text.Json; |
| 7 | +using Elastic.Ingest.Elasticsearch; |
| 8 | +using Elastic.Ingest.Elasticsearch.Indices; |
| 9 | +using Elastic.Transport; |
| 10 | +using Microsoft.Extensions.Hosting; |
| 11 | +using Microsoft.Extensions.Logging; |
| 12 | + |
| 13 | +namespace Elastic.Documentation.Search; |
| 14 | + |
| 15 | +public record DocumentationDocument |
| 16 | +{ |
| 17 | + public string? Title { get; set; } |
| 18 | +} |
| 19 | + |
| 20 | +public class IngestCollector : IDisposable |
| 21 | +{ |
| 22 | + private readonly IndexChannel<DocumentationDocument> _channel; |
| 23 | + private readonly ILogger<IngestCollector> _logger; |
| 24 | + |
| 25 | + public IngestCollector(ILoggerFactory logFactory, string url, string apiKey) |
| 26 | + { |
| 27 | + _logger = logFactory.CreateLogger<IngestCollector>(); |
| 28 | + var uri = new Uri(url); |
| 29 | + var moniker = $"{uri.Host}${Guid.NewGuid():N}"; |
| 30 | + var base64 = Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes(moniker)); |
| 31 | + var cloudId = $"name:{base64}"; |
| 32 | + |
| 33 | + var pool = new CloudNodePool(cloudId, new ApiKey(apiKey)); |
| 34 | + var configuration = new TransportConfiguration(pool); |
| 35 | + var transport = new DistributedTransport(configuration); |
| 36 | + var options = new IndexChannelOptions<DocumentationDocument>(transport) |
| 37 | + { |
| 38 | + IndexFormat = "documentation", |
| 39 | + ExportExceptionCallback = e => _logger.LogError(e, "Failed to export document"), |
| 40 | + ServerRejectionCallback = items => _logger.LogInformation("Server rejection: {Rejection}", items.First().Item2) |
| 41 | + }; |
| 42 | + _channel = new IndexChannel<DocumentationDocument>(options); |
| 43 | + } |
| 44 | + |
| 45 | + public async ValueTask<bool> TryWrite(DocumentationDocument document, CancellationToken ctx = default) |
| 46 | + { |
| 47 | + if (_channel.TryWrite(document)) |
| 48 | + return true; |
| 49 | + |
| 50 | + if (await _channel.WaitToWriteAsync(ctx)) |
| 51 | + return _channel.TryWrite(document); |
| 52 | + return false; |
| 53 | + } |
| 54 | + |
| 55 | + public void Dispose() |
| 56 | + { |
| 57 | + _channel.Complete(); |
| 58 | + _channel.Dispose(); |
| 59 | + GC.SuppressFinalize(this); |
| 60 | + } |
| 61 | +} |
0 commit comments