Chain.java 文件源码

@SuppressWarnings({"unchecked"})
public void collect(K key, V value) throws IOException {
  if (nextMapperIndex < mappers.size()) {
    // there is a next mapper in chain

    // only need to ser/deser if there is next mapper in the chain
    if (keySerialization != null) {
      key = makeCopyForPassByValue(keySerialization, key);
      value = makeCopyForPassByValue(valueSerialization, value);
    }

    // gets ser/deser and mapper of next in chain
    Serialization nextKeySerialization =
      mappersKeySerialization.get(nextMapperIndex);
    Serialization nextValueSerialization =
      mappersValueSerialization.get(nextMapperIndex);
    Mapper nextMapper = mappers.get(nextMapperIndex);

    // invokes next mapper in chain
    nextMapper.map(key, value,
                   new ChainOutputCollector(nextMapperIndex,
                                            nextKeySerialization,
                                            nextValueSerialization,
                                            output, reporter),
                   reporter);
  } else {
    // end of chain, user real output collector
    output.collect(key, value);
  }
}