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

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

#include "vtkAbstractVideoEncoder.h"
#include "vtkBoundingBox.h"
#include "vtkCallbackCommand.h"
#include "vtkCamera.h"
#include "vtkChannelSubscription.h"
#include "vtkClientSession.h"
#include "vtkCodecTypes.h"
#include "vtkCodedVideoPacket.h"
#include "vtkCommand.h"
#include "vtkFFMPEGSoftwareDecoder.h"
#include "vtkGenericOpenGLRenderWindow.h"
#include "vtkImageData.h"
#include "vtkLogger.h"
#include "vtkObjectFactory.h"
#include "vtkOpenGLState.h"
#include "vtkPVCoreApplication.h"
#include "vtkPVCoreApplicationOptions.h"
#include "vtkPVRenderingCapabilitiesInformation.h"
#include "vtkPVXMLElement.h"
#include "vtkPointData.h"
#include "vtkRawVideoFrame.h"
#include "vtkRemoteObjectProviderViews.h"
#include "vtkRemotingCoreUtilities.h"
#include "vtkRenderWindow.h"
#include "vtkRenderWindowInteractor.h"
#include "vtkRenderer.h"
#include "vtkSMIntVectorProperty.h"
#include "vtkSMParaViewPipelineControllerWithRendering.h"
#include "vtkSMProxyIterator.h"
#include "vtkSMProxyProperty.h"
#include "vtkSMSessionProxyManager.h"
#include "vtkSMSourceProxy.h"
#include "vtkSMUncheckedPropertyHelper.h"
#include "vtkServiceEndpoint.h"
#include "vtkSmartPointer.h"
#include "vtkSynchronizedRenderers.h"
#include "vtkTextActor.h"
#include "vtkTextProperty.h"
#include "vtkTextRepresentation.h"
#include "vtkTimerLog.h"
#include "vtkUnsignedCharArray.h"
#include "vtkWEBMWriter.h"
#include "vtkVariant.h"

#include "vtk_glew.h"

#include <vtk_fmt.h> // needed for `fmt`
// clang-format off
#include VTK_FMT(fmt/core.h)
// clang-format on

#include <cstdint>
#include <sstream>

namespace vtkSMViewProxyNS
{
const char* GetRepresentationNameFromHints(const char* viewType, vtkPVXMLElement* hints, int port)
{
  if (!hints)
  {
    return nullptr;
  }

  for (unsigned int cc = 0, max = hints->GetNumberOfNestedElements(); cc < max; ++cc)
  {
    vtkPVXMLElement* child = hints->GetNestedElement(cc);
    if (child == nullptr || child->GetName() == nullptr)
    {
      continue;
    }

    // LEGACY: support DefaultRepresentations hint.
    // <Hints>
    //    <DefaultRepresentations representation="Foo" />
    // </Hints>
    if (strcmp(child->GetName(), "DefaultRepresentations") == 0)
    {
      return child->GetAttribute("representation");
    }

    // <Hints>
    //    <Representation port="outputPort" view="ViewName" type="ReprName" />
    // </Hints>
    else if (strcmp(child->GetName(), "Representation") == 0 &&
      // has an attribute "view" that matches the viewType.
      child->GetAttribute("view") && strcmp(child->GetAttribute("view"), viewType) == 0 &&
      child->GetAttribute("type") != nullptr)
    {
      // if port is present, it must match "port".
      int xmlPort;
      if (child->GetScalarAttribute("port", &xmlPort) == 0 || xmlPort == port)
      {
        return child->GetAttribute("type");
      }
    }
  }
  return nullptr;
}

} // namespace vtkSMViewProxyNS

class vtkSMViewProxy::vtkInternals
{
public:
  const std::thread::id OwnerTID{ std::this_thread::get_id() };
  vtkSMViewProxy* Self{ nullptr };
  vtkSmartPointer<vtkRenderWindow> RenderWindow;
  vtkSmartPointer<vtkRenderer> Renderer;
  unsigned long RenderWindowObserverId{ 0 };
  std::string DefaultRepresentationName;
  vtkBoundingBox VisiblePropBounds;
  vtkSmartPointer<vtkTextRepresentation> CornerAnnotation;
  vtkSmartPointer<vtkRenderer> Renderer2D;
  vtkNew<vtkFFMPEGSoftwareDecoder> Decoder;
  vtkNew<vtkWEBMWriter> WebmWriter;
  vtkNew<vtkRawVideoFrame> VideoFrame;

