package org.apache.solr.schema;
/*
 * 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.
 */

import org.apache.solr.cloud.ZkSolrResourceLoader;
import org.apache.solr.common.SolrException.ErrorCode;
import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZooKeeperException;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.xml.sax.InputSource;

import java.io.ByteArrayInputStream;
import java.util.concurrent.TimeUnit;

/** Keeps a ManagedIndexSchema up-to-date when changes are made to the serialized managed schema in ZooKeeper */
public class ZkIndexSchemaReader {
  private static final Logger log = LoggerFactory.getLogger(ZkIndexSchemaReader.class);
  private final ManagedIndexSchemaFactory managedIndexSchemaFactory;
  private SolrZkClient zkClient;
  private String managedSchemaPath;

  public ZkIndexSchemaReader(ManagedIndexSchemaFactory managedIndexSchemaFactory) {
    this.managedIndexSchemaFactory = managedIndexSchemaFactory;
    ZkSolrResourceLoader zkLoader = (ZkSolrResourceLoader)managedIndexSchemaFactory.getResourceLoader();
    this.zkClient = zkLoader.getZkController().getZkClient();
    managedSchemaPath = zkLoader.getCollectionZkPath() + "/" + managedIndexSchemaFactory.getManagedSchemaResourceName();
    createSchemaWatcher();
  }

  public Object getSchemaUpdateLock() { 
    return managedIndexSchemaFactory.getSchemaUpdateLock(); 
  }

  public void createSchemaWatcher() {
    log.info("Creating ZooKeeper watch for the managed schema at " + managedSchemaPath + " ...");

    try {
      zkClient.exists(managedSchemaPath, new Watcher() {
        @Override
        public void process(WatchedEvent event) {
          // session events are not change events, and do not remove the watcher
          if (Event.EventType.None.equals(event.getType())) {
            return;
          }
          log.info("A schema change: {}, has occurred - updating schema from ZooKeeper ...", event);
          try {
            updateSchema(this);
          } catch (KeeperException e) {
            if (e.code() == KeeperException.Code.SESSIONEXPIRED || e.code() == KeeperException.Code.CONNECTIONLOSS) {
              log.warn("ZooKeeper watch triggered, but Solr cannot talk to ZK");
              return;
            }
            log.error("", e);
            throw new ZooKeeperException(ErrorCode.SERVER_ERROR, "", e);
          } catch (InterruptedException e) {
            // Restore the interrupted status
            Thread.currentThread().interrupt();
            log.warn("", e);
          }
        }
      }, true);
    } catch (KeeperException e) {
      final String msg = "Error creating ZooKeeper watch for the managed schema";
      log.error(msg, e);
      throw new ZooKeeperException(ErrorCode.SERVER_ERROR, msg, e);
    } catch (InterruptedException e) {
      // Restore the interrupted status
      Thread.currentThread().interrupt();
      log.warn("", e);
    }
  }

  private void updateSchema(Watcher watcher) throws KeeperException, InterruptedException {
    Stat stat = new Stat();
    synchronized (getSchemaUpdateLock()) {
      final ManagedIndexSchema oldSchema = managedIndexSchemaFactory.getSchema();
      byte[] data = zkClient.getData(managedSchemaPath, watcher, stat, true);
      if (stat.getVersion() != oldSchema.schemaZkVersion) {
        log.info("Retrieved schema from ZooKeeper");
        long start = System.nanoTime();
        InputSource inputSource = new InputSource(new ByteArrayInputStream(data));
        ManagedIndexSchema newSchema = oldSchema.reloadFields(inputSource, stat.getVersion());
        managedIndexSchemaFactory.setSchema(newSchema);
        long stop = System.nanoTime();
        log.info("Finished refreshing schema in " + TimeUnit.MILLISECONDS.convert(stop - start, TimeUnit.NANOSECONDS) + " ms");
      }
    }
  }
}