/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.client.block.stream;

import alluxio.client.Cancelable;
import alluxio.client.file.FileSystemContext;
import alluxio.client.file.options.OutStreamOptions;
import alluxio.conf.AlluxioConfiguration;
import alluxio.conf.PropertyKey;
import alluxio.grpc.RequestType;
import alluxio.util.CommonUtils;
import alluxio.util.network.NettyUtils;
import alluxio.util.network.NetworkAddressUtils;
import alluxio.wire.WorkerNetAddress;

import io.netty.buffer.ByteBuf;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.IOException;
import java.util.Optional;
import javax.annotation.concurrent.ThreadSafe;

/**
 * The interface to write data.
 */
public interface DataWriter extends Closeable, Cancelable {

  /**
   * Factory for {@link DataWriter}.
   */
  @ThreadSafe
  class Factory {
    public static final Logger LOG = LoggerFactory.getLogger(DataWriter.Factory.class);

    private Factory() {} // prevent instantiation

    /**
     * @param context the file system context
     * @param blockId the block ID
     * @param blockSize the block size in bytes
     * @param address the Alluxio worker address
     * @param options the out stream options
     * @return the {@link DataWriter} instance
     */
    public static DataWriter create(FileSystemContext context, long blockId, long blockSize,
        WorkerNetAddress address, OutStreamOptions options) throws IOException {
      AlluxioConfiguration alluxioConf = context.getClusterConf();
      boolean workerIsLocal = CommonUtils.isLocalHost(address, alluxioConf);
      boolean nettyTransEnabled =
          alluxioConf.getBoolean(PropertyKey.USER_NETTY_DATA_TRANSMISSION_ENABLED);

      if (workerIsLocal) {
        LOG.debug("Creating worker process local output stream for block {} @ {}",
            blockId, address);
        return BlockWorkerDataWriter.create(context, blockId, blockSize, options);
      }
      LOG.debug("Doesn't create worker process local output stream for block {} @ {} "
          + "(data locates in local worker: {}, client locates in local worker process: {})",
          blockId, address, workerIsLocal, context.hasProcessLocalWorker());

      boolean domainSocketSupported = NettyUtils.isDomainSocketSupported(address);
      if (nettyTransEnabled) {
        LOG.debug("Creating netty output stream for block {} @ {} from client {} "
                + "(data locates in local worker: {}, domainSocketSupported: {})",
            blockId, address, NetworkAddressUtils.getClientHostName(alluxioConf),
            workerIsLocal, domainSocketSupported);
        // TODO(JiamingMai): implement the netty writer here
        return NettyDataWriter
            .create(context, address, blockId, blockSize, RequestType.ALLUXIO_BLOCK,
                options);
      }

      LOG.debug("Creating gRPC output stream for block {} @ {} from client {} "
          + "(data locates in local worker: {}, domainSocketSupported: {})",
          blockId, address, NetworkAddressUtils.getClientHostName(alluxioConf),
          workerIsLocal, domainSocketSupported);
      return GrpcDataWriter
          .create(context, address, blockId, blockSize, RequestType.ALLUXIO_BLOCK,
              options);
    }
  }

  /**
   * @return the content hash of the file if it is written to the UFS. Will only
   * return a non-empty value after the data writer has been closed.
   */
  Optional<String> getUfsContentHash();

  /**
   * Writes a chunk. This method takes the ownership of this chunk even if it fails to write
   * the chunk.
   *
   * @param chunk the chunk
   */
  void writeChunk(ByteBuf chunk) throws IOException;

  /**
   *  Flushes all the pending chunks.
   */
  void flush() throws IOException;

  /**
   * @return the chunk size in bytes used
   */
  int chunkSize();

  /**
   * @return the current pos which is the same as the totally number of bytes written so far
   */
  long pos();
}