  enum EncodedPacketHandleModeEnum: uint8_t
  {
    DecodeToRenderWindow = 1 << 0,
    WriteWebmChunks = 1 << 1,
    DecodeToRenderWindowAndWriteWebmChunks = 1 << 2
  };

  int EncodedPacketHandleMode = EncodedPacketHandleModeEnum::DecodeToRenderWindow;

  void SetupInteractor();

  ~vtkInternals()
  {
    if (auto* iren = this->RenderWindow->GetInteractor())
    {
      for (const auto& id : this->InteractorObserverIds)
      {
        iren->RemoveObserver(id);
      }
    }

    if (this->RenderWindowObserverId)
    {
      this->RenderWindow->RemoveObserver(this->RenderWindowObserverId);
    }
  }

  void PushImageToWindow()
  {
    vtkLogScopeFunction(TRACE);
    if (this->VideoFrame->GetWidth() && this->VideoFrame->GetHeight() &&
      vtkRemoteObjectProviderViews::UsesCompressedStreams())
    {
      int size[2], low_point[2];
      this->Renderer->GetTiledSizeAndOrigin(&size[0], &size[1], &low_point[0], &low_point[1]);
      if (size[0] <= 0 || size[1] <= 0)
      {
        vtkGenericWarningMacro("Viewport empty. Cannot push to screen.");
        return;
      }

      vtkOpenGLRenderWindow* oglRenWin = vtkOpenGLRenderWindow::SafeDownCast(this->RenderWindow);
      vtkOpenGLState* ostate = oglRenWin->GetState();
      ostate->vtkglEnable(GL_SCISSOR_TEST);
      ostate->vtkglViewport(low_point[0], low_point[1], size[0], size[1]);
      ostate->vtkglScissor(low_point[0], low_point[1], size[0], size[1]);

      // framebuffers have their color premultiplied by alpha.
      vtkOpenGLState::ScopedglBlendFuncSeparate bfsaver(ostate);
      vtkOpenGLState::ScopedglEnableDisable bsaver(ostate, GL_BLEND);
      if (false)
      {
        vtkLogF(TRACE, "PushToFrameBuffer: using blend");
        ostate->vtkglEnable(GL_BLEND);
        ostate->vtkglBlendFuncSeparate(
          GL_ONE, GL_ONE_MINUS_SRC_ALPHA, GL_ONE, GL_ONE_MINUS_SRC_ALPHA);
      }
      else
      {
        ostate->vtkglDisable(GL_BLEND);
        vtkLogF(TRACE, "PushToFrameBuffer: not-using blend");
      }
      if (this->VideoFrame->GetWidth() == 0 || this->VideoFrame->GetHeight() == 0)
      {
        vtkLog(ERROR, "Size messed up");
        return;
      }
      vtkLogF(TRACE, "Args %d, %d, %d, %d, %d, %d, %d, %d, %d, %d", low_point[0], low_point[1],
        low_point[0] + size[0] - 1, low_point[1] + size[1] - 1, 0, 0,
        this->VideoFrame->GetWidth() - 1, this->VideoFrame->GetHeight() - 1,
        this->VideoFrame->GetWidth(), this->VideoFrame->GetHeight());
      this->Renderer->Clear();

      unsigned char* buffers[3];
      this->VideoFrame->GetData(buffers[0], 0);
      this->VideoFrame->GetData(buffers[1], 1);
      this->VideoFrame->GetData(buffers[2], 2);
      oglRenWin->DrawYCbCr420Pixels(this->VideoFrame->GetWidth(), this->VideoFrame->GetHeight(),
        buffers, this->VideoFrame->GetStrides());
    }
    else if (this->Image.IsValid() && !vtkRemoteObjectProviderViews::UsesCompressedStreams())
    {
      vtkLogF(TRACE, "PushToViewport");
      this->Image.PushToViewport(this->Renderer);
      auto data = vtkSmartPointer<vtkImageData>::New();
      data->GetPointData()->SetScalars(this->Image.GetRawPtr());
      data->SetDimensions(this->Image.GetWidth(), this->Image.GetHeight(), 1);
      this->ImageStream.get_subscriber().on_next(data);
    }
  }

  void PushImage(vtkSynchronizedRenderers::vtkRawImage&& image)
  {
    this->InRender = true;
    this->Image = image;
    this->RenderWindow->Render();
    this->InRender = false;
  }

