LogParser extends Parser<byte[], KeyValPair<String, String>> { @Override public void beginWindow(long windowId) { super.beginWindow(windowId); parsedOutputCount = 0; }  @Override Object convert(byte[] tuple); @Override KeyValPair<String, String> processErrorTuple(byte[] bytes); @Override void beginWindow(long windowId); @Override void setup(Context.OperatorContext context); @Override void processTuple(byte[] inputTuple); void emitError(String tuple, String errorMsg); void setupLog(); void setLogFileFormat(String logFileFormat); String geLogFileFormat(); String getEncoding(); void setEncoding(String encoding); LogSchemaDetails getLogSchemaDetails(); void setLogSchemaDetails(LogSchemaDetails logSchemaDetails); Class<?> getClazz(); void setClazz(Class<?> clazz); transient DefaultOutputPort<Object> parsedOutput; }
@Test public void TestEmptyInput() { String tuple = ""; logParser.beginWindow(0); logParser.in.process(tuple.getBytes()); logParser.endWindow(); Assert.assertEquals(0, pojoPort.collectedTuples.size()); Assert.assertEquals(1, error.collectedTuples.size()); }
@Test public void TestNullInput() { logParser.beginWindow(0); logParser.in.process(null); logParser.endWindow(); Assert.assertEquals(0, pojoPort.collectedTuples.size()); Assert.assertEquals(1, error.collectedTuples.size()); }