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

  Copyright (c) Kitware, Inc.
  All rights reserved.

     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 "mvCollaborationClient.h"

#include "vtkCallbackCommand.h"
#include "vtkCamera.h"
#include "vtkEventData.h"
#include "vtkOpenGLAvatar.h"
#include "vtkOpenVRModel.h"
#include "vtkOpenVRRenderer.h"
#include "vtkOpenVRRenderWindow.h"
#include "vtkOpenVRRenderWindowInteractor.h"
#include "vtkProperty.h"
#include "vtkTextProperty.h"
#include "vtkTimerLog.h"
#include "vtkTransform.h"
#include "zhelpers.hpp"

#include <sstream>

namespace
{
const int LPOS = 4;
const int LORIENT = 3;
const int LTRANSLATION = 3;
const int LDIRECTION = 3;
const double AVATAR_TIMEOUT = 10.0; // in seconds
const int HEARTBEAT_INTERVAL = 1.0; // in seconds
const int LIVE_COUNT = 3;

// http://colorbrewer2.org/#type=qualitative&scheme=Pastel1&n=9
double AVATAR_COLORS[][3] = {
  { 179 / 255.0, 205 / 255.0, 227 / 255.0 },
  { 204 / 255.0, 235 / 255.0, 197 / 255.0 },
  { 222 / 255.0, 203 / 255.0, 228 / 255.0 },
  { 254 / 255.0, 217 / 255.0, 166 / 255.0 },
  { 255 / 255.0, 255 / 255.0, 204 / 255.0 },
  { 229 / 255.0, 216 / 255.0, 189 / 255.0 },
  { 253 / 255.0, 218 / 255.0, 236 / 255.0 },
  { 242 / 255.0, 242 / 255.0, 242 / 255.0 },
  { 251 / 255.0, 180 / 255.0, 174 / 255.0 },
};

const int NUM_COLORS = sizeof(AVATAR_COLORS) / sizeof(AVATAR_COLORS[0]);
}

#define mvLog(x) \
  if (this->Callback) \
  { \
    std::ostringstream ss; ss << x; \
    this->Callback(ss.str(), this->CallbackClientData); \
  } \
  else { std::cout << x; }

mvCollaborationClient::mvCollaborationClient() :
  Context(1),
  Connected(false),
  MoveObserver(-1),
  Callback(nullptr),
  CallbackClientData(nullptr),
  Requester(Context, ZMQ_DEALER),
  Subscriber(Context, ZMQ_SUB),
  CollabPollItems{
    { this->Requester, 0, ZMQ_POLLIN, 0 },
    { this->Subscriber, 0, ZMQ_POLLIN, 0 }
  }
{
  this->CollabPort = 5555;
  // Position MineView Zeromq, default when none is specified.
  this->CollabSession = "PMVZ";
  this->RetryCount = 1;  // start in retry state.
  this->NeedHeartbeat = 0;
  this->NeedReply = 0;
  this->PublishAvailable = false; // publish socket not sending yet.

  this->EventCommand = vtkCallbackCommand::New();
  this->EventCommand->SetClientData(this);
  this->EventCommand->SetCallback(
    mvCollaborationClient::EventCallback);

  // setup default scale callback
  this->ScaleCallback = [this](void *) {
    auto ovrrw = vtkOpenVRRenderWindow::SafeDownCast(this->RenderWindow);
    return ovrrw ? ovrrw->GetPhysicalScale() : 1.0;
  };
}

mvCollaborationClient::~mvCollaborationClient()
{
  this->Disconnect();
  this->EventCommand->Delete();
}

void mvCollaborationClient::Disconnect()
{
  if (!this->Connected)
  {
    return;
  }

  mvLog("Collab server disconnecting. " << std::endl);

  if (this->Requester.connected())
  {
    this->Requester.close();
  }
  if (this->Subscriber.connected())
  {
    this->Subscriber.close();
  }
  for (auto it : this->AvatarUpdateTime)
  {
    this->Renderer->RemoveActor(this->Avatars[it.first]);
    this->Avatars.erase(it.first);
  }
  this->AvatarUpdateTime.clear();

  if (this->MoveObserver >= 0 &&
      this->RenderWindow && this->RenderWindow->GetInteractor())
  {
    this->RenderWindow->GetInteractor()->RemoveObserver(this->MoveObserver);
    this->MoveObserver = -1;
  }
  this->Connected = false;
}