  // TODO(jaswant): Replace vtkRawImage with a vtkVideoFrame that knows how to push itself into an
  // opengl render window.
  void PushVideoFrameToWindow(vtkRawVideoFrame* frame)
  {
    this->InRender = true;
    // TODO(jaswant): client (parat) calls render once to update camera, then pushes camera to the
    // render service.
    //                due to that render call, we arrive inside PushImageToWindow well before
    //                videoframe is ever set. consequently, a blank screen is displayed. this causes
    //                a flicker effect. we can fix it for now by making a full copy of the decoded
    //                frame and reusing it in the initial render for each interaction.

    // vtkRawVideoFrame does not implement DeepCopy. so let's do it manually by setting only what we
    // use in ::PushImageToWindow
    this->VideoFrame->SetWidth(frame->GetWidth());
    this->VideoFrame->SetHeight(frame->GetHeight());
    this->VideoFrame->SetPixelFormat(frame->GetPixelFormat());
    this->VideoFrame->SetStrides(frame->GetStrides(), VTK_RAW_VIDEO_FRAME_MAX_NUM_PLANES);
    for (int i = 0; i < VTK_RAW_VIDEO_FRAME_MAX_NUM_PLANES; ++i)
    {
      unsigned char* buffer = nullptr;
      int size = frame->GetData(buffer, i);
      if (buffer == nullptr)
      {
        continue;
      }
      this->VideoFrame->CopyData(buffer, size, i);
    }
    this->RenderWindow->Render();
    this->InRender = false;
  }

  void RequestRender() { this->UpdateCameraIfNeeded(/*force*/ true); }
  void RequestProgressiveRenderingPass()
  {
    this->UpdateCameraIfNeeded(/*force*/ true, /* progressiveRendering*/ true);
  }

  void AddMetadataAsAnnotations(const vtkNJson& metadata)
  {
    if (metadata.empty())
    {
      this->CornerAnnotation->SetVisibility(false);
    }
    else
    {
      std::ostringstream stream;
      if (metadata.contains("using-raytracing") && metadata.at("using-raytracing").get<bool>())
      {
        stream << fmt::format(
          "Rendering Pass {:3d}/{:3d}\n", this->RenderingPass, this->TotalRenderingPasses);
      }

      if (metadata.contains("timestamp"))
      {
        const double delta =
          vtkTimerLog::GetUniversalTime() - metadata.at("timestamp").get<double>();
        stream << fmt::format("Frame rate: {:0.3}\n", 1.0 / delta);
      }

      this->CornerAnnotation->SetVisibility(true);
      this->CornerAnnotation->SetText(stream.str().c_str());
    }
  }

  rxcpp::subjects::subject<vtkSmartPointer<vtkImageData>> ImageStream;
  rxcpp::subjects::subject<vtkSmartPointer<vtkUnsignedCharArray>> WebmChunkStream;

  void UpdateRayTracingState(const vtkNJson& metadata);
  bool GetContinueRendering() const;

private:
  std::vector<unsigned long> InteractorObserverIds;
  vtkMTimeType LastCameraMTime{ 0 };
  // keep track of the raytracing information so we can request for the next rendering pass
  //@{
  bool UsingRayTracing{ false };
  uint64_t RenderingPass{ 0 };
  uint64_t TotalRenderingPasses{ 0 };
  //@}
  bool InRender{ false };
  vtkSynchronizedRenderers::vtkRawImage Image;

  void HandleInteractorEvent(vtkObject* /*caller*/, unsigned long eventId, void* /*calldata*/)
  {
    switch (eventId)
    {
      case vtkCommand::RenderEvent:
        if (!this->InRender)
        {
          this->UpdateCameraIfNeeded();
        }
        break;

      case vtkCommand::WindowResizeEvent:
      case vtkCommand::ConfigureEvent:
        this->UpdateSize();
        break;
    }
  }

  void UpdateCameraIfNeeded(bool force = false, bool progressiveRendering = false)
  {
    auto* camera = this->Renderer->GetActiveCamera();
    if (force || progressiveRendering || this->LastCameraMTime != camera->GetMTime())
    {
      this->RenderingPass = progressiveRendering ? this->RenderingPass + 1 : 0;
      vtkLogF(TRACE, "push camera");
      this->LastCameraMTime = camera->GetMTime();
      auto packet = vtkRemoteObjectProviderViews::Render(this->Self->GetGlobalID(), camera);
      this->Self->GetSession()->SendMessage(vtkClientSession::RENDER_SERVER, std::move(packet));
    }
  }

  void UpdateSize()
  {
    auto& window = this->RenderWindow;
    vtkSMPropertyHelper(this->Self, "ViewSize").Set(window->GetSize(), 2);
    vtkLogF(TRACE, "UpdateSize(%d,%d)", window->GetSize()[0], window->GetSize()[1]);
    this->Self->UpdateVTKObjects();
    this->UpdateCameraIfNeeded(true);
  }
};

