alluxio-master-2.x/core/client/fs/src/main/java/alluxio/client/block/stream/DataWriter.java
/*
 * 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.WriteType;
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 shortCircuit = alluxioConf.getBoolean(PropertyKey.USER_SHORT_CIRCUIT_ENABLED);
      boolean shortCircuitPreferred =
          alluxioConf.getBoolean(PropertyKey.USER_SHORT_CIRCUIT_PREFERRED);
      boolean ufsFallbackEnabled = options.getWriteType() == WriteType.ASYNC_THROUGH
          && alluxioConf.getBoolean(PropertyKey.USER_FILE_UFS_TIER_ENABLED);
      boolean workerIsLocal = CommonUtils.isLocalHost(address, alluxioConf);

      if (workerIsLocal && context.hasProcessLocalWorker() && !ufsFallbackEnabled) {
        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: {}, "
          + "ufs fallback enabled: {})", blockId, address,
          workerIsLocal, context.hasProcessLocalWorker(), ufsFallbackEnabled);

      boolean domainSocketSupported = NettyUtils.isDomainSocketSupported(address);
      if (workerIsLocal && shortCircuit
          && (shortCircuitPreferred || !domainSocketSupported)) {
        if (ufsFallbackEnabled) {
          LOG.info("Creating UFS-fallback short circuit output stream for block {} @ {}", blockId,
              address);
          return UfsFallbackLocalFileDataWriter.create(
              context, address, blockId, blockSize, options);
        }
        LOG.debug("Creating short circuit output stream for block {} @ {}", blockId, address);
        return LocalFileDataWriter.create(context, address, blockId, blockSize, options);
      } else {
        LOG.debug("Creating gRPC output stream for block {} @ {} from client {} "
            + "(data locates in local worker: {}, shortCircuitEnabled: {}, "
            + "shortCircuitPreferred: {}, domainSocketSupported: {})",
            blockId, address, NetworkAddressUtils.getClientHostName(alluxioConf),
            workerIsLocal, shortCircuit, shortCircuitPreferred, 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();
}