void mvCollaborationClient::AddArguments(vtksys::CommandLineArguments &arguments)
{
  typedef vtksys::CommandLineArguments argT;

  arguments.AddArgument("--collab-server", argT::EQUAL_ARGUMENT, &(this->CollabHost),
    "(optional) Connect to collaboration server at this hostname");
  arguments.AddArgument("--collab-port", argT::EQUAL_ARGUMENT, &(this->CollabPort),
    "(default:5555) Connect to collaboration server at this port");
  arguments.AddArgument("--collab-session", argT::EQUAL_ARGUMENT, &(this->CollabSession),
    "Connect to a separate collaboration session - each collaborator should use a matching value");
  arguments.AddArgument("--collab-name", argT::EQUAL_ARGUMENT, &(this->CollabName),
    "Name to display over your avatar to other collaborators");
  this->DisplayOwnAvatar = false;
  arguments.AddBooleanArgument("--show-my-avatar", &this->DisplayOwnAvatar,
    "(default false) Show an avatar at my own position.");
}

void mvCollaborationClient::Render()
{
  if (this->Connected)
  {
    this->HandleCollabMessage();
    this->EraseIdleAvatars();
  }
}

template <typename T>
void mvCollaborationClient::SendMessage(std::string const &msgType, std::vector<T> const &vals)
{
  if (this->CollabID.empty())
  {
    return;
  }

  // send header, our ID, session.
  s_sendmore(this->Requester, "PMVZ");
  s_sendmore(this->Requester, this->CollabID);
  s_sendmore(this->Requester, this->CollabSession);
  s_sendmore(this->Requester, msgType);

  unsigned int count = static_cast<unsigned int>(vals.size());
  zmq::message_t countMsg (sizeof(count));
  memcpy (countMsg.data(), &count, sizeof(count));
  this->Requester.send(countMsg, ZMQ_SNDMORE);

  zmq::message_t request (sizeof(T) * vals.size());
  memcpy (request.data(), &vals[0], sizeof(T) * vals.size());
  this->Requester.send(request);
}
template void mvCollaborationClient::SendMessage<double>(std::string const &msgType, std::vector<double> const &vals);

void mvCollaborationClient::SendDevicePoseMessage(vtkEventDataDevice device, double pos[], double orient[])
{
  // Note: size of arrays not maintained when passed as params.

  // don't send a message if we haven't gotten one during the last
  // heartbeat. View messages, however, are always sent (queued).
  if (this->RetryCount > 0)
  {
    return;
  }
  // if (within(100) == 1) {
  //   std::cout << "Send: [" << "X" << ", " <<
  //       static_cast<int>(device) << "] " <<
  //       pos[0] << ", " <<
  //       pos[1] << ", " <<
  //       pos[2] << " " <<
  //       std::endl;
  //   std::cout << "      orient: " <<
  //       orient[0] << ", " <<
  //       orient[1] << ", " <<
  //       orient[2] << " " <<
  //       std::endl;
  // }
  // Don't send hand messages if the head is idle
  if (device != vtkEventDataDevice::HeadMountedDisplay && this->AvatarIdle(this->CollabID))
  {
    return;
  }
  // send header, our ID, session.
  s_sendmore(this->Requester, "PMVZ");
  s_sendmore(this->Requester, this->CollabID);
  s_sendmore(this->Requester, this->CollabSession);
  // send avatar change type.
  s_sendmore(this->Requester, "A");
  zmq::message_t devMsg (sizeof(device));
  memcpy (devMsg.data(), &device, sizeof(device));
  this->Requester.send(devMsg, ZMQ_SNDMORE);
  zmq::message_t request (sizeof(pos[0]) * LPOS);
  memcpy (request.data(), &pos[0], sizeof(pos[0]) * LPOS);
  // std::cout << "Sending device " << static_cast<int>(device) << " ..." << std::endl;
  this->Requester.send(request, ZMQ_SNDMORE);

  zmq::message_t request2 (sizeof(orient[0]) * LORIENT);
  memcpy (request2.data(), &orient[0], sizeof(orient[0]) * LORIENT);
  this->Requester.send(request2);
}