//----------------------------------------------------------------------------
void vtkSMViewProxy::vtkInternals::SetupInteractor()
{
  auto* iren = this->RenderWindow->GetInteractor();
  if (!iren)
  {
    return;
  }

  this->Self->InitializeInteractor(iren);

  this->InteractorObserverIds.push_back(iren->AddObserver(
    vtkCommand::StartInteractionEvent, this, &vtkInternals::HandleInteractorEvent));
  this->InteractorObserverIds.push_back(
    iren->AddObserver(vtkCommand::EndInteractionEvent, this, &vtkInternals::HandleInteractorEvent));
  this->InteractorObserverIds.push_back(
    iren->AddObserver(vtkCommand::RenderEvent, this, &vtkInternals::HandleInteractorEvent));
  this->InteractorObserverIds.push_back(
    iren->AddObserver(vtkCommand::WindowResizeEvent, this, &vtkInternals::HandleInteractorEvent));

  // XOpenGLRenderWindow fires ConfigureEvent when window resizes.
  this->InteractorObserverIds.push_back(
    iren->AddObserver(vtkCommand::ConfigureEvent, this, &vtkInternals::HandleInteractorEvent));

  // Remove the observer so we don't end up calling this method again.
  this->RenderWindow->RemoveObserver(this->RenderWindowObserverId);
  this->RenderWindowObserverId = 0;
}
//----------------------------------------------------------------------------
void vtkSMViewProxy::vtkInternals::UpdateRayTracingState(const vtkNJson& metadata)
{
  if (metadata.empty())
  {
    return;
  }
  if (metadata.contains("using-raytracing"))
  {
    this->UsingRayTracing = metadata.at("using-raytracing").get<bool>();
  }

  if (metadata.contains("total-rendering-passes"))
  {
    this->TotalRenderingPasses = metadata.at("total-rendering-passes").get<int>();
  }
}

//----------------------------------------------------------------------------
bool vtkSMViewProxy::vtkInternals::GetContinueRendering() const
{
  if (!this->UsingRayTracing)
  {
    return false;
  }
  if (this->RenderingPass < this->TotalRenderingPasses)
  {
    return true;
  }
  return false;
}
//============================================================================
vtkStandardNewMacro(vtkSMViewProxy);
bool vtkSMViewProxy::UseGenericOpenGLRenderWindow = false;
//----------------------------------------------------------------------------
vtkSMViewProxy::vtkSMViewProxy()
  : Internals(new vtkSMViewProxy::vtkInternals())
{
  auto& internals = (*this->Internals);
  internals.Self = this;

  auto* options = vtkPVCoreApplication::GetInstance()->GetOptions();
  if (vtkSMViewProxy::GetUseGenericOpenGLRenderWindow())
  {
    internals.RenderWindow = vtk::TakeSmartPointer(vtkGenericOpenGLRenderWindow::New());
  }
  else if (options->GetForceOffscreenRendering())
  {
    internals.RenderWindow = vtkPVRenderingCapabilitiesInformation::NewOffscreenRenderWindow();
  }
  else
  {
    internals.RenderWindow = vtk::TakeSmartPointer(vtkRenderWindow::New());
  }

  internals.Renderer = vtk::TakeSmartPointer(vtkRenderer::New());
  internals.Renderer->SetBackground(0, 0, 0); // must be black.
  internals.RenderWindow->AddRenderer(internals.Renderer);
  internals.RenderWindow->SetNumberOfLayers(2);

  internals.Renderer2D = vtk::TakeSmartPointer(vtkRenderer::New());
  internals.Renderer2D->SetLayer(1);
  internals.Renderer2D->EraseOff();
  internals.Renderer2D->InteractiveOff();
  internals.RenderWindow->AddRenderer(internals.Renderer2D);

  // Setup Corner annotation for rendering passes
  internals.CornerAnnotation = vtk::TakeSmartPointer(vtkTextRepresentation::New());
  internals.Renderer2D->AddActor(internals.CornerAnnotation);
  internals.CornerAnnotation->SetText("Annotation");
  internals.CornerAnnotation->SetRenderer(internals.Renderer2D);
  internals.CornerAnnotation->BuildRepresentation();
  internals.CornerAnnotation->SetWindowLocation(vtkTextRepresentation::UpperLeftCorner);
  internals.CornerAnnotation->GetTextActor()->SetTextScaleModeToNone();
  internals.CornerAnnotation->GetTextActor()->GetTextProperty()->SetJustificationToLeft();
  internals.CornerAnnotation->SetVisibility(false);

  vtkNew<vtkCallbackCommand> onEmitFrameEventCb;
  onEmitFrameEventCb->SetClientData(&internals);
  onEmitFrameEventCb->SetCallback(
    [](vtkObject*, unsigned long, void* clientData, void* frame) {
      auto internals = reinterpret_cast<vtkSMViewProxy::vtkInternals*>(clientData);
      internals->PushVideoFrameToWindow(reinterpret_cast<vtkRawVideoFrame*>(frame));
    });
  internals.Decoder->SetCodecType(VTKCodecType::VP9);
  internals.Decoder->AddObserver(vtkCommand::ProgressEvent, onEmitFrameEventCb);
  internals.WebmWriter->SetWriteToMemory(true);
  internals.WebmWriter->SetLiveMode(true);
  internals.WebmWriter->SetForceNewClusters(true);
  internals.WebmWriter->SetOutputCues(false);

  // monitor render window so we can setup observers on interactor
  // for events that send render requests to the server.
  internals.RenderWindowObserverId = internals.RenderWindow->AddObserver(
    vtkCommand::ModifiedEvent, &internals, &vtkSMViewProxy::vtkInternals::SetupInteractor);
  internals.Renderer->AddObserver(
    vtkCommand::EndEvent, &internals, &vtkSMViewProxy::vtkInternals::PushImageToWindow);
}

