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

  Program:   ParaView
  Module:    vtkPTSActiveObjects.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 "vtkPTSActiveObjects.h"

#include "vtkClientSession.h"
#include "vtkObjectFactory.h"
#include "vtkPVCoreApplication.h"
#include "vtkReactiveCommand.h"
#include "vtkRemotingCoreUtilities.h"
#include "vtkSMParaViewPipelineController.h"
#include "vtkSMProxy.h"
#include "vtkSMProxySelectionModel.h"
#include "vtkSMSessionProxyManager.h"
#include "vtkSmartPointer.h"

#include <exception> // for std::runtime_error

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

  rxcpp::subjects::subject<vtkSmartPointer<vtkSMProxy>> Active;
  rxcpp::subjects::subject<std::vector<vtkSmartPointer<vtkSMProxy>>> Selection;

  vtkSmartPointer<vtkClientSession> Session;
  std::string ModelName;

  rxcpp::composite_subscription ChangeEventSubscription;

  std::vector<rxcpp::subscriber<vtkSmartPointer<vtkSMProxy>>> ActiveSubscribers;
  std::vector<rxcpp::subscriber<std::vector<vtkSmartPointer<vtkSMProxy>>>> SelectionSubscribers;

  ~vtkInternals() { this->UnsubscribeAll(); }
  void UnsubscribeAll()
  {
    for (auto& s : this->ActiveSubscribers)
    {
      s.on_completed();
    }
    this->ActiveSubscribers.clear();

    for (auto& s : this->SelectionSubscribers)
    {
      s.on_completed();
    }
    this->SelectionSubscribers.clear();
  }

  void DoNext()
  {
    vtkRemotingCoreUtilities::EnsureThread(this->OwnerTID);
    this->Active.get_subscriber().on_next(this->GetActive());
    this->Selection.get_subscriber().on_next(this->GetSelection());
  }

  vtkSmartPointer<vtkSMProxy> GetActive() const
  {
    auto* selModel = this->GetSelectionModel();
    return selModel ? selModel->GetCurrentProxy() : nullptr;
  }

  std::vector<vtkSmartPointer<vtkSMProxy>> GetSelection() const
  {
    if (auto* selModel = this->GetSelectionModel())
    {
      const auto& selection = selModel->GetSelection();

      std::vector<vtkSmartPointer<vtkSMProxy>> result;
      result.reserve(selection.size());
      std::copy(selection.begin(), selection.end(), std::back_inserter(result));
      return result;
    }
    return {};
  }

  vtkSMProxySelectionModel* GetSelectionModel() const
  {
    auto* pxm = this->Session ? this->Session->GetProxyManager() : nullptr;
    return pxm ? pxm->GetSelectionModel(this->ModelName.c_str()) : nullptr;
  }

  void SetupEventSubscriptions()
  {
    if (auto* selModel = this->GetSelectionModel())
    {
      this->ChangeEventSubscription.clear();
      this->ChangeEventSubscription.add(
        rxvtk::from_event(selModel, vtkCommand::CurrentChangedEvent)
          .subscribe([this](const rxvtk::ObservedType /*item*/) { this->DoNext(); }));

      this->ChangeEventSubscription.add(
        rxvtk::from_event(selModel, vtkCommand::SelectionChangedEvent)
          .subscribe([this](const rxvtk::ObservedType /*item*/) { this->DoNext(); }));
    }
  }
};

vtkStandardNewMacro(vtkPTSActiveObjects);
//-----------------------------------------------------------------------------
vtkPTSActiveObjects::vtkPTSActiveObjects()
  : Internals(new vtkPTSActiveObjects::vtkInternals())
{
}

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

//-----------------------------------------------------------------------------
void vtkPTSActiveObjects::SetSession(vtkClientSession* session)
{
  auto& internals = (*this->Internals);
  vtkRemotingCoreUtilities::EnsureThread(internals.OwnerTID);
  if (internals.Session != session)
  {
    vtkNew<vtkSMParaViewPipelineController> controller;
    controller->InitializeSession(session);
    internals.Session = session;
    internals.DoNext();
    internals.SetupEventSubscriptions();
  }
}

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

//-----------------------------------------------------------------------------
void vtkPTSActiveObjects::SetModelName(const std::string& name)
{
  auto& internals = (*this->Internals);
  vtkRemotingCoreUtilities::EnsureThread(internals.OwnerTID);
  if (internals.ModelName != name)
  {
    internals.ModelName = name;
    internals.DoNext();
    internals.SetupEventSubscriptions();
  }
}

