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

Program:   ParaView
Module:    TestPTSPipelineViewer.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 "vtkPTSPipelineViewer.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 DoPTSPipelineViewerTest(vtkClientSession* session, rxcpp::schedulers::run_loop& runLoop)
{

  vtkNew<vtkPTSPipelineBuilder> pipelineBuilder;
  vtkNew<vtkPTSPipelineViewer> pipelineViewer;

  pipelineBuilder->SetSession(session);
  pipelineViewer->SetSession(session);

  pipelineViewer->GetObservable().subscribe(
    [](const std::vector<vtkSmartPointer<vtkPipelineElement>>& pipeline) {
      vtkLogF(WARNING, "pipeline size %ld", pipeline.size());
      vtkLogF(INFO, "pipeline : {");
      for (const auto& item : pipeline)
      {
        vtkLogF(INFO, "name %s, item %d , parents %d", item->GetName().c_str(), item->GetID(),
          item->GetParentIDs().empty() ? 0 : item->GetParentIDs()[0]);
      }
      vtkLogF(INFO, " }");
    });

  vtkSmartPointer<vtkSMSourceProxy> source1 = nullptr;
  bool done = false;
  pipelineBuilder->CreateSource("sources", "SphereSource")
    .subscribe([&source1](vtkSMSourceProxy* newProxy) {
      VALIDATE(newProxy);
      source1 = vtk::TakeSmartPointer(newProxy);
    });

  vtkSmartPointer<vtkSMSourceProxy> source2 = nullptr;
  pipelineBuilder->CreateSource("sources", "RTAnalyticSource")
    .subscribe([&source2](vtkSMSourceProxy* newProxy) {
      VALIDATE(newProxy);
      source2 = vtk::TakeSmartPointer(newProxy);
    });

  vtkSmartPointer<vtkSMSourceProxy> source3 = nullptr;
  pipelineBuilder->CreateSource("sources", "CylinderSource")
    .subscribe([&source3, &done](vtkSMSourceProxy* newProxy) {
      VALIDATE(newProxy);
      source3 = vtk::TakeSmartPointer(newProxy);
      done = true;
    });
  // before moving on lets make source all sources have been constructed
  ProcessEventsUntil(runLoop, [&done]() { return done; });
  VALIDATE(source3 != nullptr);

  done = false;
  vtkSmartPointer<vtkSMSourceProxy> filter1;
  pipelineBuilder->CreateFilter("filters", "ShrinkFilter", source1)
    .subscribe([&filter1](vtkSMSourceProxy* newProxy) {
      VALIDATE(newProxy);
      filter1 = vtk::TakeSmartPointer(newProxy);
    });
  vtkSmartPointer<vtkSMSourceProxy> filter2;
  pipelineBuilder->CreateFilter("filters", "ShrinkFilter", source2)
    .subscribe([&filter2](vtkSMSourceProxy* newProxy) {
      VALIDATE(newProxy);
      filter2 = vtk::TakeSmartPointer(newProxy);
    });

  vtkSmartPointer<vtkSMSourceProxy> filter3;
  pipelineBuilder->CreateFilter("filters", "ShrinkFilter", source3)
    .subscribe([&filter3, &done](vtkSMSourceProxy* newProxy) {
      VALIDATE(newProxy);
      filter3 = vtk::TakeSmartPointer(newProxy);
      done = true;
    });

  ProcessEventsUntil(runLoop, [&done]() { return done; });
  VALIDATE(filter3 != nullptr);

  return true;
}

int TestPTSPipelineViewer(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("TestPTSPipelineViewer 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());
  app->CreateBuiltinSession().subscribe(
    [&](vtkTypeUInt32 id) { DoPTSPipelineViewerTest(app->GetSession(id), runLoop); });
  app->WaitForExit(runLoop, std::chrono::seconds(1));
  app->Finalize();
  return app->GetExitCode();
}