//----------------------------------------------------------------------------
vtkSMViewProxy::~vtkSMViewProxy()
{
  this->UnsubscribeImageStreams();
  this->UnsubscribeWebmChunkStreams();
}

//----------------------------------------------------------------------------
vtkRenderWindow* vtkSMViewProxy::GetRenderWindow() const
{
  const auto& internals = (*this->Internals);
  return internals.RenderWindow;
}

//----------------------------------------------------------------------------
vtkRenderWindowInteractor* vtkSMViewProxy::GetRenderWindowInteractor() const
{
  const auto& internals = (*this->Internals);
  return internals.RenderWindow ? internals.RenderWindow->GetInteractor() : nullptr;
}

//----------------------------------------------------------------------------
vtkCamera* vtkSMViewProxy::GetCamera() const
{
  const auto& internals = (*this->Internals);
  return internals.Renderer->GetActiveCamera();
}

//----------------------------------------------------------------------------
void vtkSMViewProxy::InitializeInteractor(vtkRenderWindowInteractor* vtkNotUsed(iren)) {}

//----------------------------------------------------------------------------
void vtkSMViewProxy::RequestProgressiveRenderingPassIfNeeded()
{

  auto& internals = (*this->Internals);
  if (internals.GetContinueRendering())
  {
    internals.RequestProgressiveRenderingPass();
  }
}

//----------------------------------------------------------------------------
void vtkSMViewProxy::PostRenderingResult(const vtkPacket& packet)
{
  auto& internals = (*this->Internals);
  vtkRemotingCoreUtilities::EnsureThread(internals.OwnerTID);
  if (vtkRemoteObjectProviderViews::UsesCompressedStreams())
  {
    vtkLogF(TRACE, "got packet");
    const auto& json = packet.GetJSON();
    auto array = packet.GetPayload<vtkUnsignedCharArray>("packet");
    assert(array);
    vtkNew<vtkCodedVideoPacket> codedPkt;
    codedPkt->GetData()->DeepCopy(array);
    codedPkt->SetWidth(json["width"]);
    codedPkt->SetHeight(json["height"]);
    codedPkt->SetIsKeyFrame(json["keyframe"]);
    codedPkt->SetPresentationTS(json["pts"]);
    if (internals.EncodedPacketHandleMode & vtkInternals::DecodeToRenderWindow)
    {
      internals.Decoder->Push(codedPkt);
    }
    if (internals.EncodedPacketHandleMode & vtkInternals::WriteWebmChunks)
    {
      if (codedPkt->GetIsKeyFrame())
      {
        if (codedPkt->GetPresentationTS() > 1)
        {
          internals.WebmWriter->WriteFileTrailer();
        }
      }
      if (!internals.WebmWriter->GetIsHeaderWritten())
      {
        internals.WebmWriter->SetWidth(codedPkt->GetWidth());
        internals.WebmWriter->SetHeight(codedPkt->GetHeight());
        internals.WebmWriter->SetFramerate(30);
        internals.WebmWriter->WriteVP9FileHeader();
      }
      internals.WebmWriter->WriteWebmBlock(codedPkt);
      auto webmChunk = vtk::TakeSmartPointer(vtkUnsignedCharArray::New());
      webmChunk->DeepCopy(internals.WebmWriter->GetDynamicBuffer());
      internals.WebmChunkStream.get_subscriber().on_next(webmChunk);
      internals.WebmWriter->Flush();
    }
  }
  else
  {
    vtkLogF(TRACE, "got image");
    const auto& json = packet.GetJSON();
    auto array = packet.GetPayload<vtkUnsignedCharArray>("image");
    assert(array);
    vtkSynchronizedRenderers::vtkRawImage image;
    image.Initialize(json.at("width").get<int>(), json.at("height").get<int>(), array);
    image.MarkValid();

    if (json.contains("metadata"))
    {
      auto& metadata = json["metadata"];
      internals.UpdateRayTracingState(metadata);
      internals.AddMetadataAsAnnotations(metadata);
    }
    else
    {
      internals.CornerAnnotation->SetVisibility(false);
    }
    internals.PushImage(std::move(image));
  }
}

