import org.fusesource.hawtbuf.ByteArrayInputStream; //导入依赖的package包/类
@Test
public void invokeShouldReturnGzipContent() throws Exception {
	MvcResult result = this.mvc.perform(get("/heapdump")).andExpect(status().isOk())
			.andReturn();
	byte[] bytes = result.getResponse().getContentAsByteArray();
	GZIPInputStream stream = new GZIPInputStream(new ByteArrayInputStream(bytes));
	byte[] uncompressed = FileCopyUtils.copyToByteArray(stream);
	assertThat(uncompressed).isEqualTo("HEAPDUMP".getBytes());
}
 

import org.fusesource.hawtbuf.ByteArrayInputStream; //导入依赖的package包/类
public void invokeOnTrigger(final Integer queryTimeout, final String query, final boolean incomingFlowFile, final boolean setQueryProperty)
    throws InitializationException, ClassNotFoundException, SQLException, IOException {

    if (queryTimeout != null) {
        runner.setProperty(ExecuteOracleSQL.QUERY_TIMEOUT, queryTimeout.toString() + " secs");
    }

    // remove previous test database, if any
    final File dbLocation = new File(DB_LOCATION);
    dbLocation.delete();

    // load test data to database
    final Connection con = ((DBCPService) runner.getControllerService("dbcp")).getConnection();
    TestJdbcHugeStream.loadTestData2Database(con, 100, 200, 100);
    LOGGER.info("test data loaded");

    // ResultSet size will be 1x200x100 = 20 000 rows
    // because of where PER.ID = ${person.id}
    final int nrOfRows = 20000;

    if (incomingFlowFile) {
        // incoming FlowFile content is not used, but attributes are used
        final Map<String, String> attributes = new HashMap<>();
        attributes.put("person.id", "10");
        if (!setQueryProperty) {
            runner.enqueue(query.getBytes(), attributes);
        } else {
            runner.enqueue("Hello".getBytes(), attributes);
        }
    }

    if(setQueryProperty) {
        runner.setProperty(ExecuteOracleSQL.SQL_SELECT_QUERY, query);
    }

    runner.run();
    runner.assertAllFlowFilesTransferred(ExecuteOracleSQL.SUCCESS, 1);

    final List<MockFlowFile> flowfiles = runner.getFlowFilesForRelationship(ExecuteOracleSQL.SUCCESS);
    final InputStream in = new ByteArrayInputStream(flowfiles.get(0).toByteArray());
    final DatumReader<GenericRecord> datumReader = new GenericDatumReader<>();
    try (DataFileStream<GenericRecord> dataFileReader = new DataFileStream<>(in, datumReader)) {
        GenericRecord record = null;
        long recordsFromStream = 0;
        while (dataFileReader.hasNext()) {
            // Reuse record object by passing it to next(). This saves us from
            // allocating and garbage collecting many objects for files with
            // many items.
            record = dataFileReader.next(record);
            recordsFromStream += 1;
        }

        LOGGER.info("total nr of records from stream: " + recordsFromStream);
        assertEquals(nrOfRows, recordsFromStream);
    }
}
 

import org.fusesource.hawtbuf.ByteArrayInputStream; //导入依赖的package包/类
private Map<String, Object> unmarsallProperties(Buffer marshalledProperties) throws IOException {
    return MarshallingSupport.unmarshalPrimitiveMap(new DataInputStream(new ByteArrayInputStream(marshalledProperties)));
}