spark/src/DataStreamSuite/DataStreamPublisher/Program.cs

using BCG;
using BCG.Utilities.General;

using bts;
using bts.api.core;
using bts.utils;

using Bts.DataStreamExpressions;
using Bts.DataStreamPublisher;
using Bts.PubSub;

using CommandLine;

using mbts.config;

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

await Parser.Default.ParseArguments<DataStreamPublisherConfig>(args)
   .MapResult(RunAsync, errs => Task.FromResult(1));

async Task<int> RunAsync(DataStreamPublisherConfig cfg)
{
   try
   {
      cfg.MergeWithConfigFile();
      cfg.MergeWith(DataStreamPublisherConfig.Default());
      PubSubConfig.SetGlobalConfig(Log.notnull(cfg.PubSub));
      Configuration.initialize(cfg.Sparkd);
      Log.Gen.info("Starting DataStreamPublisher");
      var service_ctx = ServiceContext.create(cfg.Sparkd);
      await service_ctx.login(cfg.User ?? "bts", cfg.Password ?? string.Empty);
      _publish_streams(cfg.Publishers ?? new(0));
   }
   catch (Exception ex)
   {
      // Write the error to console first, in case the logging system is compromised.
      var error_msg = $"DataStreamPublisher threw exception:\n{ex}";
      Console.Error.WriteLine(error_msg);
      Console.Error.Flush();
      bts.Log.Gen.error(error_msg);
      return 1;
   }
   return 0;
}

static ViewStream? build_stream(string? stream_expression)
{
   // Console.WriteLine("Hit key to continue");
   // Console.ReadLine();
   if (stream_expression is null)
   {
      return null;
   }
   try
   {
      bts.Log.Gen.info($"build_stream stream_expression: {stream_expression}");
      var exp = DataStreamExpressionsInterpreter.Evaluate(stream_expression);
      var rv = exp.getView(new Dictionary<string, StatementExpression>(0));
      return rv;

   }
   catch (Exception ex)
   {
      bts.Log.assert(false, $"Could not build stream: {stream_expression} ex: {ex}");
      return null;
   }
}

void _publish_streams(IEnumerable<PublisherNode> publishers)
{
   PubTableBroker pb_broker = new PubTableBroker();
   RemoteQueryManager csm = new RemoteQueryManager();
   var csvPublishers = new List<CSVPublisher>();

   foreach (var publisher in publishers)
   {
      if (Directory.Exists(publisher.CsvDir))
      {
         csvPublishers.Add(new(publisher.CsvDir, pb_broker));
      }
   }

   var viewPublishers = publishers
      .SelectMany(p => p.Streams.empty_if_null().Select(createViewPublisher))
      .ToList();

   while (true)
   {
      System.Threading.Thread.Sleep(TimeSpan.FromSeconds(1));
      ViewStreamDatabase.instance.doWork();
      csm.doWork();
      MasterWorker.getInstance(false).doWork();
      ManagedSubscriber.instance.doWork();
   }

   ViewPublisher? createViewPublisher(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 = build_stream(ps.Expression);
      if (vs is null)
      {
         Log.Gen.warn(
            $"Failed to instantiate ViewPublisher. Invalid 'expression': {JsonSerializer.Serialize(ps)}");
         return null;
      }
      var vpub = new ViewPublisher(ps.Name, pb_broker, vs);
      vs.onChange += on_change;
      return vpub;
   }
}

void on_change(object obj)
{
   bts.Log.Gen.info($"DSPublisher on_change dp: {obj}");
}

public class CSVPublisher : IWorker
{
   private IDictionary<string, ViewPublisher> _pubs = new Dictionary<string, ViewPublisher>();
   private readonly string _csv_directory;
   private PubTableBroker _pub_table_broker;

   public CSVPublisher(string csv_directory, PubTableBroker pub_table_broker)
   {
      _csv_directory = csv_directory;
      _pub_table_broker = pub_table_broker;
      MasterWorker.getInstance(false).addWorker(this, TimeSpan.FromSeconds(1));
   }
   public void doWork()
   {
      var file_names = Directory.GetFiles(_csv_directory);
      foreach (var fn in file_names)
      {
         if (!_pubs.ContainsKey(fn))
         {
            _handle_new_file(fn);
         }
      }
   }

   private void _handle_new_file(string file_name)
   {
      if (file_name.EndsWith(".csv"))
      {
         ViewStream vs = new CSVReaderStream(file_name);
         vs = new SchemaGuesserStream(vs);
         var toks = file_name.Split('\\');
         string table_name = toks.Last();
         table_name = table_name.Remove(table_name.Length - ".csv".Length);
         ViewPublisher vp = new ViewPublisher(table_name, _pub_table_broker, vs);
         _pubs.Add(file_name, vp);
      }
   }
}