//----------------------------------------------------------------------------
void vtkSMViewProxy::SetUseGenericOpenGLRenderWindow(bool val)
{
  vtkSMViewProxy::UseGenericOpenGLRenderWindow = val;
}

//----------------------------------------------------------------------------
bool vtkSMViewProxy::GetUseGenericOpenGLRenderWindow()
{
  return vtkSMViewProxy::UseGenericOpenGLRenderWindow;
}

//----------------------------------------------------------------------------
void vtkSMViewProxy::UseCompressedStreamsOn()
{
  vtkRemoteObjectProviderViews::UseCompressedStreamsOn();
}

//----------------------------------------------------------------------------
void vtkSMViewProxy::UseCompressedStreamsOff()
{
  vtkRemoteObjectProviderViews::UseCompressedStreamsOff();
}

//----------------------------------------------------------------------------
void vtkSMViewProxy::DecodeStreamToRenderWindowOnly()
{
  auto& internals = (*this->Internals);
  internals.EncodedPacketHandleMode = vtkInternals::EncodedPacketHandleModeEnum::DecodeToRenderWindow;
}

//----------------------------------------------------------------------------
void vtkSMViewProxy::WriteWebmChunksOnly()
{
  auto& internals = (*this->Internals);
  internals.EncodedPacketHandleMode = vtkInternals::EncodedPacketHandleModeEnum::WriteWebmChunks;
}

//----------------------------------------------------------------------------
void vtkSMViewProxy::DecodeStreamToRenderWindowAndWriteWebmChunks()
{
  auto& internals = (*this->Internals);
  internals.EncodedPacketHandleMode |= vtkInternals::EncodedPacketHandleModeEnum::DecodeToRenderWindow;
  internals.EncodedPacketHandleMode |= vtkInternals::EncodedPacketHandleModeEnum::WriteWebmChunks;
}

//----------------------------------------------------------------------------
rxcpp::observable<bool> vtkSMViewProxy::Update()
{
  auto& internals = (*this->Internals);
  vtkRemotingCoreUtilities::EnsureThread(internals.OwnerTID);

  using BehaviorT = rxcpp::subjects::behavior<vtkVariant>;
  auto behavior = std::make_shared<BehaviorT>(vtkVariant());
  this->ExecuteCommand("Update", vtkClientSession::DATA_SERVER).subscribe([behavior, this](bool) {
    this->ExecuteCommand("UpdateForRendering", vtkClientSession::RENDER_SERVER)
      .subscribe([=](bool status) {
        behavior->get_subscriber().on_next(vtkVariant(status ? 1 : 0));
        this->StillRender();
      });
  });

  return behavior->get_observable()
    .filter([](const vtkVariant& variant) { return variant.IsValid(); })
    .take(1)
    .map([behavior](const vtkVariant& variant) {
      // capture behavior to ensure it hangs around until the returned value hangs around.
      (void)behavior;
      return variant.ToInt() != 0;
    });
}

//----------------------------------------------------------------------------
void vtkSMViewProxy::ResetCameraUsingVisiblePropBounds()
{
  auto& internals = (*this->Internals);
  vtkRemotingCoreUtilities::EnsureThread(internals.OwnerTID);
  if (internals.VisiblePropBounds.IsValid())
  {
    double bds[6];
    internals.VisiblePropBounds.GetBounds(bds);
    internals.Renderer->ResetCamera(bds);

    // If 'CenterOfRotation' is present, set it to the center of bounds as well.
    if (auto* centerProperty = this->GetProperty("CenterOfRotation"))
    {
      double center[3];
      internals.VisiblePropBounds.GetCenter(center);
      vtkSMPropertyHelper(centerProperty).Set(center, 3);
    }
  }
}