void mvCollaborationClient::SendMessage(
  std::string const &msgType,
  int index,
  double translation[],
  double direction[])
{
  // Note: size of arrays not maintained when passed as params.

  // don't send a message if we haven't gotten one during the last
  // heartbeat. View messages, however, are always sent (queued).
  if (this->RetryCount > 0)
  {
    return;
  }

  // send header, our ID, session.
  s_sendmore(this->Requester, "PMVZ");
  s_sendmore(this->Requester, this->CollabID);
  s_sendmore(this->Requester, this->CollabSession);
  // send avatar change type.
  s_sendmore(this->Requester, msgType);
  zmq::message_t indexMsg (sizeof(index));
  memcpy (indexMsg.data(), &index, sizeof(index));
  this->Requester.send(indexMsg, ZMQ_SNDMORE);
  zmq::message_t request (sizeof(translation[0]) * LTRANSLATION);
  memcpy (request.data(), &translation[0], sizeof(translation[0]) * LTRANSLATION);
  this->Requester.send(request, ZMQ_SNDMORE);

  zmq::message_t request2 (sizeof(direction[0]) * LDIRECTION);
  memcpy (request2.data(), &direction[0], sizeof(direction[0]) * LDIRECTION);
  this->Requester.send(request2);
}

void mvCollaborationClient::SendMessage(std::string const &msgType)
{
  if (this->CollabID.empty())
  {
    return;
  }
  // send header, our ID, session.
  s_sendmore(this->Requester, "PMVZ");
  s_sendmore(this->Requester, this->CollabID);
  s_sendmore(this->Requester, this->CollabSession);
  // send view change type.
  s_send(this->Requester, msgType);
}

void mvCollaborationClient::SendMessage(std::string const &msgType, int index)
{
  if (this->CollabID.empty())
  {
    return;
  }
  // send header, our ID, session.
  s_sendmore(this->Requester, "PMVZ");
  s_sendmore(this->Requester, this->CollabID);
  s_sendmore(this->Requester, this->CollabSession);
  // send view change type.
  s_sendmore(this->Requester, msgType);
  zmq::message_t indexMsg (sizeof(index));
  memcpy (indexMsg.data(), &index, sizeof(index));
  this->Requester.send(indexMsg);

  //  Get the reply.
  // std::string ack = s_recv(this->Requester);
}

void mvCollaborationClient::SendMessage(
  std::string const &msgType,
  std::string const &val)
{
  if (this->CollabID.empty())
  {
    return;
  }
  // send header, our ID, session.
  s_sendmore(this->Requester, "PMVZ");
  s_sendmore(this->Requester, this->CollabID);
  s_sendmore(this->Requester, this->CollabSession);
  // send view change type.
  s_sendmore(this->Requester, msgType);
  s_send(this->Requester, val);
}

void mvCollaborationClient::SendMessage(
  std::string const &msgType,
  std::vector<std::string> const& vals)
{
  if (this->CollabID.empty())
  {
    return;
  }
  // send header, our ID, session.
  s_sendmore(this->Requester, "PMVZ");
  s_sendmore(this->Requester, this->CollabID);
  s_sendmore(this->Requester, this->CollabSession);
  s_sendmore(this->Requester, msgType);
  for (auto &val: vals)
  {
    s_sendmore(this->Requester, val);
  }
  s_send(this->Requester, "");
}

