content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package org.apache.ambari.server.agent.stomp;
  
  import java.util.Map;
  import java.util.Objects;
  import java.util.concurrent.ConcurrentHashMap;
  
  import jakarta.inject.Inject;
  
  import org.apache.ambari.server.AmbariException;
  import org.apache.ambari.server.AmbariRuntimeException;
  import org.apache.ambari.server.agent.stomp.dto.Hashable;
  import org.apache.ambari.server.events.STOMPEvent;
  import org.apache.ambari.server.events.STOMPHostEvent;
  import org.apache.ambari.server.events.publishers.STOMPUpdatePublisher;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * Is used to saving and updating last version of event in host scope
   * @param <T> event with hash to control version
   */
  public abstract class AgentHostDataHolder<T extends STOMPHostEvent & Hashable> extends AgentDataHolder<T> {
    public static final Logger LOG = LoggerFactory.getLogger(AgentHostDataHolder.class);
  
    @Inject
    private STOMPUpdatePublisher STOMPUpdatePublisher;
  
    private final ConcurrentHashMap<Long, T> data = new ConcurrentHashMap<>();
  
    protected abstract T getCurrentData(Long hostId) throws AmbariException;
    protected abstract T handleUpdate(T current, T update) throws AmbariException;
  
    public T getUpdateIfChanged(String agentHash, Long hostId) throws AmbariException {
      T hostData = initializeDataIfNeeded(hostId, true);
      return !Objects.equals(agentHash, hostData.getHash()) ? hostData : getEmptyData();
    }
  
    public T initializeDataIfNeeded(Long hostId, boolean regenerateHash) throws AmbariRuntimeException {
      return data.computeIfAbsent(hostId, id -> initializeData(hostId, regenerateHash));
    }
  
    private T initializeData(Long hostId, boolean regenerateHash) {
      T hostData;
      try {
        hostData = getCurrentData(hostId);
      } catch (AmbariException e) {
        LOG.error("Error during retrieving initial value for host: {} and class {}", hostId, getClass().getName(), e);
        throw new AmbariRuntimeException("Error during retrieving initial value for host: " + hostId + " and class: " + getClass().getName(), e);
      }
      if (regenerateHash) {
        regenerateDataIdentifiers(hostData);
      }
      return hostData;
    }
  
    /**
     * Apply an incremental update to the data (host-specific), and publish the
     * event to listeners.
     */
    public void updateData(T update) throws AmbariException {
      try {
        data.compute(update.getHostId(), (id, current) -> {
          if (current == null) {
            current = initializeData(id, true);
          }
          T updated;
          try {
            updated = handleUpdate(current, update);
          } catch (AmbariException e) {
            LOG.error("Error during handling update for host: {} and class {}", id, getClass().getName(), e);
            throw new AmbariRuntimeException("Error during handling update for host: " + id + " and class: " + getClass().getName(), e);
          }
          if (updated == null) {
            return current;
          } else {
            regenerateDataIdentifiers(updated);
            setIdentifiersToEventUpdate(update, updated);
            return updated;
          }
        });
      } catch(AmbariRuntimeException e) {
        throw new AmbariException(e.getMessage(), e);
      }
      if (isIdentifierValid(update)) {
        if (update.getType().equals(STOMPEvent.Type.AGENT_CONFIGS)) {
          LOG.info("Configs update with hash {} will be sent to host {}", update.getHash(), update.getHostId());
        }
        STOMPUpdatePublisher.publish(update);
      }
    }
  
    protected void setIdentifiersToEventUpdate(T update, T hostData) {
      update.setHash(hostData.getHash());
    }
  
    /**
     * Reset data for the given host.  Used if changes are complex and it's easier to re-create data from scratch.
     */
    public final void resetData(Long hostId) throws AmbariException {
      T newData = getCurrentData(hostId);
      data.replace(hostId, newData);
      STOMPUpdatePublisher.publish(newData);
    }
  
    /**
     * Remove data for the given host.
     */
    public final void onHostRemoved(Long hostId) {
      data.remove(hostId);
    }
  
    public Map<Long, T> getData() {
      return data;
    }
  
    public T getData(Long hostId) {
      return data.get(hostId);
    }
  
    public void setData(T data, Long hostId) {
      this.data.put(hostId, data);
    }
  }
  