java类org.apache.hadoop.io.DataInputBuffer的实例源码

/**
 * Set bytes into the passed Writable by calling its
 * {@link Writable#readFields(java.io.DataInput)}.
 * @param bytes serialized bytes
 * @param offset offset into array
 * @param length length of data
 * @param w An empty Writable (usually made by calling the null-arg
 * constructor).
 * @return The passed Writable after its readFields has been called fed
 * by the passed <code>bytes</code> array or IllegalArgumentException
 * if passed null or an empty <code>bytes</code> array.
 * @throws IOException e
 * @throws IllegalArgumentException
 */
public static Writable getWritable(final byte [] bytes, final int offset,
  final int length, final Writable w)
throws IOException {
  if (bytes == null || length <=0) {
    throw new IllegalArgumentException("Can't build a writable with empty " +
      "bytes array");
  }
  if (w == null) {
    throw new IllegalArgumentException("Writable cannot be null");
  }
  DataInputBuffer in = new DataInputBuffer();
  try {
    in.reset(bytes, offset, length);
    w.readFields(in);
    return w;
  } finally {
    in.close();
  }
}

@Test
public void testAMRMTokenIdentifier() throws IOException {
  ApplicationAttemptId appAttemptId = ApplicationAttemptId.newInstance(
      ApplicationId.newInstance(1, 1), 1);
  int masterKeyId = 1;

  AMRMTokenIdentifier token = new AMRMTokenIdentifier(appAttemptId, masterKeyId);

  AMRMTokenIdentifier anotherToken = new AMRMTokenIdentifier();
  byte[] tokenContent = token.getBytes();
  DataInputBuffer dib = new DataInputBuffer();
  dib.reset(tokenContent, tokenContent.length);
  anotherToken.readFields(dib);

  // verify the whole record equals with original record
  Assert.assertEquals("Token is not the same after serialization " +
      "and deserialization.", token, anotherToken);

  Assert.assertEquals("ApplicationAttemptId from proto is not the same with original token",
      anotherToken.getApplicationAttemptId(), appAttemptId);

  Assert.assertEquals("masterKeyId from proto is not the same with original token",
      anotherToken.getKeyId(), masterKeyId);
}

@Test(timeout=10000)
public void testLocalResourceStatusSerDe() throws Exception {
  LocalResourceStatus rsrcS = createLocalResourceStatus();
  assertTrue(rsrcS instanceof LocalResourceStatusPBImpl);
  LocalResourceStatusPBImpl rsrcPb = (LocalResourceStatusPBImpl) rsrcS;
  DataOutputBuffer out = new DataOutputBuffer();
  rsrcPb.getProto().writeDelimitedTo(out);
  DataInputBuffer in = new DataInputBuffer();
  in.reset(out.getData(), 0, out.getLength());
  LocalResourceStatusProto rsrcPbD =
    LocalResourceStatusProto.parseDelimitedFrom(in);
  assertNotNull(rsrcPbD);
  LocalResourceStatus rsrcD =
    new LocalResourceStatusPBImpl(rsrcPbD);

  assertEquals(rsrcS, rsrcD);
  assertEquals(createResource(), rsrcS.getResource());
  assertEquals(createResource(), rsrcD.getResource());
}

@Test(timeout=10000)
public void testLocalizerStatusSerDe() throws Exception {
  LocalizerStatus rsrcS = createLocalizerStatus();
  assertTrue(rsrcS instanceof LocalizerStatusPBImpl);
  LocalizerStatusPBImpl rsrcPb = (LocalizerStatusPBImpl) rsrcS;
  DataOutputBuffer out = new DataOutputBuffer();
  rsrcPb.getProto().writeDelimitedTo(out);
  DataInputBuffer in = new DataInputBuffer();
  in.reset(out.getData(), 0, out.getLength());
  LocalizerStatusProto rsrcPbD =
    LocalizerStatusProto.parseDelimitedFrom(in);
  assertNotNull(rsrcPbD);
  LocalizerStatus rsrcD =
    new LocalizerStatusPBImpl(rsrcPbD);

  assertEquals(rsrcS, rsrcD);
  assertEquals("localizer0", rsrcS.getLocalizerId());
  assertEquals("localizer0", rsrcD.getLocalizerId());
  assertEquals(createLocalResourceStatus(), rsrcS.getResourceStatus(0));
  assertEquals(createLocalResourceStatus(), rsrcD.getResourceStatus(0));
}

