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

  Program:   ParaView
  Module:    vtkPTSPipelineViewer.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 "vtkPTSPipelineViewer.h"
#include "vtkClientSession.h"
#include "vtkCollection.h"
#include "vtkObjectFactory.h"
#include "vtkReactiveCommand.h"
#include "vtkRemotingCoreUtilities.h"
#include "vtkSMProxy.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

vtkStandardNewMacro(vtkPipelineElement);

//-----------------------------------------------------------------------------
vtkPipelineElement::vtkPipelineElement() = default;
//-----------------------------------------------------------------------------
vtkPipelineElement::~vtkPipelineElement() = default;
//-----------------------------------------------------------------------------
void vtkPipelineElement::PrintSelf(std::ostream& os, vtkIndent indent)
{
  this->Superclass::PrintSelf(os, indent);
  os << indent << "ID : " << this->GetID() << "\n";
  os << indent << "name : " << this->GetName() << "\n";
  auto nextIndent = indent.GetNextIndent();
  os << indent << "Parent IDs:\n";
  for (int i = 0; i < this->GetParentIDs().size(); i++)
  {
    os << nextIndent << this->GetParentIDs()[i];
    if (i < this->GetParentIDs().size() - 1)
    {
      os << " , ";
    }
  }
  os << "\n";
}

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

  rxcpp::subjects::subject<std::vector<vtkSmartPointer<vtkPipelineElement>>> Connectivity;
  vtkSmartPointer<vtkClientSession> Session;

private:
  rxcpp::composite_subscription ChangeEventSubscription;

public:
  ~vtkInternals() { this->Connectivity.get_subscriber().on_completed(); }

  void DoNext()
  {
    vtkRemotingCoreUtilities::EnsureThread(this->OwnerTID);
    this->Connectivity.get_subscriber().on_next(this->GetConnectivity());
  }

  std::vector<vtkSmartPointer<vtkPipelineElement>> GetConnectivity()
  {
    if (this->Session == nullptr)
    {
      return {};
    }

    std::vector<vtkSmartPointer<vtkPipelineElement>> result;

    // Collect Pipeline Objects
    auto* proxyManager = this->Session->GetProxyManager();
    vtkNew<vtkCollection> proxies;
    const std::vector<std::string> groupsToLook = { "sources", "filters", "readers" };
    for (const auto& groupName : groupsToLook)
    {
      proxyManager->GetProxies(groupName.c_str(), proxies);
      proxies->InitTraversal();
      while (vtkSMProxy* proxy = vtkSMProxy::SafeDownCast(proxies->GetNextItemAsObject()))
      {
        vtkNew<vtkPipelineElement> element;
        element->SetProxy(proxy);
        // FIXME name is set aftrer registration so GetLogName is not ready yet
        element->SetName(proxyManager->GetProxyName(groupName.c_str(), proxy));
        result.push_back(element);
      }
    }
    // sort by GlobalID to get the same ordering each time.
    std::sort(result.begin(), result.end(),
      [](const vtkSmartPointer<vtkPipelineElement>& a,
        const vtkSmartPointer<vtkPipelineElement>& b) { return a->GetID() < b->GetID(); });

    return result;
  }

  void SetupEventSubscriptions()
  {
    if (this->Session)
    {
      auto* proxyManager = this->Session->GetProxyManager();
      auto isPipelineElement = [](vtkSMSessionProxyManager::RegisteredProxyInformation* info) {
        return strcmp(info->GroupName, "sources") == 0 || strcmp(info->GroupName, "filters") == 0 ||
          strcmp(info->GroupName, "readers") == 0;
      };

      rxvtk::from_event(proxyManager, vtkCommand::RegisterEvent)
        .filter(rxcpp::util::apply_to(
          [&](vtkObject* /*object*/, unsigned long /*eventId*/, void* callData) {
            vtkSMSessionProxyManager::RegisteredProxyInformation* info =
              static_cast<vtkSMSessionProxyManager::RegisteredProxyInformation*>(callData);
            return isPipelineElement(info);
          }))
        .subscribe([this](const rxvtk::ObservedType /*item*/) { this->DoNext(); });

      rxvtk::from_event(proxyManager, vtkCommand::UnRegisterEvent)
        .filter(rxcpp::util::apply_to(
          [&](vtkObject* /*object*/, unsigned long /*eventId*/, void* callData) {
            vtkSMSessionProxyManager::RegisteredProxyInformation* info =
              static_cast<vtkSMSessionProxyManager::RegisteredProxyInformation*>(callData);
            return isPipelineElement(info);
          }))
        .subscribe([this](const rxvtk::ObservedType /*item*/) { this->DoNext(); });
    }
  }
};

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

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

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

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

//-----------------------------------------------------------------------------
void vtkPTSPipelineViewer::UnsubscribeAll()
{
  this->Internals->Connectivity.get_subscriber().on_completed();
}

//-----------------------------------------------------------------------------
rxcpp::observable<std::vector<vtkSmartPointer<vtkPipelineElement>>>
vtkPTSPipelineViewer::GetObservable() const
{
  vtkRemotingCoreUtilities::EnsureThread(this->Internals->OwnerTID);

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

      // setup subscription
      internals.Connectivity.get_observable().subscribe(subscriber);

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

  return observable;
}

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