void mvCollaborationClient::HandleBroadcastMessage(
  std::string const &otherID,
  std::string const &type)
{
  if (type == "A")
  {
    zmq::message_t update;
    double updatePos[LPOS] = { 0 };
    double updateOrient[LORIENT] = { 0 };
    vtkEventDataDevice device;

    this->Subscriber.recv(&update);
    memcpy (&device, update.data(), sizeof(device));
    this->Subscriber.recv(&update);
    memcpy (&updatePos[0], update.data(), sizeof(updatePos));
    this->Subscriber.recv(&update);
    memcpy (&updateOrient[0], update.data(), sizeof(updateOrient));
    // std::cout << "Recvd: [" << otherID << ", " <<
    //     static_cast<int>(device) << "] " <<
    //     updatePos[0] << ", " <<
    //     updatePos[1] << ", " <<
    //     updatePos[2] << " " <<
    //     std::endl;
    // std::cout << "      orient: " <<
    //     updateOrient[0] << ", " <<
    //     updateOrient[1] << ", " <<
    //     updateOrient[2] << " " <<
    //     std::endl;
    // if this update is from us, we ignore it by default.
    if (otherID != this->CollabID || this->DisplayOwnAvatar)
    {
      auto avatar = this->GetAvatar(otherID);
      if (device == vtkEventDataDevice::LeftController)
      {
        avatar->SetLeftHandPosition(updatePos);
        avatar->SetLeftHandOrientation(updateOrient);
        if (!avatar->GetUseLeftHand())
        {
          avatar->UseLeftHandOn();
        }
      }
      else if (device == vtkEventDataDevice::RightController)
      {
        avatar->SetRightHandPosition(updatePos);
        avatar->SetRightHandOrientation(updateOrient);
        if (!avatar->GetUseRightHand())
        {
          avatar->UseRightHandOn();
        }
      }
      else if (device == vtkEventDataDevice::HeadMountedDisplay)
      {
        avatar->SetHeadPosition(updatePos);
        avatar->SetHeadOrientation(updateOrient);
      }
    double scale = this->ScaleCallback(this->ScaleCallbackClientData);
    avatar->SetScale(0.3 * scale);
    }
    // Check if we were idle, and re-send join messages.
    if (otherID == this->CollabID && this->AvatarIdle(this->CollabID) &&
        device == vtkEventDataDevice::HeadMountedDisplay)
    {
      mvLog("Collab " << otherID << " return from idle " <<std::endl);
      this->SendMessage("J", this->CollabID);
    }
    this->AvatarUpdateTime[otherID][static_cast<int>(device)] = vtkTimerLog::GetUniversalTime();
  }
  else if (type == "J")
  {
    // Join message, send our list of views.
    std::string extraID = s_recv(this->Subscriber);

    // if we are idle, don't respond to join messages - send a join when
    // we are not idle anymore.
    if (this->AvatarIdle(this->CollabID))
    {
      return;
    }
    mvLog("Collab " << otherID << ", Join" << std::endl);
    if (!this->CollabName.empty())
    {
      this->SendMessage("N", this->CollabName);
    }
  }
  else if (type == "SR" || type == "HR")
  {
    int device;
    // show/hide a ray
    zmq::message_t update;
    this->Subscriber.recv(&update);
    memcpy (&device, update.data(), sizeof(device));
    bool show = (type == "SR");
    if (this->Avatars.count(otherID) != 0)
    {
      auto avatar = this->GetAvatar(otherID);
      if (device == static_cast<int>(vtkEventDataDevice::LeftController))
      {
        avatar->SetLeftShowRay(show);
      }
      else if (device == static_cast<int>(vtkEventDataDevice::RightController))
      {
        avatar->SetRightShowRay(show);
      }
      double scale = this->ScaleCallback(this->ScaleCallbackClientData);
      avatar->SetRayLength(RAY_LENGTH * scale);
    }
  }
  else if (type == "N")
  {
    // Set avatar's name, displayed above head.
    std::string avatarName = s_recv(this->Subscriber);
    mvLog("Collab " << otherID << ", Name " << avatarName << std::endl);
    if (!avatarName.empty() && otherID != this->CollabID)
    {
      this->GetAvatar(otherID)->SetLabel(avatarName.c_str());
    }
  }
}

