import org.apache.hadoop.crypto.CryptoOutputStream; //导入依赖的package包/类
/**
 * Create IOStreamPair of {@link org.apache.hadoop.crypto.CryptoInputStream}
 * and {@link org.apache.hadoop.crypto.CryptoOutputStream}
 * 
 * @param conf the configuration
 * @param cipherOption negotiated cipher option
 * @param out underlying output stream
 * @param in underlying input stream
 * @param isServer is server side
 * @return IOStreamPair the stream pair
 * @throws IOException for any error
 */
public static IOStreamPair createStreamPair(Configuration conf,
    CipherOption cipherOption, OutputStream out, InputStream in, 
    boolean isServer) throws IOException {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Creating IOStreamPair of CryptoInputStream and " +
        "CryptoOutputStream.");
  }
  CryptoCodec codec = CryptoCodec.getInstance(conf, 
      cipherOption.getCipherSuite());
  byte[] inKey = cipherOption.getInKey();
  byte[] inIv = cipherOption.getInIv();
  byte[] outKey = cipherOption.getOutKey();
  byte[] outIv = cipherOption.getOutIv();
  InputStream cIn = new CryptoInputStream(in, codec, 
      isServer ? inKey : outKey, isServer ? inIv : outIv);
  OutputStream cOut = new CryptoOutputStream(out, codec, 
      isServer ? outKey : inKey, isServer ? outIv : inIv);
  return new IOStreamPair(cIn, cOut);
}
 

import org.apache.hadoop.crypto.CryptoOutputStream; //导入依赖的package包/类
/**
 * Wraps the stream in a CryptoOutputStream if the underlying file is
 * encrypted.
 */
public HdfsDataOutputStream createWrappedOutputStream(DFSOutputStream dfsos,
    FileSystem.Statistics statistics, long startPos) throws IOException {
  final FileEncryptionInfo feInfo = dfsos.getFileEncryptionInfo();
  if (feInfo != null) {
    // File is encrypted, wrap the stream in a crypto stream.
    // Currently only one version, so no special logic based on the version #
    getCryptoProtocolVersion(feInfo);
    final CryptoCodec codec = getCryptoCodec(conf, feInfo);
    KeyVersion decrypted = decryptEncryptedDataEncryptionKey(feInfo);
    final CryptoOutputStream cryptoOut =
        new CryptoOutputStream(dfsos, codec,
            decrypted.getMaterial(), feInfo.getIV(), startPos);
    return new HdfsDataOutputStream(cryptoOut, statistics, startPos);
  } else {
    // No FileEncryptionInfo present so no encryption.
    return new HdfsDataOutputStream(dfsos, statistics, startPos);
  }
}
 

import org.apache.hadoop.crypto.CryptoOutputStream; //导入依赖的package包/类
/**
 * Create IOStreamPair of {@link org.apache.hadoop.crypto.CryptoInputStream}
 * and {@link org.apache.hadoop.crypto.CryptoOutputStream}
 *
 * @param conf the configuration
 * @param cipherOption negotiated cipher option
 * @param out underlying output stream
 * @param in underlying input stream
 * @param isServer is server side
 * @return IOStreamPair the stream pair
 * @throws IOException for any error
 */
public static IOStreamPair createStreamPair(Configuration conf,
    CipherOption cipherOption, OutputStream out, InputStream in,
    boolean isServer) throws IOException {
  LOG.debug("Creating IOStreamPair of CryptoInputStream and "
      + "CryptoOutputStream.");
  CryptoCodec codec = CryptoCodec.getInstance(conf,
      cipherOption.getCipherSuite());
  byte[] inKey = cipherOption.getInKey();
  byte[] inIv = cipherOption.getInIv();
  byte[] outKey = cipherOption.getOutKey();
  byte[] outIv = cipherOption.getOutIv();
  InputStream cIn = new CryptoInputStream(in, codec,
      isServer ? inKey : outKey, isServer ? inIv : outIv);
  OutputStream cOut = new CryptoOutputStream(out, codec,
      isServer ? outKey : inKey, isServer ? outIv : inIv);
  return new IOStreamPair(cIn, cOut);
}
 

import org.apache.hadoop.crypto.CryptoOutputStream; //导入依赖的package包/类
/**
 * Sync buffered data to DataNodes (flush to disk devices).
 * 
 * @param syncFlags
 *          Indicate the detailed semantic and actions of the hsync.
 * @throws IOException
 * @see FSDataOutputStream#hsync()
 */
