class TaskStatusListener {

  @Subscribe
  public void onTaskUpdateEvent(TaskUpdateEvent event) {
    LOG.debug("Received task update event {}", event);
    List<HostRoleCommand> hostRoleCommandListAll = event.getHostRoleCommands();
    List<HostRoleCommand>  hostRoleCommandWithReceivedStatus =  new ArrayList<>();
    Set<StageEntityPK> stagesWithReceivedTaskStatus = new HashSet<>();
    Set<Long> requestIdsWithReceivedTaskStatus =  new HashSet<>();
    Set<RequestUpdateEvent> requestsToPublish = new HashSet<>();

    for (HostRoleCommand hostRoleCommand : hostRoleCommandListAll) {
      Long reportedTaskId = hostRoleCommand.getTaskId();
      HostRoleCommand activeTask =  activeTasksMap.get(reportedTaskId);
      if (activeTask == null) {
        LOG.error(String.format("Received update for a task %d which is not being tracked as running task", reportedTaskId));
      } else  {
        hostRoleCommandWithReceivedStatus.add(hostRoleCommand);
        StageEntityPK stageEntityPK = new StageEntityPK();
        stageEntityPK.setRequestId(hostRoleCommand.getRequestId());
        stageEntityPK.setStageId(hostRoleCommand.getStageId());
        stagesWithReceivedTaskStatus.add(stageEntityPK);
        requestIdsWithReceivedTaskStatus.add(hostRoleCommand.getRequestId());

        if (!activeTasksMap.get(reportedTaskId).getStatus().equals(hostRoleCommand.getStatus())) {
          // Ignore requests not related to any cluster. "requests" topic is used for cluster requests only.
          Long clusterId = activeRequestMap.get(hostRoleCommand.getRequestId()).getClusterId();
          if (clusterId != null && clusterId != -1) {
            Set<RequestUpdateEvent.HostRoleCommand> hostRoleCommands = new HashSet<>();
            hostRoleCommands.add(new RequestUpdateEvent.HostRoleCommand(hostRoleCommand.getTaskId(),
                hostRoleCommand.getRequestId(),
                hostRoleCommand.getStatus(),
                hostRoleCommand.getHostName()));
            requestsToPublish.add(new RequestUpdateEvent(hostRoleCommand.getRequestId(),
                activeRequestMap.get(hostRoleCommand.getRequestId()).getStatus(), hostRoleCommands));
          } else {
            LOG.debug("No STOMP request update event was fired for host component status change due no cluster related, " +
                    "request id: {}, role: {}, role command: {}, host: {}, task id: {}, old state: {}, new state: {}",
                hostRoleCommand.getRequestId(),
                hostRoleCommand.getRole(),
                hostRoleCommand.getRoleCommand(),
                hostRoleCommand.getHostName(),
                hostRoleCommand.getTaskId(),
                activeTasksMap.get(reportedTaskId).getStatus(),
                hostRoleCommand.getStatus());
          }
        }
      }
    }
    updateActiveTasksMap(hostRoleCommandWithReceivedStatus);
    Boolean didAnyStageStatusUpdated = updateActiveStagesStatus(stagesWithReceivedTaskStatus, hostRoleCommandListAll);
    // Presumption: If there is no update in any of the running stage's status
    // then none of the running request status needs to be updated
    if (didAnyStageStatusUpdated) {
      updateActiveRequestsStatus(requestIdsWithReceivedTaskStatus, stagesWithReceivedTaskStatus);
    }
    for (RequestUpdateEvent requestToPublish : requestsToPublish) {
      STOMPUpdatePublisher.publish(requestToPublish);
    }
  }

  @Inject public  TaskStatusListener(TaskEventPublisher taskEventPublisher, StageDAO stageDAO, RequestDAO requestDAO,
                            STOMPUpdatePublisher STOMPUpdatePublisher);

}