vtkSmartPointer<vtkOpenGLAvatar> mvCollaborationClient::GetAvatar(std::string otherID)
{
  // if it's from a new collaborator, add an avatar
  if (this->Avatars.count(otherID) == 0)
  {
    mvLog("Adding Avatar " << otherID << std::endl);
    this->Avatars[otherID] = vtkSmartPointer<vtkOpenGLAvatar>::New();
    auto newAvatar = this->Avatars[otherID];
    this->Renderer->AddActor(newAvatar);
    // meters -> ft conversion.
    double scale = this->ScaleCallback(this->ScaleCallbackClientData);
    newAvatar->SetScale(0.3 * scale);
    newAvatar->SetUpVector(0, 0, 1);
    size_t colorIndex = this->Avatars.size() - 1;
    // base the color on the server's index of avatars.
    try
    {
      colorIndex = std::stoi(otherID);
    }
    catch (...) { }
    newAvatar->GetProperty()->SetColor(AVATAR_COLORS[(colorIndex) % NUM_COLORS]);
    newAvatar->GetLabelTextProperty()->SetColor(AVATAR_COLORS[(colorIndex) % NUM_COLORS]);
    newAvatar->GetLabelTextProperty()->SetFontSize ( 16 );
    if (otherID == this->CollabID)
    {
      // Display only the hands
      newAvatar->SetShowHandsOnly(true);
      auto ovrrw = vtkOpenVRRenderWindow::SafeDownCast(this->RenderWindow);
      if (ovrrw)
      {
        vtkOpenVRModel *cmodel =
          ovrrw->GetTrackedDeviceModel(vtkEventDataDevice::LeftController);
        if (cmodel)
        {
          cmodel->SetVisibility(false);
        }
        cmodel =
          ovrrw->GetTrackedDeviceModel(vtkEventDataDevice::RightController);
        if (cmodel)
        {
          cmodel->SetVisibility(false);
        }
      }
    }
    // else
    // {
    //   newAvatar->SetLabel(("Avatar " + otherID).c_str());
    // }
    this->AvatarUpdateTime[otherID][0] = 0;
    this->AvatarUpdateTime[otherID][1] = 0;
    this->AvatarUpdateTime[otherID][2] = 0;
  }
  return this->Avatars[otherID];
}

void mvCollaborationClient::HandleCollabMessage()
{
  double currTime = vtkTimerLog::GetUniversalTime();
  bool receivedMsg = true;
  do
  {
    // timeout is 0, return immediately.
    zmq::poll(&(this->CollabPollItems)[0], 2, 0);
    if (this->CollabPollItems[0].revents & ZMQ_POLLIN)
    {
      // reply on the request-reply (dealer) socket - expect ID or error.
      std::string reply = s_recv_noblock(this->Requester);
      if (reply == "ERROR")
      {
        mvLog("Collab server returned error " << std::endl);
      }
      else if (reply == "pong")
      {
        // update server alive time, below.
        // std::cout << "pong " << this->CollabID << std::endl;
      }
      else if (reply.empty())
      {
        // error, do nothing.
        mvLog("Error: empty reply " << std::endl);
      }
      else
      {
        this->CollabID = reply;
        mvLog("Received ID " << this->CollabID << std::endl);
        this->RetryCount = 0;
        // ideally send "J" join message here, but pub-sub not ready yet.
      }

    }


    // handle broadcast messages
    //
    // A - avatar position update
    // J - New client joined message
    // N - Client name
    // SR/HR - show or hide a ray
    // V - View change
    // P - New TourStop
    // VL - ViewList
    //
    if (this->CollabPollItems[1].revents & ZMQ_POLLIN)
    {
      zmq::message_t update;
      if (this->Subscriber.recv(&update, ZMQ_DONTWAIT))
      {
        if (update.size() == 0)
        {
          std::cout << "Error: empty session header" << std::endl;
          s_clear(this->Subscriber);
          continue;
        }
        // verify the signature
        std::string sig = std::string(static_cast<const char *>(update.data()), update.size());
        // we can get bad data, so make sure the first message contains the
        // correct data before requesting other pieces (which could block
        // and hang the app if the data was bad)
        if (sig == this->CollabSession)
        {
          // the first sub-msg contains the session string for the subscription
          //  process other avatar updates
          std::string otherID = s_recv_noblock(this->Subscriber);
          std::string type = s_recv_noblock(this->Subscriber);
          if (otherID.empty() || type.empty())
          {
            // error, ignore
            std::cout << "empty ID or ID " << otherID << ",  " << type << std::endl;
            s_clear(this->Subscriber);
            continue;
          }

          this->HandleBroadcastMessage(otherID, type);
        }
        else
        {
          std::cout << "Error: mismatched session header" << std::endl;
          s_clear(this->Subscriber);
        }

        // we got a message on the publish socket, see if this is the first one.
        if (!this->PublishAvailable)
        {
          this->PublishAvailable = true;
          // send join message, to trigger view setup.
          this->SendMessage("J", this->CollabID);
        }
      }
    }

    receivedMsg = (this->CollabPollItems[0].revents & ZMQ_POLLIN || this->CollabPollItems[1].revents & ZMQ_POLLIN);
    if (receivedMsg)
    {
      // got a message, reset heartbeat.
      this->NeedHeartbeat = currTime + HEARTBEAT_INTERVAL;
      this->NeedReply = currTime + HEARTBEAT_INTERVAL * LIVE_COUNT;
      this->RetryCount = 0;
    }
    else if (currTime > this->NeedHeartbeat && !this->CollabID.empty())
    {
      // heartbeat only if we have an ID.
      // send ping, expect pong
      // std::cout << "sent ping" << std::endl;
      if (this->RetryCount == 0)
      {
        this->RetryCount = 1;
      }
      s_sendmore(this->Requester, "ping");
      s_send(this->Requester, this->CollabID);
      this->NeedHeartbeat = currTime + HEARTBEAT_INTERVAL;
    }

    // if heartbeat fails multiple times
    if (currTime > this->NeedReply)
    {
      if (this->RetryCount > LIVE_COUNT)
      {
        this->NeedReply = currTime + HEARTBEAT_INTERVAL * LIVE_COUNT * this->RetryCount;
        mvLog("Collab server disconnected, waiting. " << std::endl);
      }
      else
      {
        mvLog("Collab server not responding, retry " << this->RetryCount << std::endl);
        ++this->RetryCount;
        // disconnect and reconnect sockets, clear ID
        this->Initialize(this->Renderer);
      }
    }
  }
  while(receivedMsg);
}

