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

Program:   ParaView
Module:    TestPTSPipelineBuilder.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 vtkPTSPipelineBuilder.
 */
#include "vtkClientSession.h"
#include "vtkDataObject.h"
#include "vtkDistributedEnvironment.h"
#include "vtkLogger.h"
#include "vtkPTSPipelineBuilder.h"
#include "vtkPVApplication.h"
#include "vtkPVApplicationOptions.h"
#include "vtkSMProxy.h"
#include "vtkSMSessionProxyManager.h"
#include "vtkSMSourceProxy.h"
#include "vtkSmartPointer.h"
#include "vtkTestUtilities.h"

#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();
}

namespace Globals
{
std::string filename;
};

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

  int errorsCaught = 0;

  vtkNew<vtkPTSPipelineBuilder> pipelineBuilder;

  // if session is not set an error is raised
  pipelineBuilder->CreateSource("sources", "SphereSource")
    .subscribe([](vtkSMProxy* /*newProxy*/) { VALIDATE(false); },
      [&errorsCaught](rxcpp::util::error_ptr ep) {
        errorsCaught++;
        vtkLogF(INFO, "%s", rxcpp::util::what(ep).c_str());
      });

  pipelineBuilder->SetSession(session);

  // if the proxy does not exist an error is raised
  pipelineBuilder->CreateSource("sources", "123SphereSource123")
    .subscribe([](vtkSMProxy* /*newProxy*/) { VALIDATE(false); },
      [&errorsCaught](rxcpp::util::error_ptr ep) {
        errorsCaught++;
        vtkLogF(INFO, "%s", rxcpp::util::what(ep).c_str());
      });

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

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

  // Create a filter
  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);

  vtkSmartPointer<vtkSMSourceProxy> reader = nullptr;

  // Create a reader
  done = false;
  pipelineBuilder->CreateReader("sources", "IOSSReader", Globals::filename)
    .subscribe(
      [&reader, &done](vtkSMSourceProxy* newProxy) {
        VALIDATE(newProxy);
        reader = 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(reader != nullptr);

  // Attempt to delete a source with depedencies
  vtkTypeUInt32 id = source->GetGlobalID();
  done = false;
  pipelineBuilder->DeleteProxy(source).subscribe([&done](bool status) {
    VALIDATE(status == false);
    done = true;
  });
  ProcessEventsUntil(runLoop, [&done]() { return done; });

  auto proxyManager = session->GetProxyManager();
  VALIDATE(proxyManager->FindProxy(id) != nullptr);

  // Delete a proxy with no depedencies
  id = filter->GetGlobalID();

  done = false;
  pipelineBuilder->DeleteProxy(filter).subscribe([&done](bool status) {
    VALIDATE(status);
    done = true;
  });
  ProcessEventsUntil(runLoop, [&done]() { return done; });
  filter = nullptr;
  VALIDATE(proxyManager->FindProxy(id) == nullptr);

  VALIDATE(errorsCaught == 2);

  return true;
}

int TestPTSPipelineBuilder(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("TestPTSPipelineBuilder 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);
  }
  char* name = vtkTestUtilities::ExpandDataFileName(argc, argv, "Testing/Data/can.ex2");
  Globals::filename = name;
  delete[] name;

  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) { DoPTSPipelineBuilderTest(app->GetSession(id), runLoop); });
  app->WaitForExit(runLoop, std::chrono::seconds(1));
  app->Finalize();
  return app->GetExitCode();
}