//-----------------------------------------------------------------------------
const std::string& vtkPTSActiveObjects::GetModelName() const
{
  const auto& internals = (*this->Internals);
  vtkRemotingCoreUtilities::EnsureThread(internals.OwnerTID);
  return internals.ModelName;
}

//-----------------------------------------------------------------------------
rxcpp::observable<vtkSmartPointer<vtkSMProxy>> vtkPTSActiveObjects::GetCurrentObservable() const
{
  vtkRemotingCoreUtilities::EnsureThread(this->Internals->OwnerTID);

  auto observable = rxcpp::observable<>::create<vtkSmartPointer<vtkSMProxy>>(
    [this](rxcpp::subscriber<vtkSmartPointer<vtkSMProxy>> subscriber) {
      auto& internals = (*this->Internals);
      if (!internals.Active.get_subscriber().is_subscribed())
      {
        throw std::runtime_error(
          "UnsubscribeAll() already called! Need to create new vtkPTSActiveObjects instance.");
      }

      // setup subscription
      internals.Active.get_observable().subscribe(subscriber);
      internals.ActiveSubscribers.push_back(subscriber);

      // pass current value.
      subscriber.on_next(internals.GetActive());
    }).distinct_until_changed();

  auto* pvapp = vtkPVCoreApplication::GetInstance();
  assert(pvapp != nullptr);
  return observable;
}

//-----------------------------------------------------------------------------
rxcpp::observable<std::vector<vtkSmartPointer<vtkSMProxy>>>
vtkPTSActiveObjects::GetSelectionObservable() const
{
  vtkRemotingCoreUtilities::EnsureThread(this->Internals->OwnerTID);

  auto observable = rxcpp::observable<>::create<std::vector<vtkSmartPointer<vtkSMProxy>>>(
    [this](rxcpp::subscriber<std::vector<vtkSmartPointer<vtkSMProxy>>> subscriber) {
      auto& internals = (*this->Internals);

      if (!internals.Selection.get_subscriber().is_subscribed())
      {
        throw std::runtime_error(
          "UnsubscribeAll() already called! Need to create new vtkPTSActiveObjects instance.");
      }

      // setup subscription
      internals.Selection.get_observable().subscribe(subscriber);
      internals.SelectionSubscribers.push_back(subscriber);

      // pass current value.
      subscriber.on_next(internals.GetSelection());
    }).distinct_until_changed();

  auto* pvapp = vtkPVCoreApplication::GetInstance();
  assert(pvapp != nullptr);
  return observable;
}

//-----------------------------------------------------------------------------
void vtkPTSActiveObjects::SetCurrentProxy(vtkSMProxy* proxy, int command)
{
  auto& internals = (*this->Internals);
  if (auto* selModel = internals.GetSelectionModel())
  {
    selModel->SetCurrentProxy(proxy, command);
  }
}

//-----------------------------------------------------------------------------
void vtkPTSActiveObjects::Select(
  const std::vector<vtkSmartPointer<vtkSMProxy>>& proxies, int command)
{
  auto& internals = (*this->Internals);
  if (auto* selModel = internals.GetSelectionModel())
  {
    vtkSMProxySelectionModel::SelectionType proxyList(proxies.begin(), proxies.end());
    selModel->Select(proxyList, command);
  }
}

//-----------------------------------------------------------------------------
void vtkPTSActiveObjects::Select(vtkSMProxy* proxy, int command)
{
  auto& internals = (*this->Internals);
  if (auto* selModel = internals.GetSelectionModel())
  {
    selModel->Select(proxy, command);
  }
}

//-----------------------------------------------------------------------------
void vtkPTSActiveObjects::UnsubscribeAll()
{
  auto& internals = (*this->Internals);
  internals.UnsubscribeAll();
}

//-----------------------------------------------------------------------------
void vtkPTSActiveObjects::PrintSelf(ostream& os, vtkIndent indent)
{
  this->Superclass::PrintSelf(os, indent);
  const auto& internals = (*this->Internals);
  os << indent << "Session: " << internals.Session << "\n";
  os << indent << "Selection Model Name: " << internals.ModelName << "\n";
  os << indent << "SelectionModel:  " << internals.GetSelectionModel() << "\n";
  os << indent
     << "Number of subscriptions for CurrentObservable: " << internals.ActiveSubscribers.size()
     << "\n";
  os << indent
     << "Number of subscriptions for SelectionObservable: " << internals.SelectionSubscribers.size()
     << "\n";
}
