spark/src/DataStreamSuite/DataStream/Offloading/RemoteQueryManager.cs

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

using bts;

using Microsoft.Extensions.Hosting;

using Spark.Core.Disposables;

using System;
using System.Collections.Generic;
using System.Reactive.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace BCG {
   // listens to the requests posted on "cloud_streams" request table, and 
   // if unfulfilled starts a ViewPubsliher
   public class RemoteQueryManager : IHostedService {
      public class StreamPubUnit {
         public string name;
         public string stream_expression;
         public ViewStream stream;
         public bool persist;
         public ViewPublisher view_publisher;

         public StreamPubUnit(string name_, string stream_expression_, ViewStream stream_, bool persist_, ViewPublisher view_publisher_) {
            name = name_;
            stream_expression = stream_expression_;
            stream = stream_;
            persist = persist_;
            view_publisher = view_publisher_;
         }
      }
      static ViewStream? build_stream(string stream_expression)
      {
         try
         {
            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;
         }
      }
      private IDictionary<string, StreamPubUnit>
         _stream_pub_map = new Dictionary<string, StreamPubUnit>();
      readonly IPubTableBroker _broker;
      readonly IManagedSubscriber _sub;
      readonly object syncRoot = new();
      IDisposable? cloudStreamsSub;
      IDisposable? remoteRequestsSub;
      bool isDisposed;


      public RemoteQueryManager(IManagedSubscriber managedSubscriber, IPubTableBroker pubTableBroker) {
         _sub = managedSubscriber;
         _broker = pubTableBroker;
      }

      public Task StartAsync(CancellationToken cancellationToken) {
         lock (syncRoot) {
            Log.assert(!isDisposed);
            Log.isnull(cloudStreamsSub);
            Log.isnull(remoteRequestsSub);
            cloudStreamsSub = new TableStream("cloud_streams").AsObservable().Subscribe(_on_str);
            remoteRequestsSub = _sub.Observe("RemoteRequest").Subscribe(data => {
               RemoteRequest req = RemoteRequest.construct_from_bytes(data);
               _spawn_remote_stream(req.published_table_name, req.stream_expression, false);
            });
         }
         return Task.CompletedTask;
      }

      public Task StopAsync(CancellationToken cancellationToken) {
         lock (syncRoot) {
            if (!isDisposed) {
               isDisposed = true;
               Disposable.dispose(ref cloudStreamsSub);
               Disposable.dispose(ref remoteRequestsSub);
            }
         }
         return Task.CompletedTask;
      }

      private void _on_str(DataPacket dp)
      {
         foreach (IBlueRow r in dp.newRows)
         {
            string name = r.get<string>("name");
            bool persist = r.get<bool>("persist");
            string stream_exp = r.get<string>("stream_exp");
            if (!_stream_pub_map.ContainsKey(name)) {
               _spawn_remote_stream(name, stream_exp, persist);
            }
         }
      }

      private void _spawn_remote_stream(string name, string stream_expression, bool persist) {
         bts.Log.Gen.info($"CloudStreamManager.assign name: {name} stream_expression: {stream_expression}");
         ViewStream stream = build_stream(stream_expression);
         _expose_to_ws(name, stream);
         if (persist) {
            stream = new PersistorStream(stream, name, "C:\\Temp\\", 1000000);   // 10MBs
         }
         ViewPublisher view_pub = new ViewPublisher(name, _broker, stream);
         StreamPubUnit u = new StreamPubUnit(name, stream_expression, stream, persist, view_pub);
         _stream_pub_map.Add(name, u);
      }

      private void _expose_to_ws(string name, ViewStream stream) {
         // TODO: john
      }
   }
}