NativeCollectorOnlyHandler implements CommandDispatcher, Closeable { @Override public ReadWriteBuffer onCall(Command command, ReadWriteBuffer parameter) throws IOException { Path p = null; if (null == command) { return null; } if (command.equals(GET_OUTPUT_PATH)) { p = output.getOutputFileForWrite(-1); } else if (command.equals(GET_OUTPUT_INDEX_PATH)) { p = output.getOutputIndexFileForWrite(-1); } else if (command.equals(GET_SPILL_PATH)) { p = output.getSpillFileForWrite(spillNumber++, -1); } else if (command.equals(GET_COMBINE_HANDLER)) { if (null == combinerHandler) { return null; } final ReadWriteBuffer result = new ReadWriteBuffer(8); result.writeLong(combinerHandler.getId()); return result; } else { throw new IOException("Illegal command: " + command.toString()); } if (p != null) { final ReadWriteBuffer result = new ReadWriteBuffer(); result.writeString(p.toUri().getPath()); return result; } else { throw new IOException("MapOutputFile can't allocate spill/output file"); } } protected  NativeCollectorOnlyHandler(TaskContext context, INativeHandler nativeHandler,
      BufferPusher<K, V> kvPusher, ICombineHandler combiner); static NativeCollectorOnlyHandler<K, V> create(TaskContext context); void collect(K key, V value, int partition); void flush(); @Override void close(); @Override ReadWriteBuffer onCall(Command command, ReadWriteBuffer parameter); static final String NAME; static final Command GET_OUTPUT_PATH; static final Command GET_OUTPUT_INDEX_PATH; static final Command GET_SPILL_PATH; static final Command GET_COMBINE_HANDLER; }
@Test public void testGetCombiner() throws IOException { this.handler = new NativeCollectorOnlyHandler(taskContext, nativeHandler, pusher, combiner); Mockito.when(combiner.getId()).thenReturn(100L); final ReadWriteBuffer result = handler.onCall( NativeCollectorOnlyHandler.GET_COMBINE_HANDLER, null); Assert.assertEquals(100L, result.readLong()); }
@Test public void testOnCall() throws IOException { this.handler = new NativeCollectorOnlyHandler(taskContext, nativeHandler, pusher, combiner); boolean thrown = false; try { handler.onCall(new Command(-1), null); } catch(final IOException e) { thrown = true; } Assert.assertTrue("exception thrown", thrown); final String expectedOutputPath = StringUtils.join(File.separator, new String[] {LOCAL_DIR, "output", "file.out"}); final String expectedOutputIndexPath = StringUtils.join(File.separator, new String[] {LOCAL_DIR, "output", "file.out.index"}); final String expectedSpillPath = StringUtils.join(File.separator, new String[] {LOCAL_DIR, "output", "spill0.out"}); final String outputPath = handler.onCall( NativeCollectorOnlyHandler.GET_OUTPUT_PATH, null).readString(); Assert.assertEquals(expectedOutputPath, outputPath); final String outputIndexPath = handler.onCall( NativeCollectorOnlyHandler.GET_OUTPUT_INDEX_PATH, null).readString(); Assert.assertEquals(expectedOutputIndexPath, outputIndexPath); final String spillPath = handler.onCall( NativeCollectorOnlyHandler.GET_SPILL_PATH, null).readString(); Assert.assertEquals(expectedSpillPath, spillPath); }