spark/src/DataStreamSuite/DataStreamPublisher/Services/DataStreamPublisherService.cs

namespace Bts.DataStreamPublisher.Services;

using BCG;

using bts;

using Bts.DataStreamCore;
using Bts.DataStreamExpressions;
using Bts.PubSub;

using global::Spark.Core.Disposables;

using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Options;

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;

/// <summary>
/// An <see cref="IHostedService"/> that publishes <see cref="ViewStream"/>s to the PubSub.
/// It can be added to the application's services via <see cref="DataStreamPublisherServices.AddDataStreamPublisher"/>
/// </summary>
public sealed class DataStreamPublisherService : IHostedService, IDisposable {
   readonly DataStreamPublisherConfig config;
   readonly IPubTableBroker pubTableBroker;
   readonly IWorkScheduler workScheduler;
   List<IDisposable>? resources = new();

   public DataStreamPublisherService(IPubTableBroker pubTableBroker, IWorkScheduler workScheduler, IOptions<DataStreamPublisherConfig> config)
      : this(pubTableBroker, workScheduler, config.Value) { }

   public DataStreamPublisherService(IPubTableBroker pubTableBroker, IWorkScheduler workScheduler, DataStreamPublisherConfig config) {
      this.pubTableBroker = pubTableBroker;
      this.workScheduler = workScheduler;
      this.config = config;
   }

   public Task StartAsync(CancellationToken cancellationToken) {
      cancellationToken.ThrowIfCancellationRequested();
      resources = new();
      foreach (var publisherNode in config.PublisherNodes.empty_if_null()) {
         foreach (var publishedStream in publisherNode.Streams.empty_if_null()) {
            if (string.IsNullOrEmpty(publishedStream.Name)) {
               Log.Gen.warn("Skipping PublishedStream because it doesn't have a name (must define 'Name' in config json).");
               continue;
            }
            var vs = CreateViewStream(publishedStream);
            if (vs is null) {
               continue;
            }
            resources.Add(new ViewPublisher(publishedStream.Name, pubTableBroker, vs));
         }
      }
      var csvPublishers = config.CsvDirs.empty_if_null().Select(csvDir => new CsvPublisher(csvDir, pubTableBroker)).ToList();
      resources.AddRange(csvPublishers);
      resources.Add(workScheduler.Subscribe(new CompositeWorker(csvPublishers)));
      return Task.CompletedTask;
   }

   public Task StopAsync(CancellationToken cancellationToken) {
      cancellationToken.ThrowIfCancellationRequested();
      Dispose();
      return Task.CompletedTask;
   }

   public void Dispose() {
      resources?.DisposeAll();
      resources?.Clear();
   }

   static ViewStream? CreateViewStream(PublishedStream ps) {
      Log.Gen.info(
         $"Instantiating ViewPublisher for: {JsonSerializer.Serialize(ps)}");
      if (string.IsNullOrEmpty(ps.Name)) {
         Log.Gen.warn(
            $"Failed to instantiate ViewPublisher. Null/empty 'name': {JsonSerializer.Serialize(ps)}");
         return null;
      }
      var vs = buildStream(ps.Expression);
      if (vs is null) {
         Log.Gen.warn(
            $"Failed to instantiate ViewPublisher. Invalid 'expression': {JsonSerializer.Serialize(ps)}");
         return null;
      }
      return vs;
   }

   static ViewStream? buildStream(string? streamExpression) {
      if (streamExpression is null) {
         return null;
      }
      try {
         Log.Gen.info($"Building ViewStream: {streamExpression}");
         var exp = DataStreamExpressionsInterpreter.Evaluate(streamExpression);
         return exp.getView(new Dictionary<string, StatementExpression>(0));
      } catch (Exception ex) {
         Log.assert(false, $"Could not build stream. expression: {streamExpression}; ex: {ex}");
         return null;
      }
   }
}