//----------------------------------------------------------------------------
void vtkSMViewProxy::StillRender()
{
  auto& internals = (*this->Internals);
  vtkRemotingCoreUtilities::EnsureThread(internals.OwnerTID);
  internals.RequestRender();
}

//----------------------------------------------------------------------------
const vtkBoundingBox& vtkSMViewProxy::GetVisiblePropBounds() const
{
  return this->Internals->VisiblePropBounds;
}

//----------------------------------------------------------------------------
bool vtkSMViewProxy::HideOtherRepresentationsIfNeeded(vtkSMProxy* repr)
{
  if (repr == nullptr || this->GetHints() == nullptr ||
    this->GetHints()->FindNestedElementByName("ShowOneRepresentationAtATime") == nullptr)
  {
    return false;
  }

  vtkPVXMLElement* oneRepr =
    this->GetHints()->FindNestedElementByName("ShowOneRepresentationAtATime");
  const char* reprType = oneRepr->GetAttribute("type");

  if (reprType && strcmp(repr->GetXMLName(), reprType) != 0)
  {
    return false;
  }

  vtkNew<vtkSMParaViewPipelineControllerWithRendering> controller;

  bool modified = false;
  vtkSMPropertyHelper helper(this, "Representations");
  for (unsigned int cc = 0, max = helper.GetNumberOfElements(); cc < max; ++cc)
  {
    auto* arepr = helper.GetAsProxy(cc);
    if (arepr && arepr != repr)
    {
      if (vtkSMPropertyHelper(arepr, "Visibility", /*quiet*/ true).GetAsInt() == 1 &&
        (!reprType || (reprType && !strcmp(arepr->GetXMLName(), reprType))))
      {
        controller->Hide(arepr, this);
        modified = true;
      }
    }
  }
  return modified;
}

//----------------------------------------------------------------------------
void vtkSMViewProxy::RepresentationVisibilityChanged(vtkSMProxy*, bool) {}

//----------------------------------------------------------------------------
int vtkSMViewProxy::ReadXMLAttributes(vtkSMSessionProxyManager* pm, vtkPVXMLElement* element)
{
  if (!this->Superclass::ReadXMLAttributes(pm, element))
  {
    return 0;
  }

  auto& internals = (*this->Internals);
  const char* repr_name = element->GetAttribute("representation_name");
  if (repr_name)
  {
    internals.DefaultRepresentationName = repr_name;
  }
  return 1;
}

//----------------------------------------------------------------------------
vtkSMProxy* vtkSMViewProxy::CreateDefaultRepresentation(vtkSMProxy* proxy, int outputPort)
{
  vtkSMSourceProxy* producer = vtkSMSourceProxy::SafeDownCast(proxy);
  if ((producer == nullptr) || (outputPort < 0) ||
    (static_cast<int>(producer->GetNumberOfOutputPorts()) <= outputPort) ||
    (producer->GetSession() != this->GetSession()))
  {
    return nullptr;
  }

  // Update with time from the view to ensure we have up-to-date data.
  double view_time = vtkSMPropertyHelper(this, "ViewTime").GetAsDouble();
  producer->UpdatePipeline(view_time);

  const char* representationType = this->GetRepresentationType(producer, outputPort);
  if (!representationType)
  {
    return nullptr;
  }

  vtkSMSessionProxyManager* pxm = this->GetSessionProxyManager();
  if (auto* repr = pxm->NewProxy("representations", representationType))
  {
    return repr;
  }
  vtkWarningMacro(
    "Failed to create representation (representations," << representationType << ").");
  return nullptr;
}

//----------------------------------------------------------------------------
const char* vtkSMViewProxy::GetRepresentationType(vtkSMSourceProxy* producer, int outputPort)
{
  assert(producer && static_cast<int>(producer->GetNumberOfOutputPorts()) > outputPort);

  // Process producer hints to see if indicates what type of representation
  // to create for this view.
  if (const char* reprName = vtkSMViewProxyNS::GetRepresentationNameFromHints(
        this->GetXMLName(), producer->GetHints(), outputPort))
  {
    return reprName;
  }

  const auto& internals = (*this->Internals);
  // check if we have default representation name specified in XML.
  if (!internals.DefaultRepresentationName.empty())
  {
    vtkSMSessionProxyManager* pxm = this->GetSessionProxyManager();
    vtkSMProxy* prototype =
      pxm->GetPrototypeProxy("representations", internals.DefaultRepresentationName.c_str());
    if (prototype)
    {
      vtkSMProperty* inputProp = prototype->GetProperty("Input");
      vtkSMUncheckedPropertyHelper helper(inputProp);
      helper.Set(producer, outputPort);
      bool acceptable = (inputProp->IsInDomains() > 0);
      helper.SetNumberOfElements(0);

      if (acceptable)
      {
        return internals.DefaultRepresentationName.c_str();
      }
    }
  }

  return nullptr;
}

