HBASE-3279 [rest] Filter for gzip content encoding that wraps both input and output side

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1039490 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -1190,6 +1190,8 @@ Release 0.90.0 - Unreleased
                just-released 3.3.2
    HBASE-3231  Update to zookeeper 3.3.2.
    HBASE-3273  Set the ZK default timeout to 3 minutes
+   HBASE-3279  [rest] Filter for gzip content encoding that wraps both input
+               and output side.
 
 
   NEW FEATURES

src/main/java/org/apache/hadoop/hbase/rest/Main.java

@@ -30,6 +30,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.rest.filter.GzipFilter;
 
 import java.util.List;
 import java.util.ArrayList;
@@ -37,7 +38,6 @@ import java.util.ArrayList;
 import org.mortbay.jetty.Server;
 import org.mortbay.jetty.servlet.Context;
 import org.mortbay.jetty.servlet.ServletHolder;
-import org.mortbay.servlet.GzipFilter;
 
 import com.sun.jersey.spi.container.servlet.ServletContainer;
 

src/main/java/org/apache/hadoop/hbase/rest/client/Response.java

@@ -73,18 +73,22 @@ public class Response {
     return headers;
   }
 
-  /**
-   * @return the value of the Location header
-   */
-  public String getLocation() {
+  public String getHeader(String key) {
     for (Header header: headers) {
-      if (header.getName().equals("Location")) {
+      if (header.getName().equalsIgnoreCase(key)) {
         return header.getValue();
       }
     }
     return null;
   }
 
+  /**
+   * @return the value of the Location header
+   */
+  public String getLocation() {
+    return getHeader("Location");
+  }
+
   /**
    * @return true if a response body was sent
    */

src/main/java/org/apache/hadoop/hbase/rest/filter/GZIPRequestStream.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.hbase.rest.filter;
+
+import java.io.IOException;
+import java.util.zip.GZIPInputStream;
+
+import javax.servlet.ServletInputStream;
+import javax.servlet.http.HttpServletRequest;
+
+public class GZIPRequestStream extends ServletInputStream
+{
+  private GZIPInputStream in;
+
+  public GZIPRequestStream(HttpServletRequest request) throws IOException {
+    this.in = new GZIPInputStream(request.getInputStream());
+  }
+
+  @Override
+  public int read() throws IOException {
+    return in.read();
+  }
+
+  @Override
+  public int read(byte[] b) throws IOException {
+    return in.read(b);
+  }
+
+  @Override
+  public int read(byte[] b, int off, int len) throws IOException {
+    return in.read(b, off, len);
+  }
+
+  @Override
+  public void close() throws IOException {
+    in.close();
+  }
+}

src/main/java/org/apache/hadoop/hbase/rest/filter/GZIPRequestWrapper.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.hbase.rest.filter;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+
+import javax.servlet.ServletInputStream;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletRequestWrapper;
+
+public class GZIPRequestWrapper extends HttpServletRequestWrapper {
+  private ServletInputStream is;
+  private BufferedReader reader;
+
+  public GZIPRequestWrapper(HttpServletRequest request) throws IOException {
+    super(request);
+    this.is = new GZIPRequestStream(request);
+    this.reader = new BufferedReader(new InputStreamReader(this.is));
+  }
+
+  @Override
+  public ServletInputStream getInputStream() throws IOException {
+    return is;
+  }
+
+  @Override
+  public BufferedReader getReader() throws IOException {
+    return reader;
+  }
+}

src/main/java/org/apache/hadoop/hbase/rest/filter/GZIPResponseStream.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.hbase.rest.filter;
+
+import java.io.IOException;
+import java.util.zip.GZIPOutputStream;
+
+import javax.servlet.ServletOutputStream;
+import javax.servlet.http.HttpServletResponse;
+
+public class GZIPResponseStream extends ServletOutputStream
+{
+  private HttpServletResponse response;
+  private GZIPOutputStream out;
+
+  public GZIPResponseStream(HttpServletResponse response) throws IOException {
+    this.response = response;
+    this.out = new GZIPOutputStream(response.getOutputStream());
+    response.addHeader("Content-Encoding", "gzip");
+  }
+
+  public void resetBuffer() {
+    if (out != null && !response.isCommitted()) {
+      response.setHeader("Content-Encoding", null);
+    }
+    out = null;
+  }
+
+  @Override
+  public void write(int b) throws IOException {
+    out.write(b);
+  }
+
+  @Override
+  public void write(byte[] b) throws IOException {
+    out.write(b);
+  }
+
+  @Override
+  public void write(byte[] b, int off, int len) throws IOException {
+    out.write(b, off, len);
+  }
+
+  @Override
+  public void close() throws IOException {
+    finish();
+    out.close();
+  }
+
+  @Override
+  public void flush() throws IOException {
+    out.flush();
+  }
+
+  public void finish() throws IOException {
+    out.finish();
+  }
+}

src/main/java/org/apache/hadoop/hbase/rest/filter/GZIPResponseWrapper.java

@@ -0,0 +1,144 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.hbase.rest.filter;
+
+import java.io.IOException;
+import java.io.PrintWriter;
+
+import javax.servlet.ServletOutputStream;
+import javax.servlet.http.HttpServletResponse;
+import javax.servlet.http.HttpServletResponseWrapper;
+
+public class GZIPResponseWrapper extends HttpServletResponseWrapper {
+  private HttpServletResponse response;
+  private GZIPResponseStream os;
+  private PrintWriter writer;
+  private boolean compress = true;
+
+  public GZIPResponseWrapper(HttpServletResponse response) {
+    super(response);
+    this.response = response;
+  }
+
+  @Override
+  public void setStatus(int status) {
+    if (status < 200 || status >= 300) {
+      compress = false;
+    }
+  }
+
+  @Override
+  public void addHeader(String name, String value) {
+    if (!"content-length".equalsIgnoreCase(name)) {
+      super.addHeader(name, value);
+    }
+  }
+
+  @Override
+  public void setContentLength(int length) {
+    // do nothing
+  }
+
+  @Override
+  public void setIntHeader(String name, int value) {
+    if (!"content-length".equalsIgnoreCase(name)) {
+      super.setIntHeader(name, value);
+    }
+  }
+
+  @Override
+  public void setHeader(String name, String value) {
+    if (!"content-length".equalsIgnoreCase(name)) {
+      super.setHeader(name, value);
+    }
+  }
+
+  @Override
+  public void flushBuffer() throws IOException {
+    if (writer != null) {
+      writer.flush();
+    }
+    if (os != null) {
+      os.finish();
+    } else {
+      getResponse().flushBuffer();
+    }
+  }
+
+  @Override
+  public void reset() {
+    super.reset();
+    if (os != null) {
+      os.resetBuffer();
+    }
+    writer = null;
+    os = null;
+    compress = true;
+  }
+
+  @Override
+  public void resetBuffer() {
+    super.resetBuffer();
+    if (os != null) {
+      os.resetBuffer();
+    }
+    writer = null;
+    os = null;
+  }
+
+  @Override
+  public void sendError(int status, String msg) throws IOException {
+    resetBuffer();
+    super.sendError(status, msg);
+  }
+
+  @Override
+  public void sendError(int status) throws IOException {
+    resetBuffer();
+    super.sendError(status);
+  }
+
+  @Override
+  public void sendRedirect(String location) throws IOException {
+    resetBuffer();
+    super.sendRedirect(location);
+  }
+
+  @Override
+  public ServletOutputStream getOutputStream() throws IOException {
+    if (!response.isCommitted() && compress) {
+      if (os == null) {
+        os = new GZIPResponseStream(response);
+      }
+      return os;
+    } else {
+      return response.getOutputStream();
+    }
+  }
+
+  @Override
+  public PrintWriter getWriter() throws IOException {
+    if (writer == null) {
+      writer = new PrintWriter(getOutputStream());
+    }
+    return writer;
+  }
+}

src/main/java/org/apache/hadoop/hbase/rest/filter/GzipFilter.java

@@ -0,0 +1,57 @@
+package org.apache.hadoop.hbase.rest.filter;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.StringTokenizer;
+
+import javax.servlet.Filter;
+import javax.servlet.FilterChain;
+import javax.servlet.FilterConfig;
+import javax.servlet.ServletException;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+public class GzipFilter implements Filter {
+  private Set<String> mimeTypes = new HashSet<String>();
+
+  @Override
+  public void init(FilterConfig filterConfig) throws ServletException {
+    String s = filterConfig.getInitParameter("mimeTypes");
+    if (s != null) {
+      StringTokenizer tok = new StringTokenizer(s, ",", false);
+      while (tok.hasMoreTokens()) {
+        mimeTypes.add(tok.nextToken());
+      }
+    }
+  }
+
+  @Override
+  public void destroy() {
+  }
+
+  @Override
+  public void doFilter(ServletRequest req, ServletResponse rsp,
+      FilterChain chain) throws IOException, ServletException {
+    HttpServletRequest request = (HttpServletRequest)req;
+    HttpServletResponse response = (HttpServletResponse)rsp;
+    String contentEncoding = request.getHeader("content-encoding");
+    String acceptEncoding = request.getHeader("accept-encoding");
+    String contentType = request.getHeader("content-type");
+    if ((contentEncoding != null) &&
+        (contentEncoding.toLowerCase().indexOf("gzip") > -1)) {
+      request = new GZIPRequestWrapper(request);
+    }
+    if (((acceptEncoding != null) &&
+          (acceptEncoding.toLowerCase().indexOf("gzip") > -1)) ||
+        ((contentType != null) && mimeTypes.contains(contentType))) {
+      response = new GZIPResponseWrapper(response);
+    }
+    chain.doFilter(request, response);
+    if ((response instanceof GZIPResponseWrapper)) {
+      ((GZIPResponseStream)response.getOutputStream()).finish();
+    }
+  }
+}

src/test/java/org/apache/hadoop/hbase/rest/HBaseRESTTestingUtility.java

@@ -22,6 +22,7 @@ package org.apache.hadoop.hbase.rest;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.rest.filter.GzipFilter;
 import org.apache.hadoop.util.StringUtils;
 import org.mortbay.jetty.Server;
 import org.mortbay.jetty.servlet.Context;
@@ -66,11 +67,12 @@ public class HBaseRESTTestingUtility {
       // set up context
     Context context = new Context(server, "/", Context.SESSIONS);
     context.addServlet(sh, "/*");
+    context.addFilter(GzipFilter.class, "/*", 0);
       // start the server
     server.start();
       // get the port
     testServletPort = server.getConnectors()[0].getLocalPort();
-    
+
     LOG.info("started " + server.getClass().getName() + " on port " + 
       testServletPort);
   }

src/test/java/org/apache/hadoop/hbase/rest/TestGzipFilter.java

@@ -0,0 +1,132 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.hbase.rest;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.util.zip.GZIPInputStream;
+import java.util.zip.GZIPOutputStream;
+
+import org.apache.commons.httpclient.Header;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.HBaseAdmin;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.rest.client.Client;
+import org.apache.hadoop.hbase.rest.client.Cluster;
+import org.apache.hadoop.hbase.rest.client.Response;
+import org.apache.hadoop.hbase.util.Bytes;
+
+import static org.junit.Assert.*;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class TestGzipFilter {
+  private static final String TABLE = "TestGzipFilter";
+  private static final String CFA = "a";
+  private static final String COLUMN_1 = CFA + ":1";
+  private static final String COLUMN_2 = CFA + ":2";
+  private static final String ROW_1 = "testrow1";
+  private static final byte[] VALUE_1 = Bytes.toBytes("testvalue1");
+
+  private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
+  private static final HBaseRESTTestingUtility REST_TEST_UTIL =
+    new HBaseRESTTestingUtility();
+  private static Client client;
+
+  @BeforeClass
+  public static void setUpBeforeClass() throws Exception {
+    TEST_UTIL.startMiniCluster(3);
+    REST_TEST_UTIL.startServletContainer(TEST_UTIL.getConfiguration());
+    client = new Client(new Cluster().add("localhost",
+      REST_TEST_UTIL.getServletPort()));
+    HBaseAdmin admin = TEST_UTIL.getHBaseAdmin();
+    if (admin.tableExists(TABLE)) {
+      return;
+    }
+    HTableDescriptor htd = new HTableDescriptor(TABLE);
+    htd.addFamily(new HColumnDescriptor(CFA));
+    admin.createTable(htd);
+  }
+
+  @AfterClass
+  public static void tearDownAfterClass() throws Exception {
+    REST_TEST_UTIL.shutdownServletContainer();
+    TEST_UTIL.shutdownMiniCluster();
+  }
+
+  @Test
+  public void testGzipFilter() throws Exception {
+    String path = "/" + TABLE + "/" + ROW_1 + "/" + COLUMN_1;
+
+    ByteArrayOutputStream bos = new ByteArrayOutputStream();
+    GZIPOutputStream os = new GZIPOutputStream(bos);
+    os.write(VALUE_1);
+    os.close();
+    byte[] value_1_gzip = bos.toByteArray();
+
+    // input side filter
+
+    Header[] headers = new Header[2];
+    headers[0] = new Header("Content-Type", Constants.MIMETYPE_BINARY);
+    headers[1] = new Header("Content-Encoding", "gzip");
+    Response response = client.put(path, headers, value_1_gzip);
+    assertEquals(response.getCode(), 200);
+
+    HTable table = new HTable(TEST_UTIL.getConfiguration(), TABLE);
+    Get get = new Get(Bytes.toBytes(ROW_1));
+    get.addColumn(Bytes.toBytes(CFA), Bytes.toBytes("1"));
+    Result result = table.get(get);
+    byte[] value = result.getValue(Bytes.toBytes(CFA), Bytes.toBytes("1"));
+    assertNotNull(value);
+    assertTrue(Bytes.equals(value, VALUE_1));
+
+    // output side filter
+
+    headers[0] = new Header("Accept", Constants.MIMETYPE_BINARY);
+    headers[1] = new Header("Accept-Encoding", "gzip");
+    response = client.get(path, headers);
+    assertEquals(response.getCode(), 200);
+    ByteArrayInputStream bis = new ByteArrayInputStream(response.getBody());
+    GZIPInputStream is = new GZIPInputStream(bis);
+    value = new byte[VALUE_1.length];
+    is.read(value, 0, VALUE_1.length);
+    assertTrue(Bytes.equals(value, VALUE_1));
+    is.close();
+  }
+
+  @Test
+  public void testErrorNotGzipped() throws Exception {
+    String path = "/" + TABLE + "/" + ROW_1 + "/" + COLUMN_2;
+    Header[] headers = new Header[2];
+    headers[0] = new Header("Accept", Constants.MIMETYPE_BINARY);
+    headers[1] = new Header("Accept-Encoding", "gzip");
+    Response response = client.get(path, headers);
+    assertEquals(response.getCode(), 404);
+    String contentEncoding = response.getHeader("Content-Encoding");
+    assertTrue(contentEncoding == null || !contentEncoding.contains("gzip"));
+  }
+}