/*
 * Copyright DataStax, Inc.
 *
 * Licensed 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 com.datastax.oss.driver.internal.core.metadata.schema.refresh;

import com.datastax.oss.driver.api.core.CqlIdentifier;
import com.datastax.oss.driver.api.core.metadata.schema.KeyspaceMetadata;
import com.datastax.oss.driver.internal.core.context.InternalDriverContext;
import com.datastax.oss.driver.internal.core.metadata.DefaultMetadata;
import com.datastax.oss.driver.internal.core.metadata.MetadataRefresh;
import com.datastax.oss.driver.internal.core.metadata.schema.events.AggregateChangeEvent;
import com.datastax.oss.driver.internal.core.metadata.schema.events.FunctionChangeEvent;
import com.datastax.oss.driver.internal.core.metadata.schema.events.KeyspaceChangeEvent;
import com.datastax.oss.driver.internal.core.metadata.schema.events.TableChangeEvent;
import com.datastax.oss.driver.internal.core.metadata.schema.events.TypeChangeEvent;
import com.datastax.oss.driver.internal.core.metadata.schema.events.ViewChangeEvent;
import com.datastax.oss.driver.shaded.guava.common.annotations.VisibleForTesting;
import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
import com.datastax.oss.driver.shaded.guava.common.collect.Sets;
import java.util.Map;
import java.util.function.BiFunction;
import java.util.function.Function;
import net.jcip.annotations.ThreadSafe;

@ThreadSafe
public class SchemaRefresh implements MetadataRefresh {

  @VisibleForTesting public final Map<CqlIdentifier, KeyspaceMetadata> newKeyspaces;

  public SchemaRefresh(Map<CqlIdentifier, KeyspaceMetadata> newKeyspaces) {
    this.newKeyspaces = newKeyspaces;
  }

  @Override
  public Result compute(
      DefaultMetadata oldMetadata, boolean tokenMapEnabled, InternalDriverContext context) {
    ImmutableList.Builder<Object> events = ImmutableList.builder();

    Map<CqlIdentifier, KeyspaceMetadata> oldKeyspaces = oldMetadata.getKeyspaces();
    for (CqlIdentifier removedKey : Sets.difference(oldKeyspaces.keySet(), newKeyspaces.keySet())) {
      events.add(KeyspaceChangeEvent.dropped(oldKeyspaces.get(removedKey)));
    }
    for (Map.Entry<CqlIdentifier, KeyspaceMetadata> entry : newKeyspaces.entrySet()) {
      CqlIdentifier key = entry.getKey();
      computeEvents(oldKeyspaces.get(key), entry.getValue(), events);
    }

    return new Result(
        oldMetadata.withSchema(this.newKeyspaces, tokenMapEnabled, context), events.build());
  }

  
Computes the exact set of events to emit when a keyspace has changed.
We can't simply emit KeyspaceChangeEvent.updated(KeyspaceMetadata, KeyspaceMetadata) because this method might be called as part of a full schema refresh, or a keyspace refresh initiated by coalesced child element refreshes. We need to traverse all children to check what has exactly changed. /**
   * Computes the exact set of events to emit when a keyspace has changed.
   *
   * <p>We can't simply emit {@link KeyspaceChangeEvent#updated(KeyspaceMetadata, KeyspaceMetadata)}
   * because this method might be called as part of a full schema refresh, or a keyspace refresh
   * initiated by coalesced child element refreshes. We need to traverse all children to check what
   * has exactly changed.
   */
  private void computeEvents(
      KeyspaceMetadata oldKeyspace,
      KeyspaceMetadata newKeyspace,
      ImmutableList.Builder<Object> events) {
    if (oldKeyspace == null) {
      events.add(KeyspaceChangeEvent.created(newKeyspace));
    } else {
      if (!oldKeyspace.shallowEquals(newKeyspace)) {
        events.add(KeyspaceChangeEvent.updated(oldKeyspace, newKeyspace));
      }
      computeChildEvents(oldKeyspace, newKeyspace, events);
    }
  }

  private void computeChildEvents(
      KeyspaceMetadata oldKeyspace,
      KeyspaceMetadata newKeyspace,
      ImmutableList.Builder<Object> events) {
    computeChildEvents(
        oldKeyspace.getTables(),
        newKeyspace.getTables(),
        TableChangeEvent::dropped,
        TableChangeEvent::created,
        TableChangeEvent::updated,
        events);
    computeChildEvents(
        oldKeyspace.getViews(),
        newKeyspace.getViews(),
        ViewChangeEvent::dropped,
        ViewChangeEvent::created,
        ViewChangeEvent::updated,
        events);
    computeChildEvents(
        oldKeyspace.getUserDefinedTypes(),
        newKeyspace.getUserDefinedTypes(),
        TypeChangeEvent::dropped,
        TypeChangeEvent::created,
        TypeChangeEvent::updated,
        events);
    computeChildEvents(
        oldKeyspace.getFunctions(),
        newKeyspace.getFunctions(),
        FunctionChangeEvent::dropped,
        FunctionChangeEvent::created,
        FunctionChangeEvent::updated,
        events);
    computeChildEvents(
        oldKeyspace.getAggregates(),
        newKeyspace.getAggregates(),
        AggregateChangeEvent::dropped,
        AggregateChangeEvent::created,
        AggregateChangeEvent::updated,
        events);
  }

  private <K, V> void computeChildEvents(
      Map<K, V> oldChildren,
      Map<K, V> newChildren,
      Function<V, Object> newDroppedEvent,
      Function<V, Object> newCreatedEvent,
      BiFunction<V, V, Object> newUpdatedEvent,
      ImmutableList.Builder<Object> events) {
    for (K removedKey : Sets.difference(oldChildren.keySet(), newChildren.keySet())) {
      events.add(newDroppedEvent.apply(oldChildren.get(removedKey)));
    }
    for (Map.Entry<K, V> entry : newChildren.entrySet()) {
      K key = entry.getKey();
      V newChild = entry.getValue();
      V oldChild = oldChildren.get(key);
      if (oldChild == null) {
        events.add(newCreatedEvent.apply(newChild));
      } else if (!oldChild.equals(newChild)) {
        events.add(newUpdatedEvent.apply(oldChild, newChild));
      }
    }
  }
}