HADOOP-13327 Output Stream Specification

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/CryptoOutputStream.java

@@ -28,6 +28,7 @@
 import org.apache.hadoop.fs.CanSetDropBehind;
 import org.apache.hadoop.fs.StreamCapabilities;
 import org.apache.hadoop.fs.Syncable;
+import org.apache.hadoop.fs.impl.StoreImplementationUtils;
 
 import com.google.common.base.Preconditions;
 
@@ -308,9 +309,6 @@ private void freeBuffers() {
 
   @Override
   public boolean hasCapability(String capability) {
-    if (out instanceof StreamCapabilities) {
-      return ((StreamCapabilities) out).hasCapability(capability);
-    }
-    return false;
+    return StoreImplementationUtils.hasCapability(out, capability);
   }
 }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CanSetDropBehind.java

@@ -36,6 +36,6 @@ public interface CanSetDropBehind {
    *         UnsupportedOperationException  If this stream doesn't support
    *                                        setting the drop-behind.
    */
-  public void setDropBehind(Boolean dropCache) 
+  void setDropBehind(Boolean dropCache)
       throws IOException, UnsupportedOperationException;
 }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSDataInputStream.java

@@ -29,6 +29,7 @@
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.impl.StoreImplementationUtils;
 import org.apache.hadoop.io.ByteBufferPool;
 import org.apache.hadoop.util.IdentityHashStore;
 
@@ -234,10 +235,7 @@ public void unbuffer() {
 
   @Override
   public boolean hasCapability(String capability) {
-    if (in instanceof StreamCapabilities) {
-      return ((StreamCapabilities) in).hasCapability(capability);
-    }
-    return false;
+    return StoreImplementationUtils.hasCapability(in, capability);
   }
 
   /**

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSDataOutputStream.java

@@ -24,6 +24,7 @@
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.impl.StoreImplementationUtils;
 
 /** Utility that wraps a {@link OutputStream} in a {@link DataOutputStream}.
  */
@@ -122,10 +123,7 @@ public OutputStream getWrappedStream() {
 
   @Override
   public boolean hasCapability(String capability) {
-    if (wrappedStream instanceof StreamCapabilities) {
-      return ((StreamCapabilities) wrappedStream).hasCapability(capability);
-    }
-    return false;
+    return StoreImplementationUtils.hasCapability(wrappedStream, capability);
   }
 
   @Override  // Syncable

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSOutputSummer.java

@@ -33,7 +33,8 @@
  */
 @InterfaceAudience.LimitedPrivate({"HDFS"})
 @InterfaceStability.Unstable
-abstract public class FSOutputSummer extends OutputStream {
+abstract public class FSOutputSummer extends OutputStream implements
+    StreamCapabilities {
   // data checksum
   private final DataChecksum sum;
   // internal buffer for storing data before it is checksumed
@@ -254,4 +255,9 @@ protected synchronized void setChecksumBufSize(int size) {
   protected synchronized void resetChecksumBufSize() {
     setChecksumBufSize(sum.getBytesPerChecksum() * BUFFER_NUM_CHUNKS);
   }
+
+  @Override
+  public boolean hasCapability(String capability) {
+    return false;
+  }
 }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/Syncable.java

@@ -27,16 +27,16 @@
 @InterfaceAudience.Public
 @InterfaceStability.Evolving
 public interface Syncable {
-  
+
   /** Flush out the data in client's user buffer. After the return of
    * this call, new readers will see the data.
    * @throws IOException if any error occurs
    */
-  public void hflush() throws IOException;
+  void hflush() throws IOException;
 
   /** Similar to posix fsync, flush out the data in client's user buffer 
    * all the way to the disk device (but the disk may have it in its cache).
    * @throws IOException if error occurs
    */
-  public void hsync() throws IOException;
+  void hsync() throws IOException;
 }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/impl/StoreImplementationUtils.java

@@ -0,0 +1,95 @@
+/*
+ * 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.
+ */
+
+package org.apache.hadoop.fs.impl;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.StreamCapabilities;
+
+import static org.apache.hadoop.fs.StreamCapabilities.HFLUSH;
+import static org.apache.hadoop.fs.StreamCapabilities.HSYNC;
+
+/**
+ * Utility classes to help implementing filesystems and streams.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public final class StoreImplementationUtils {
+
+  private StoreImplementationUtils() {
+  }
+
+  /**
+   * Check the supplied capabilities for being those required for full
+   * {@code Syncable.hsync()} and {@code Syncable.hflush()} functionality.
+   * @param capability capability string.
+   * @return true if either refers to one of the Syncable operations.
+   */
+  public static boolean supportsSyncable(String capability) {
+    return capability.equalsIgnoreCase(HSYNC) ||
+        capability.equalsIgnoreCase((HFLUSH));
+  }
+
+  /**
+   * Probe for an object having a capability; returns true
+   * iff the stream implements {@link StreamCapabilities} and its
+   * {@code hasCapabilities()} method returns true for the capability.
+   * This is a package private method intended to provided a common
+   * implementation for input, output streams; the stronger typed
+   * {@link #hasCapability()} call is for public use.
+   * @param object object to probe.
+   * @param capability capability to probe for
+   * @return true iff the object implements stream capabilities and
+   * declares that it supports the capability.
+   */
+  static boolean objectHasCapability(Object object, String capability) {
+    if (object instanceof StreamCapabilities) {
+      return ((StreamCapabilities) object).hasCapability(capability);
+    }
+    return false;
+  }
+
+  /**
+   * Probe for an output stream having a capability; returns true
+   * iff the stream implements {@link StreamCapabilities} and its
+   * {@code hasCapabilities()} method returns true for the capability.
+   * @param out output stream
+   * @param capability capability to probe for
+   * @return true iff the stream declares that it supports the capability.
+   */
+  public static boolean hasCapability(OutputStream out, String capability) {
+    return objectHasCapability(out, capability);
+  }
+
+  /**
+   * Probe for an input stream having a capability; returns true
+   * iff the stream implements {@link StreamCapabilities} and its
+   * {@code hasCapabilities()} method returns true for the capability.
+   * @param out output stream
+   * @param capability capability to probe for
+   * @return true iff the stream declares that it supports the capability.
+   */
+  public static boolean hasCapability(InputStream out, String capability) {
+    return objectHasCapability(out, capability);
+  }
+
+}