001/* 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, 013 * software distributed under the License is distributed on an 014 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 015 * KIND, either express or implied. See the License for the 016 * specific language governing permissions and limitations 017 * under the License. 018 */ 019package org.apache.reef.examples.suspend; 020 021import org.apache.hadoop.io.Writable; 022import org.apache.reef.wake.remote.Codec; 023import org.apache.reef.wake.remote.exception.RemoteRuntimeException; 024 025import java.io.*; 026import java.util.logging.Level; 027import java.util.logging.Logger; 028 029/** 030 * Codec for Hadoop Writable object serialization. 031 * 032 * @param <T> Class derived from Hadoop Writable. 033 */ 034public class ObjectWritableCodec<T extends Writable> implements Codec<T> { 035 036 /** 037 * Standard Java logger. 038 */ 039 private static final Logger LOG = Logger.getLogger(ObjectWritableCodec.class.getName()); 040 041 /** 042 * we need it to invoke the class constructor. 043 */ 044 private final Class<? extends T> writableClass; 045 046 /** 047 * Create a new codec for Hadoop Writables. 048 * 049 * @param clazz we need it to invoke the class constructor. 050 */ 051 public ObjectWritableCodec(final Class<? extends T> clazz) { 052 this.writableClass = clazz; 053 } 054 055 /** 056 * Encodes Hadoop Writable object into a byte array. 057 * 058 * @param writable the object to encode. 059 * @return serialized object as byte array. 060 * @throws RemoteRuntimeException if serialization fails. 061 */ 062 @Override 063 public byte[] encode(final T writable) { 064 try (final ByteArrayOutputStream bos = new ByteArrayOutputStream(); 065 final DataOutputStream dos = new DataOutputStream(bos)) { 066 writable.write(dos); 067 return bos.toByteArray(); 068 } catch (final IOException ex) { 069 LOG.log(Level.SEVERE, "Cannot encode object " + writable, ex); 070 throw new RemoteRuntimeException(ex); 071 } 072 } 073 074 /** 075 * Decode Hadoop Writable object from a byte array. 076 * 077 * @param buffer serialized version of the Writable object (as a byte array). 078 * @return a Writable object. 079 * @throws RemoteRuntimeException if deserialization fails. 080 */ 081 @Override 082 public T decode(final byte[] buffer) { 083 try (final ByteArrayInputStream bis = new ByteArrayInputStream(buffer); 084 final DataInputStream dis = new DataInputStream(bis)) { 085 final T writable = this.writableClass.newInstance(); 086 writable.readFields(dis); 087 return writable; 088 } catch (final IOException | InstantiationException | IllegalAccessException ex) { 089 LOG.log(Level.SEVERE, "Cannot decode class " + this.writableClass, ex); 090 throw new RemoteRuntimeException(ex); 091 } 092 } 093}