bool mvCollaborationClient::AvatarIdle(std::string id)
{
  double currTime = vtkTimerLog::GetUniversalTime();
  auto times = this->AvatarUpdateTime[id];

  // if we've never received a head position message, the avatar isn't idle.
  if (times[0] == 0) return false;

  double avatarTime = times[0]; //max(max(times[0], times[1]), times[2]);
  // consider ourselves idle slightly before any collaborators do, avoiding races.
  double timeout = id == this->CollabID ? 0.98 * AVATAR_TIMEOUT : AVATAR_TIMEOUT;
  return (currTime - avatarTime > timeout);
}


void mvCollaborationClient::EraseIdleAvatars()
{
  double currTime = vtkTimerLog::GetUniversalTime();
  for (auto it : this->AvatarUpdateTime)
  {
    if (it.second[0] == 0) continue;
    double avatarTime = it.second[0]; //max(max(it.second[0], it.second[1]), it.second[2]);
    if (currTime - avatarTime > AVATAR_TIMEOUT &&
      it.first != this->CollabID && this->Avatars.count(it.first) != 0)
    {
      mvLog("Removing Avatar: " << it.first << std::endl);
      this->Renderer->RemoveActor(this->Avatars[it.first]);
      this->Avatars.erase(it.first);
      this->AvatarUpdateTime.erase(it.first);
      // send join message, to trigger view setup.
      this->SendMessage("J", this->CollabID);
      break;
    }

    if (this->Avatars.count(it.first) == 0)
    {
      continue;
    }

    // see if the hands are idle, or not present at all.
    int device = static_cast<int>(vtkEventDataDevice::LeftController);
    if (currTime - it.second[device] > AVATAR_TIMEOUT)
    {
      auto currAvatar = this->Avatars[it.first];
      if (currAvatar->GetUseLeftHand())
      {
        currAvatar->UseLeftHandOff();
      }
    }
    device = static_cast<int>(vtkEventDataDevice::RightController);
    if (currTime - it.second[device] > AVATAR_TIMEOUT)
    {
      auto currAvatar = this->Avatars[it.first];
      if (currAvatar->GetUseRightHand())
      {
        currAvatar->UseRightHandOff();
      }
    }
  }
}

