/*=========================================================================

  Program:   ParaView
  Module:    vtkPTSProgressObserver.cxx

  Copyright (c) Kitware, Inc.
  All rights reserved.
  See Copyright.txt or http://www.paraview.org/HTML/Copyright.html for details.

     This software is distributed WITHOUT ANY WARRANTY; without even
     the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
     PURPOSE.  See the above copyright notice for more information.

=========================================================================*/
#include "vtkPTSProgressObserver.h"
#include "vtkClientSession.h"
#include "vtkObjectFactory.h"
#include "vtkRemotingCoreUtilities.h"
#include "vtkSMSessionProxyManager.h"
#include "vtkSmartPointer.h"
#include <vector>

#include "vtk_rxcpp.h" // for rxcpp
// clang-format off
#include VTK_REMOTING_RXCPP(rx-util.hpp) // for rxcpp::util::apply_to
// clang-format on

//=============================================================================
class vtkPTSProgressObserver::vtkInternals
{
public:
  const std::thread::id OwnerTID{ std::this_thread::get_id() };

  std::vector<rxcpp::subscriber<std::string>> Subscribers;
  vtkSmartPointer<vtkClientSession> Session;
};
//=============================================================================
vtkStandardNewMacro(vtkPTSProgressObserver);
//-----------------------------------------------------------------------------
vtkPTSProgressObserver::vtkPTSProgressObserver()
  : Internals(new vtkPTSProgressObserver::vtkInternals())
{
}

//-----------------------------------------------------------------------------
vtkPTSProgressObserver::~vtkPTSProgressObserver() = default;

//-----------------------------------------------------------------------------
void vtkPTSProgressObserver::SetSession(vtkClientSession* session)
{
  auto& internals = (*this->Internals);
  vtkRemotingCoreUtilities::EnsureThread(internals.OwnerTID);
  if (internals.Session != session)
  {
    internals.Session = session;
  }
}

//-----------------------------------------------------------------------------
vtkClientSession* vtkPTSProgressObserver::GetSession() const
{
  auto& internals = (*this->Internals);
  vtkRemotingCoreUtilities::EnsureThread(internals.OwnerTID);
  return internals.Session;
}

//-----------------------------------------------------------------------------
void vtkPTSProgressObserver::Unsubscribe()
{
  auto& internals = (*this->Internals);
  for (auto& subscriber : internals.Subscribers)
  {
    subscriber.on_completed();
  }

  internals.Subscribers.clear();
}

//-----------------------------------------------------------------------------
rxcpp::observable<std::string> vtkPTSProgressObserver::GetServiceProgressObservable(
  const std::string& serviceName) const
{
  vtkRemotingCoreUtilities::EnsureThread(this->Internals->OwnerTID);
  auto observable = rxcpp::observable<>::create<std::string>(
    [this, serviceName](rxcpp::subscriber<std::string> subscriber) {
      auto& internals = (*this->Internals);
      // setup subscription
      internals.Session->GetProgressObservable()
        .filter(
          rxcpp::util::apply_to([serviceName](const std::string& _serviceName,
                                  const vtkRemoteObjectProvider::vtkProgressItem& /* item */) {
            return _serviceName == serviceName;
          }))
        .transform(rxcpp::util::apply_to([](const std::string& _serviceName,
                                           const vtkRemoteObjectProvider::vtkProgressItem& item) {
          // ASYNC FIXME we need to return a proper object
          vtkNJson object = vtkNJson::object();
          to_json(object, item);
          vtkNJson data;
          data[_serviceName] = object;
          return data.dump();
        }))
        .distinct_until_changed()
        .subscribe(subscriber);

      // keep track of them so we can Unsubscribe
      internals.Subscribers.push_back(subscriber);
    });

  return observable;
}

//-----------------------------------------------------------------------------
rxcpp::observable<std::string> vtkPTSProgressObserver::GetServiceMetricsObservable(
  const std::string& serviceName) const
{
  vtkRemotingCoreUtilities::EnsureThread(this->Internals->OwnerTID);
  auto observable = rxcpp::observable<>::create<std::string>(
    [this, serviceName](rxcpp::subscriber<std::string> subscriber) {
      auto& internals = (*this->Internals);
      // setup subscription
      internals.Session->GetRemoteMetricInformationObservable()
        .filter(rxcpp::util::apply_to(
          [serviceName](const std::string& _serviceName, const vtkNJson& /* item */) {
            return _serviceName == serviceName;
          }))
        .transform(rxcpp::util::apply_to([](const std::string& _serviceName, const vtkNJson& item) {
          vtkNJson data;
          data[_serviceName] = item;
          return data.dump();
        }))
        .distinct_until_changed()
        .subscribe(subscriber);

      // keep track of them so we can Unsubscribe
      internals.Subscribers.push_back(subscriber);
    });

  return observable;
}

//-----------------------------------------------------------------------------
void vtkPTSProgressObserver::PrintSelf(ostream& os, vtkIndent indent)
{
  this->Superclass::PrintSelf(os, indent);
}