public void hsync(EnumSet<SyncFlag> syncFlags) throws IOException {
  OutputStream wrappedStream = getWrappedStream();
  if (wrappedStream instanceof CryptoOutputStream) {
    ((CryptoOutputStream) wrappedStream).flush();
    wrappedStream = ((CryptoOutputStream) wrappedStream).getWrappedStream();
  }
  ((DFSOutputStream) wrappedStream).hsync(syncFlags);
}
 

import org.apache.hadoop.crypto.CryptoOutputStream; //导入依赖的package包/类
public HdfsDataOutputStream(CryptoOutputStream out,
    FileSystem.Statistics stats, long startPosition) throws IOException {
  super(out, stats, startPosition);
  Preconditions.checkArgument(
      out.getWrappedStream() instanceof DFSOutputStream,
      "CryptoOutputStream should wrap a DFSOutputStream");
}
 

import org.apache.hadoop.crypto.CryptoOutputStream; //导入依赖的package包/类
/**
 * Sync buffered data to DataNodes (flush to disk devices).
 *
 * @param syncFlags
 *          Indicate the detailed semantic and actions of the hsync.
 * @throws IOException
 * @see FSDataOutputStream#hsync()
 */
public void hsync(EnumSet<SyncFlag> syncFlags) throws IOException {
  OutputStream wrappedStream = getWrappedStream();
  if (wrappedStream instanceof CryptoOutputStream) {
    wrappedStream.flush();
    wrappedStream = ((CryptoOutputStream) wrappedStream).getWrappedStream();
  }
  ((DFSOutputStream) wrappedStream).hsync(syncFlags);
}
 

import org.apache.hadoop.crypto.CryptoOutputStream; //导入依赖的package包/类
public CryptoFSDataOutputStream(FSDataOutputStream out, CryptoCodec codec,
    int bufferSize, byte[] key, byte[] iv) throws IOException {
  super(new CryptoOutputStream(out, codec, bufferSize, key, iv, 
      out.getPos()), null, out.getPos()); 
  this.fsOut = out;
}
 

import org.apache.hadoop.crypto.CryptoOutputStream; //导入依赖的package包/类
public HdfsDataOutputStream(CryptoOutputStream out, FileSystem.Statistics stats,
    long startPosition) throws IOException {
  super(out, stats, startPosition);
  Preconditions.checkArgument(out.getWrappedStream() instanceof DFSOutputStream,
      "CryptoOutputStream should wrap a DFSOutputStream");
}
 

import org.apache.hadoop.crypto.CryptoOutputStream; //导入依赖的package包/类
/**
 * Get the actual number of replicas of the current block.
 * 
 * This can be different from the designated replication factor of the file
 * because the namenode does not maintain replication for the blocks which are
 * currently being written to. Depending on the configuration, the client may
 * continue to write to a block even if a few datanodes in the write pipeline
 * have failed, or the client may add a new datanodes once a datanode has
 * failed.
 * 
 * @return the number of valid replicas of the current block
 */
public synchronized int getCurrentBlockReplication() throws IOException {
  OutputStream wrappedStream = getWrappedStream();
  if (wrappedStream instanceof CryptoOutputStream) {
    wrappedStream = ((CryptoOutputStream) wrappedStream).getWrappedStream();
  }
  return ((DFSOutputStream) wrappedStream).getCurrentBlockReplication();
}
 

import org.apache.hadoop.crypto.CryptoOutputStream; //导入依赖的package包/类
/**
 * Get the actual number of replicas of the current block.
 *
 * This can be different from the designated replication factor of the file
 * because the namenode does not maintain replication for the blocks which are
 * currently being written to. Depending on the configuration, the client may
 * continue to write to a block even if a few datanodes in the write pipeline
 * have failed, or the client may add a new datanodes once a datanode has
 * failed.
 *
 * @return the number of valid replicas of the current block
 */
public synchronized int getCurrentBlockReplication() throws IOException {
  OutputStream wrappedStream = getWrappedStream();
  if (wrappedStream instanceof CryptoOutputStream) {
    wrappedStream = ((CryptoOutputStream) wrappedStream).getWrappedStream();
  }
  return ((DFSOutputStream) wrappedStream).getCurrentBlockReplication();
}