void mvCollaborationClient::EventCallback(
  vtkObject* , unsigned long eventID, void* clientdata, void* calldata)
{
  mvCollaborationClient* self = static_cast<mvCollaborationClient *>(clientdata);

  if (eventID == vtkCommand::Move3DEvent)
  {
    vtkEventData* edata = static_cast<vtkEventData*>(calldata);
    vtkEventDataDevice3D* edd = edata->GetAsEventDataDevice3D();
    if (!edd)
    {
      return;
    }
    if (edd->GetDevice() == vtkEventDataDevice::LeftController ||
      edd->GetDevice() == vtkEventDataDevice::RightController ||
      edd->GetDevice() == vtkEventDataDevice::HeadMountedDisplay)
    {
      double pos[4] = { 0 };
      double orient[4] = { 0 };
      edd->GetWorldPosition(pos);
      pos[3] = 1.0;
      // empirically, the Occulus sometimes gives nonsense positions
      if (fabs(pos[0]) > 1e07)
      {
        return;
      }
      double wxyz[4] = { 0 };
      edd->GetWorldOrientation(wxyz);
      // currently have a mismatch between wxyz and euler angles. Convert.
      vtkNew<vtkTransform> trans;
      trans->RotateWXYZ(wxyz[0], &wxyz[1]);
      // angles need to be rotated 90
      trans->RotateY(90);
      trans->GetOrientation(orient);
      // hands are also too far forward in x.
      if (edd->GetDevice() != vtkEventDataDevice::HeadMountedDisplay)
      {
        double adjust[3] = { -0.15, 0, 0 };
        trans->TransformPoint(adjust, adjust);
        pos[0] += adjust[0];
        pos[1] += adjust[1];
        pos[2] += adjust[2];
      }

      self->SendDevicePoseMessage(edd->GetDevice(), pos, orient);
    }
    return;
  }
}

// disconnect if needed, then connect to server.
// Retry count is set externally.
bool mvCollaborationClient::Initialize(vtkOpenGLRenderer *ren)
{
  if (!ren)
  {
    return false;
  }

  this->Renderer = ren;
  this->RenderWindow =
    static_cast<vtkOpenGLRenderWindow *>(ren->GetVTKWindow());

  if(this->CollabHost.empty())
  {
    return false;
  }

  if (this->RetryCount == 1)
  {
    mvLog("Connecting to collaboration server..." << std::endl);
  }
  std::stringstream ss;
  ss << "tcp://" << this->CollabHost << ":" << this->CollabPort;
  std::string requesterEndpoint = ss.str();
  ss.str(std::string());
  ss << "tcp://" << this->CollabHost << ":" << (this->CollabPort + 1);
  std::string subscriberEndpoint = ss.str();
  if (this->Requester.connected())
  {
    this->Requester.close();
  }
  if (this->Subscriber.connected())
  {
    this->Subscriber.close();
  }
  this->Connected = false;
  this->Requester = zmq::socket_t(this->Context, ZMQ_DEALER);
  this->Subscriber = zmq::socket_t(this->Context, ZMQ_SUB);

  this->Requester.connect(requesterEndpoint);
  this->Subscriber.connect(subscriberEndpoint);
  // Subscribe to messages for our session, subscription required by zmq
  // We won't receive messages from other sessions.
  this->Subscriber.setsockopt(ZMQ_SUBSCRIBE, this->CollabSession.c_str(), this->CollabSession.length());
  // once we close, we want the socket to close immediately, and drop messages.
  int linger = 0;
  this->Requester.setsockopt(ZMQ_LINGER, &linger, sizeof (linger));
  this->CollabPollItems[0].socket = this->Requester;
  this->CollabPollItems[1].socket = this->Subscriber;
  this->Connected = true;

  this->CollabID.clear();
  double currTime = vtkTimerLog::GetUniversalTime();
  this->NeedHeartbeat = currTime + HEARTBEAT_INTERVAL;
  this->NeedReply = currTime + HEARTBEAT_INTERVAL * LIVE_COUNT * this->RetryCount;
  this->PublishAvailable = false;
  s_send(this->Requester, "HelloPMVZ");
  // async reply, so get ID in HandleCollabMessage()

  if (this->MoveObserver == -1)
  {
    this->MoveObserver =
      this->RenderWindow->GetInteractor()->AddObserver(vtkCommand::Move3DEvent, this->EventCommand, 1.0);
  }

  return true;
}