//----------------------------------------------------------------------------
bool vtkSMViewProxy::CanDisplayData(vtkSMSourceProxy* producer, int outputPort)
{
  if (producer == nullptr || outputPort < 0 ||
    static_cast<int>(producer->GetNumberOfOutputPorts()) <= outputPort ||
    producer->GetSession() != this->GetSession())
  {
    return false;
  }

  const char* type = this->GetRepresentationType(producer, outputPort);
  if (type != nullptr)
  {
    vtkSMSessionProxyManager* pxm = this->GetSessionProxyManager();
    return (pxm->GetPrototypeProxy("representations", type) != nullptr);
  }

  return false;
}

//----------------------------------------------------------------------------
vtkSMProxy* vtkSMViewProxy::FindRepresentation(vtkSMSourceProxy* producer, int outputPort)
{
  vtkSMPropertyHelper helper(this, "Representations");
  for (unsigned int cc = 0, max = helper.GetNumberOfElements(); cc < max; ++cc)
  {
    auto* repr = helper.GetAsProxy(cc);
    if (repr && repr->GetProperty("Input"))
    {
      vtkSMPropertyHelper helper2(repr, "Input");
      if (helper2.GetAsProxy() == producer &&
        static_cast<int>(helper2.GetOutputPort()) == outputPort)
      {
        return repr;
      }
    }
  }

  return nullptr;
}

//----------------------------------------------------------------------------
vtkSMViewProxy* vtkSMViewProxy::FindView(vtkSMProxy* repr, const char* reggroup /*=views*/)
{
  if (!repr)
  {
    return nullptr;
  }

  vtkSMSessionProxyManager* pxm = repr->GetSessionProxyManager();
  vtkNew<vtkSMProxyIterator> iter;
  iter->SetSessionProxyManager(pxm);
  iter->SetModeToOneGroup();
  for (iter->Begin(reggroup); !iter->IsAtEnd(); iter->Next())
  {
    if (vtkSMViewProxy* view = vtkSMViewProxy::SafeDownCast(iter->GetProxy()))
    {
      auto reprs = vtkSMProxyProperty::SafeDownCast(view->GetProperty("Representations"));
      auto hreprs = vtkSMProxyProperty::SafeDownCast(view->GetProperty("HiddenRepresentations"));
      if ((reprs != nullptr && reprs->IsProxyAdded(repr)) ||
        (hreprs != nullptr && hreprs->IsProxyAdded(repr)))
      {
        return view;
      }
    }
  }
  return nullptr;
}
//----------------------------------------------------------------------------
rxcpp::observable<vtkSmartPointer<vtkImageData>> vtkSMViewProxy::GetImageObservable()
{
  return this->Internals->ImageStream.get_observable().tap(
    [](auto image) { vtkLogF(INFO, "push image");});
}

//----------------------------------------------------------------------------
rxcpp::observable<vtkSmartPointer<vtkUnsignedCharArray>> vtkSMViewProxy::GetWEBMChunkObservable() const
{
  return this->Internals->WebmChunkStream.get_observable().tap(
    [](auto image) { vtkLogF(INFO, "push webm chunk");});
}

//----------------------------------------------------------------------------
void vtkSMViewProxy::UnsubscribeImageStreams()
{
  this->Internals->ImageStream.get_subscriber().on_completed();
}

//----------------------------------------------------------------------------
void vtkSMViewProxy::UnsubscribeWebmChunkStreams()
{
  vtkLogScopeFunction(TRACE);
  this->Internals->WebmChunkStream.get_subscriber().on_completed();
  this->Internals->WebmWriter->WriteFileTrailer();
}

//----------------------------------------------------------------------------
void vtkSMViewProxy::ProcessPiggybackedInformation(const vtkNJson& json)
{
  this->Superclass::ProcessPiggybackedInformation(json);

  auto iter = json.find("VisiblePropBounds");
  if (iter != json.end())
  {
    std::vector<double> bounds = iter.value().get<std::vector<double>>();
    this->Internals->VisiblePropBounds.SetBounds(&bounds[0]);
  }
}

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