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, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018
019package org.apache.hadoop.io.serializer;
020
021import java.io.DataInputStream;
022import java.io.DataOutputStream;
023import java.io.IOException;
024import java.io.InputStream;
025import java.io.OutputStream;
026import java.util.Map;
027
028import org.apache.hadoop.classification.InterfaceAudience;
029import org.apache.hadoop.classification.InterfaceStability;
030import org.apache.hadoop.conf.Configuration;
031import org.apache.hadoop.conf.Configured;
032import org.apache.hadoop.io.Writable;
033import org.apache.hadoop.util.ReflectionUtils;
034
035/**
036 * A {@link Serialization} for {@link Writable}s that delegates to
037 * {@link Writable#write(java.io.DataOutput)} and
038 * {@link Writable#readFields(java.io.DataInput)}.
039 */
040@InterfaceAudience.Public
041@InterfaceStability.Evolving
042public class WritableSerialization extends Configured
043        implements Serialization<Writable> {
044  static class WritableDeserializer extends Configured
045        implements Deserializer<Writable> {
046
047    private Class<?> writableClass;
048    private DataInputStream dataIn;
049    
050    public WritableDeserializer(Configuration conf, Class<?> c) {
051      setConf(conf);
052      this.writableClass = c;
053    }
054    
055    @Override
056    public void open(InputStream in) {
057      if (in instanceof DataInputStream) {
058        dataIn = (DataInputStream) in;
059      } else {
060        dataIn = new DataInputStream(in);
061      }
062    }
063    
064    @Override
065    public Writable deserialize(Writable w) throws IOException {
066      Writable writable;
067      if (w == null) {
068        writable 
069          = (Writable) ReflectionUtils.newInstance(writableClass, getConf());
070      } else {
071        writable = w;
072      }
073      writable.readFields(dataIn);
074      return writable;
075    }
076
077    @Override
078    public void close() throws IOException {
079      dataIn.close();
080    }
081    
082  }
083  
084  static class WritableSerializer extends Configured implements
085        Serializer<Writable> {
086    
087    private DataOutputStream dataOut;
088    
089    @Override
090    public void open(OutputStream out) {
091      if (out instanceof DataOutputStream) {
092        dataOut = (DataOutputStream) out;
093      } else {
094        dataOut = new DataOutputStream(out);
095      }
096    }
097
098    @Override
099    public void serialize(Writable w) throws IOException {
100      w.write(dataOut);
101    }
102
103    @Override
104    public void close() throws IOException {
105      dataOut.close();
106    }
107
108  }
109
110  @InterfaceAudience.Private
111  @Override
112  public boolean accept(Class<?> c) {
113    return Writable.class.isAssignableFrom(c);
114  }
115
116  @InterfaceAudience.Private
117  @Override
118  public Serializer<Writable> getSerializer(Class<Writable> c) {
119    return new WritableSerializer();
120  }
121  
122  @InterfaceAudience.Private
123  @Override
124  public Deserializer<Writable> getDeserializer(Class<Writable> c) {
125    return new WritableDeserializer(getConf(), c);
126  }
127
128}