class HeartBeatHandler {

  public HeartBeatResponse handleHeartBeat(HeartBeat heartbeat)
      throws AmbariException {
    long now = System.currentTimeMillis();
    if (heartbeat.getAgentEnv() != null && heartbeat.getAgentEnv().getHostHealth() != null) {
      heartbeat.getAgentEnv().getHostHealth().setServerTimeStampAtReporting(now);
    }

    String hostname = heartbeat.getHostname();
    Long currentResponseId = hostResponseIds.get(hostname);
    HeartBeatResponse response;

    if (currentResponseId == null) {
      //Server restarted, or unknown host.
      LOG.error("CurrentResponseId unknown for " + hostname + " - send register command");
      return createRegisterCommand();
    }

    LOG.debug("Received heartbeat from host, hostname={}, currentResponseId={}, receivedResponseId={}", hostname, currentResponseId, heartbeat.getResponseId());

    response = new HeartBeatResponse();
    Host hostObject;
    try {
      hostObject = clusterFsm.getHost(hostname);
    } catch (HostNotFoundException e) {
      LOG.error("Host: {} not found. Agent is still heartbeating.", hostname);
      if (LOG.isDebugEnabled()) {
        LOG.debug("Host associated with the agent heratbeat might have been " +
            "deleted", e);
      }
      // For now return empty response with only response id.
      return response;
    }

    if (heartbeat.getResponseId() == currentResponseId - 1) {
      HeartBeatResponse heartBeatResponse = hostResponses.get(hostname);

      LOG.warn("Old responseId={} received form host {} - response was lost - returning cached response with responseId={}",
        heartbeat.getResponseId(),
        hostname,
        heartBeatResponse.getResponseId());

      return heartBeatResponse;
    } else if (heartbeat.getResponseId() != currentResponseId) {
      LOG.error("Error in responseId sequence - received responseId={} from host {} - sending agent restart command with responseId={}",
        heartbeat.getResponseId(),
        hostname,
        currentResponseId);

      return createRestartCommand(currentResponseId);
    }
    response.setResponseId(++currentResponseId);

    if (hostObject.getState().equals(HostState.HEARTBEAT_LOST)) {
      // After loosing heartbeat agent should reregister
      LOG.warn("Host {} is in HEARTBEAT_LOST state - sending register command", hostname);
      STOMPUpdatePublisher.publish(new AgentActionEvent(AgentActionEvent.AgentAction.RESTART_AGENT,
          hostObject.getHostId()));
      return createRegisterCommand();
    }

    hostResponseIds.put(hostname, currentResponseId);
    hostResponses.put(hostname, response);

    // If the host is waiting for component status updates, notify it
    if (hostObject.getState().equals(HostState.WAITING_FOR_HOST_STATUS_UPDATES)) {
      try {
        LOG.debug("Got component status updates for host {}", hostname);
        hostObject.handleEvent(new HostStatusUpdatesReceivedEvent(hostname, now));
      } catch (InvalidStateTransitionException e) {
        LOG.warn("Failed to notify the host {} about component status updates", hostname, e);
      }
    }
    if (heartbeat.getRecoveryReport() != null) {
      RecoveryReport rr = heartbeat.getRecoveryReport();
      processRecoveryReport(rr, hostname);
    }

    if (CollectionUtils.isNotEmpty(heartbeat.getStaleAlerts())) {
      alertHelper.addStaleAlerts(hostObject.getHostId(), heartbeat.getStaleAlerts());
    }

    try {
      hostObject.handleEvent(new HostHealthyHeartbeatEvent(hostname, now,
          heartbeat.getAgentEnv(), heartbeat.getMounts()));
    } catch (InvalidStateTransitionException ex) {
      LOG.warn("Asking agent to re-register due to " + ex.getMessage(), ex);
      hostObject.setState(HostState.INIT);
      return createRegisterCommand();
    }

    heartbeatProcessor.addHeartbeat(heartbeat);

    // Send commands if node is active
    if (hostObject.getState().equals(HostState.HEALTHY)) {
      annotateResponse(hostname, response);
    }

    return response;
  }

  @Inject public  HeartBeatHandler(Clusters fsm, ActionManager am,
                          Injector injector);

}