@Test(timeout=10000)
public void testLocalizerHeartbeatResponseSerDe() throws Exception {
  LocalizerHeartbeatResponse rsrcS = createLocalizerHeartbeatResponse();
  assertTrue(rsrcS instanceof LocalizerHeartbeatResponsePBImpl);
  LocalizerHeartbeatResponsePBImpl rsrcPb =
    (LocalizerHeartbeatResponsePBImpl) rsrcS;
  DataOutputBuffer out = new DataOutputBuffer();
  rsrcPb.getProto().writeDelimitedTo(out);
  DataInputBuffer in = new DataInputBuffer();
  in.reset(out.getData(), 0, out.getLength());
  LocalizerHeartbeatResponseProto rsrcPbD =
    LocalizerHeartbeatResponseProto.parseDelimitedFrom(in);
  assertNotNull(rsrcPbD);
  LocalizerHeartbeatResponse rsrcD =
    new LocalizerHeartbeatResponsePBImpl(rsrcPbD);

  assertEquals(rsrcS, rsrcD);
  assertEquals(createResource(), rsrcS.getResourceSpecs().get(0).getResource());
  assertEquals(createResource(), rsrcD.getResourceSpecs().get(0).getResource());
}

private <K> K serDeser(K conf) throws Exception {
  SerializationFactory factory = new SerializationFactory(CONF);
  Serializer<K> serializer =
    factory.getSerializer(GenericsUtil.getClass(conf));
  Deserializer<K> deserializer =
    factory.getDeserializer(GenericsUtil.getClass(conf));

  DataOutputBuffer out = new DataOutputBuffer();
  serializer.open(out);
  serializer.serialize(conf);
  serializer.close();

  DataInputBuffer in = new DataInputBuffer();
  in.reset(out.getData(), out.getLength());
  deserializer.open(in);
  K after = deserializer.deserialize(null);
  deserializer.close();
  return after;
}

public boolean nextRawKey(DataInputBuffer key) throws IOException {
  try {
    if (!positionToNextRecord(memDataIn)) {
      return false;
    }
    // Setup the key
    int pos = memDataIn.getPosition();
    byte[] data = memDataIn.getData();
    key.reset(data, pos, currentKeyLength);
    // Position for the next value
    long skipped = memDataIn.skip(currentKeyLength);
    if (skipped != currentKeyLength) {
      throw new IOException("Rec# " + recNo + 
          ": Failed to skip past key of length: " + 
          currentKeyLength);
    }

    // Record the byte
    bytesRead += currentKeyLength;
    return true;
  } catch (IOException ioe) {
    dumpOnError();
    throw ioe;
  }
}

/**
 * Write the given K,V to the cache. 
 * Write to memcache if space is available, else write to the filecache
 * @param key
 * @param value
 * @throws IOException
 */
public void write(DataInputBuffer key, DataInputBuffer value)
throws IOException {

  assert (key != null && value != null);

  if (fileCache.isActive()) {
    fileCache.write(key, value);
    return;
  }

  if (memCache.reserveSpace(key, value)) {
    memCache.write(key, value);
  } else {
    fileCache.activate();
    fileCache.write(key, value);
  }
}

public void append(DataInputBuffer key, DataInputBuffer value)
throws IOException {
  int keyLength = key.getLength() - key.getPosition();
  if (keyLength < 0) {
    throw new IOException("Negative key-length not allowed: " + keyLength + 
                          " for " + key);
  }

  int valueLength = value.getLength() - value.getPosition();
  if (valueLength < 0) {
    throw new IOException("Negative value-length not allowed: " + 
                          valueLength + " for " + value);
  }

  WritableUtils.writeVInt(out, keyLength);
  WritableUtils.writeVInt(out, valueLength);
  out.write(key.getData(), key.getPosition(), keyLength); 
  out.write(value.getData(), value.getPosition(), valueLength); 

  // Update bytes written
  decompressedBytesWritten += keyLength + valueLength + 
                  WritableUtils.getVIntSize(keyLength) + 
                  WritableUtils.getVIntSize(valueLength);
  ++numRecordsWritten;
}

public void append(DataInputBuffer key, DataInputBuffer value)
throws IOException {
  int keyLength = key.getLength() - key.getPosition();
  if (keyLength < 0) {
    throw new IOException("Negative key-length not allowed: " + keyLength + 
                          " for " + key);
  }

  int valueLength = value.getLength() - value.getPosition();
  if (valueLength < 0) {
    throw new IOException("Negative value-length not allowed: " + 
                          valueLength + " for " + value);
  }

  WritableUtils.writeVInt(out, keyLength);
  WritableUtils.writeVInt(out, valueLength);
  out.write(key.getData(), key.getPosition(), keyLength); 
  out.write(value.getData(), value.getPosition(), valueLength); 
}