src/main/java/org/apache/aurora/codec/ThriftBinaryCodec.java

/**
 * Licensed 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.aurora.codec;

import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.lang.reflect.InvocationTargetException;
import java.util.zip.Deflater;
import java.util.zip.DeflaterOutputStream;
import java.util.zip.InflaterInputStream;

import javax.annotation.Nullable;

import org.apache.aurora.common.quantity.Amount;
import org.apache.aurora.common.quantity.Data;
import org.apache.thrift.TBase;
import org.apache.thrift.TDeserializer;
import org.apache.thrift.TException;
import org.apache.thrift.TSerializer;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.protocol.TProtocolFactory;
import org.apache.thrift.transport.TIOStreamTransport;
import org.apache.thrift.transport.TTransport;

import static java.util.Objects.requireNonNull;

/**
 * Codec that works for thrift objects.
 */
public final class ThriftBinaryCodec {

  /**
   * Protocol factory used for all thrift encoding and decoding.
   */
  public static final TProtocolFactory PROTOCOL_FACTORY = new TBinaryProtocol.Factory();

  private ThriftBinaryCodec() {
    // Utility class.
  }

  /**
   * Identical to {@link #decodeNonNull(Class, byte[])}, but allows for a null buffer.
   *
   * @param clazz Class to instantiate and deserialize to.
   * @param buffer Buffer to decode.
   * @param <T> Target type.
   * @return A populated message, or {@code null} if the buffer was {@code null}.
   * @throws CodingException If the message could not be decoded.
   */
  @Nullable
  public static <T extends TBase<T, ?>> T decode(Class<T> clazz, @Nullable byte[] buffer)
      throws CodingException {

    if (buffer == null) {
      return null;
    }
    return decodeNonNull(clazz, buffer);
  }

  /**
   * Decodes a binary-encoded byte array into a target type.
   *
   * @param clazz Class to instantiate and deserialize to.
   * @param buffer Buffer to decode.
   * @param <T> Target type.
   * @return A populated message.
   * @throws CodingException If the message could not be decoded.
   */
  public static <T extends TBase<T, ?>> T decodeNonNull(Class<T> clazz, byte[] buffer)
      throws CodingException {

    requireNonNull(clazz);
    requireNonNull(buffer);

    try {
      T t = newInstance(clazz);
      new TDeserializer(PROTOCOL_FACTORY).deserialize(t, buffer);
      return t;
    } catch (TException e) {
      throw new CodingException("Failed to deserialize thrift object.", e);
    }
  }

  /**
   * Identical to {@link #encodeNonNull(TBase)}, but allows for a null input.
   *
   * @param tBase Object to encode.
   * @return Encoded object, or {@code null} if the argument was {@code null}.
   * @throws CodingException If the object could not be encoded.
   */
  @Nullable
  public static byte[] encode(@Nullable TBase<?, ?> tBase) throws CodingException {
    if (tBase == null) {
      return null;
    }
    return encodeNonNull(tBase);
  }

  /**
   * Encodes a thrift object into a binary array.
   *
   * @param tBase Object to encode.
   * @return Encoded object.
   * @throws CodingException If the object could not be encoded.
   */
  public static byte[] encodeNonNull(TBase<?, ?> tBase) throws CodingException {
    requireNonNull(tBase);

    try {
      return new TSerializer(PROTOCOL_FACTORY).serialize(tBase);
    } catch (TException e) {
      throw new CodingException("Failed to serialize: " + tBase, e);
    }
  }

  // See http://www.zlib.net/zlib_how.html
  // "If the memory is available, buffers sizes on the order of 128K or 256K bytes should be used."
  private static final int DEFLATER_BUFFER_SIZE = Amount.of(256, Data.KB).as(Data.BYTES);

  // Empirical from microbenchmarks (assuming 20MiB/s writes to the replicated log and a large
  // de-duplicated Snapshot from a production environment).
  // TODO(ksweeney): Consider making this configurable.
  private static final int DEFLATE_LEVEL = 3;

  /**
   * Encodes a thrift object into a DEFLATE-compressed binary array.
   *
   * @param tBase Object to encode.
   * @return Deflated, encoded object.
   * @throws CodingException If the object could not be encoded.
   */
  public static byte[] deflateNonNull(TBase<?, ?> tBase) throws CodingException {
    requireNonNull(tBase);

    // NOTE: Buffering is needed here for performance.
    // There are actually 2 buffers in play here - the BufferedOutputStream prevents thrift from
    // causing a call to deflate() on every encoded primitive. The DeflaterOutputStream buffer
    // allows the underlying Deflater to operate on a larger chunk at a time without stopping to
    // copy the intermediate compressed output to outBytes.
    // See http://bugs.java.com/bugdatabase/view_bug.do?bug_id=4986239
    ByteArrayOutputStream outBytes = new ByteArrayOutputStream();
    TTransport transport = new TIOStreamTransport(
        new BufferedOutputStream(
            new DeflaterOutputStream(outBytes, new Deflater(DEFLATE_LEVEL), DEFLATER_BUFFER_SIZE),
            DEFLATER_BUFFER_SIZE));
    try {
      TProtocol protocol = PROTOCOL_FACTORY.getProtocol(transport);
      tBase.write(protocol);
      transport.close(); // calls finish() on the underlying stream, completing the compression
      return outBytes.toByteArray();
    } catch (TException e) {
      throw new CodingException("Failed to serialize: " + tBase, e);
    } finally {
      transport.close();
    }
  }

  /**
   * Decodes a thrift object from a DEFLATE-compressed byte array into a target type.
   *
   * @param clazz Class to instantiate and deserialize to.
   * @param buffer Compressed buffer to decode.
   * @return A populated message.
   * @throws CodingException If the message could not be decoded.
   */
  public static <T extends TBase<T, ?>> T inflateNonNull(Class<T> clazz, byte[] buffer)
      throws CodingException {

    requireNonNull(clazz);
    requireNonNull(buffer);

    T tBase = newInstance(clazz);
    TTransport transport = new TIOStreamTransport(
          new InflaterInputStream(new ByteArrayInputStream(buffer)));
    try {
      TProtocol protocol = PROTOCOL_FACTORY.getProtocol(transport);
      tBase.read(protocol);
      return tBase;
    } catch (TException e) {
      throw new CodingException("Failed to deserialize: " + e, e);
    } finally {
      transport.close();
    }
  }

  private static <T extends TBase<T, ?>> T newInstance(Class<T> clazz) throws CodingException {
    try {
      return clazz.getConstructor().newInstance();
    } catch (InvocationTargetException e) {
      throw new CodingException("Exception in constructor for target type: " + e, e);
    } catch (NoSuchMethodException e) {
      throw new CodingException(
          "No no-args constructor for target type: "
              + clazz
              + ". Did the thrift code generator change?",
          e);
    } catch (InstantiationException e) {
      throw new CodingException("Failed to instantiate target type.", e);
    } catch (IllegalAccessException e) {
      throw new CodingException("Failed to access constructor for target type.", e);
    }
  }

  /**
   * Thrown when serialization or deserialization failed.
   */
  public static class CodingException extends RuntimeException {
    public CodingException(String message) {
      super(message);
    }
    public CodingException(String msg, Throwable cause) {
      super(msg, cause);
    }
  }
}