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

Program:   ParaView
Module:    TestPTSActiveObjects.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.

=========================================================================*/
/**
 * Tests vtkPTSActiveObjects
 * Validates that the two observables of vtkPTSActiveObjects are emitting the expected results as
 * new proxies are created and selected.
 */
#include "vtkClientSession.h"
#include "vtkDistributedEnvironment.h"
#include "vtkLogger.h"
#include "vtkPTSActiveObjects.h"
#include "vtkPTSPipelineBuilder.h"
#include "vtkPVApplication.h"
#include "vtkPVApplicationOptions.h"
#include "vtkSMProxy.h"
#include "vtkSMProxySelectionModel.h"
#include "vtkSMSessionProxyManager.h"
#include "vtkSMSourceProxy.h"
#include "vtkSmartPointer.h"
#include "vtkTestUtilities.h"

#include <vector>
#include <vtk_cli11.h>

#define VALIDATE(x)                                                                                \
  if (!(x))                                                                                        \
  {                                                                                                \
    vtkLogF(ERROR, "Validating : (" #x ")...failed!");                                             \
  }

template <typename Predicate>
bool ProcessEventsUntil(rxcpp::schedulers::run_loop& runLoop, Predicate stopProcessing,
  const std::chrono::milliseconds& timeout = std::chrono::milliseconds(1000))
{
  const auto start = std::chrono::system_clock::now();
  do
  {
    std::this_thread::sleep_for(std::chrono::milliseconds(5));
    while ((!runLoop.empty() && runLoop.peek().when < runLoop.now()))
    {
      runLoop.dispatch();
    }
  } while (!stopProcessing() && timeout > (std::chrono::system_clock::now() - start));
  return stopProcessing();
}

bool DoPTSActiveObjectsTest(vtkClientSession* session, rxcpp::schedulers::run_loop& runLoop)
{

  int errorsCaught = 0;

  vtkNew<vtkPTSPipelineBuilder> pipelineBuilder;
  vtkNew<vtkPTSActiveObjects> activeObjects;

  pipelineBuilder->SetSession(session);
  activeObjects->SetSession(session);
  activeObjects->SetModelName("ActiveSources");

  // record active objects for validation
  std::vector<std::string> activeObjectsResults;

  activeObjects->GetCurrentObservable().subscribe(
    [&activeObjectsResults](vtkSmartPointer<vtkSMProxy> proxy) {
      if (proxy)
      {
        activeObjectsResults.push_back(proxy->GetXMLName());
      }
      else
      {
        activeObjectsResults.push_back("nullptr");
      }
    });

  // record selections for validation
  std::vector<std::string> selectionResults;
  activeObjects->GetSelectionObservable().subscribe(
    [&selectionResults](const std::vector<vtkSmartPointer<vtkSMProxy>>& selection) {
      if (selection.empty())
      {
        selectionResults.push_back("<empty>");
      }
      else
      {
        std::string selectionString;
        for (auto proxy : selection)
        {
          selectionString.append(proxy->GetXMLName());
          selectionString.append(",");
        }
        selectionString.pop_back(); // remove last ","
        selectionResults.emplace_back(selectionString);
      }
    });

  vtkSmartPointer<vtkSMSourceProxy> source = nullptr;
  bool done = false;
  // creating a proxy updates "ActiveSources" CurrentProxy
  pipelineBuilder->CreateSource("sources", "SphereSource")
    .subscribe([&source, &done](vtkSMSourceProxy* newProxy) {
      VALIDATE(newProxy);
      source = vtk::TakeSmartPointer(newProxy);
      done = true;
    });

  // this should NOT emit any value since selection remained the same
  activeObjects->Select(source, vtkSMProxySelectionModel::SELECT);
  ProcessEventsUntil(runLoop, [&done]() { return done; });
  VALIDATE(source != nullptr);

  vtkSmartPointer<vtkSMSourceProxy> filter = nullptr;
  done = false;
  pipelineBuilder->CreateFilter("filters", "ShrinkFilter", source)
    .subscribe(
      [&filter, &done](vtkSMSourceProxy* newProxy) {
        VALIDATE(newProxy);
        filter = vtk::TakeSmartPointer(newProxy);
        done = true;
      },
      [&errorsCaught](
        rxcpp::util::error_ptr ep) { vtkLogF(ERROR, "%s", rxcpp::util::what(ep).c_str()); }

    );

  ProcessEventsUntil(runLoop, [&done]() { return done; });
  VALIDATE(filter != nullptr);
  // add source to selection
  activeObjects->Select(source, vtkSMProxySelectionModel::SELECT);

  // deleting the active source should set current active source to nullptr and selection to empty;
  pipelineBuilder->DeleteProxy(filter).subscribe([&done](bool status) {
    VALIDATE(status);
    done = true;
  });
  ProcessEventsUntil(runLoop, [&done]() { return done; });

  // set active source manually
  activeObjects->SetCurrentProxy(source, vtkSMProxySelectionModel::CLEAR);

  VALIDATE(errorsCaught == 0);

  const std::vector<std::string> expectedActiveObjects = { "nullptr", "SphereSource",
    "ShrinkFilter", "nullptr", "SphereSource" };

  const std::vector<std::string> expectedSelections = { "<empty>", "SphereSource", "ShrinkFilter",
    "ShrinkFilter,SphereSource", "<empty>" };

  VALIDATE(activeObjectsResults == expectedActiveObjects);
  VALIDATE(selectionResults == expectedSelections);

  return true;
}

int TestPTSActiveObjects(int argc, char* argv[])
{
  // Create the primary event-loop for the application.
  rxcpp::schedulers::run_loop runLoop;

  // initialize the MPI environment.
  vtkDistributedEnvironment environment(&argc, &argv);
  const int rank = environment.GetRank();

  // determine what mode this test is running under
  // and create appropriate application type.
  vtkNew<vtkPVApplication> app;

  // process command line arguments.
  CLI::App cli("TestPTSActiveObjects application");

  // let's use nicer formatter for help text.
  vtkPVApplicationOptions::SetupDefaults(cli);

  app->GetOptions()->Populate(cli);

  try
  {
    cli.parse(argc, argv);
  }
  catch (const CLI::ParseError& e)
  {
    return (rank == 0 ? cli.exit(e) : EXIT_SUCCESS);
  }

  app->Initialize(argv[0], rxcpp::observe_on_run_loop(runLoop), environment.GetController());

  rxcpp::observable<vtkTypeUInt32> sessionIdObservable;

  auto* options = app->GetOptions();

  if (options->GetServerURL().empty())
  {
    sessionIdObservable = app->CreateBuiltinSession();
  }
  else
  {
    sessionIdObservable = app->CreateRemoteSession(options->GetServerURL());
  }

  sessionIdObservable.subscribe(
    [&](vtkTypeUInt32 id) { DoPTSActiveObjectsTest(app->GetSession(id), runLoop); });
  app->WaitForExit(runLoop, std::chrono::seconds(1));
  app->Finalize();
  return app->GetExitCode();
}
