spark/src/DataStreamSuite/Bts.DataStreamWebApp.Server/Controllers/DataStreamSubscriptionController.cs

namespace Bts.DataStreamWebApp.Server.Controllers;

using Bts.DataStream;

using Microsoft.AspNetCore.Mvc;

[Route("/ws")]
public class DataStreamSubscriptionController : ControllerBase {
   readonly ILogger logger;
   readonly IDataStreamFactory streamFactory;

   public DataStreamSubscriptionController(ILogger<DataStreamSubscriptionController> logger, IEnumerable<IDataStreamFactory> streamFactories) {
      this.logger = logger;
      logger.LogInformation("Creating {ControllerType}", GetType().Name);
      var streamFactoryList = streamFactories.ToList();
      streamFactory = streamFactoryList.Count == 1
         ? streamFactoryList[0]
         : new CompositeDataStreamFactory(streamFactoryList);
   }

   [Route("/ws")]
   public async Task Get(CancellationToken cancellationToken) {
      if (!HttpContext.WebSockets.IsWebSocketRequest) {
         logger.LogError("Expected websocket request on path '{}'", HttpContext.Request.Path);
         return;
      }
      logger.LogInformation("Establishing websocket connection...");
      try {
         using var webSocket = await HttpContext.WebSockets.AcceptWebSocketAsync();
         logger.LogInformation("Websocket connection established.");
         using var server = new DataStreamWebSocketServer(webSocket, streamFactory, logger);
         await server.Run(cancellationToken);
      } catch (Exception ex) {
         logger.LogError("DataStreamSubscriptionController.Run threw exception: {Exception}", ex);
      }
   }
}