MongoDbGridFSIO { public static Write<String> write() { return new AutoValue_MongoDbGridFSIO_Write.Builder<String>() .setConnectionConfiguration(ConnectionConfiguration.create()) .setWriteFn( (output, outStream) -> { outStream.write(output.getBytes(StandardCharsets.UTF_8)); outStream.write('\n'); }) .build(); }  static Read<String> read(); static Write<String> write(); static Write<T> write(WriteFn<T> fn);  }
@Test public void testWriteMessage() throws Exception { ArrayList<String> data = new ArrayList<>(100); ArrayList<Integer> intData = new ArrayList<>(100); for (int i = 0; i < 1000; i++) { data.add("Message " + i); } for (int i = 0; i < 100; i++) { intData.add(i); } pipeline .apply("String", Create.of(data)) .apply( "StringInternal", MongoDbGridFSIO.write() .withUri("mongodb: .withDatabase(DATABASE) .withChunkSize(100L) .withBucket("WriteTest") .withFilename("WriteTestData")); pipeline .apply("WithWriteFn", Create.of(intData)) .apply( "WithWriteFnInternal", MongoDbGridFSIO.<Integer>write( (output, outStream) -> { outStream.write(output.byteValue()); }) .withUri("mongodb: .withDatabase(DATABASE) .withBucket("WriteTest") .withFilename("WriteTestIntData")); pipeline.run(); Mongo client = null; try { StringBuilder results = new StringBuilder(); client = new Mongo("localhost", port); DB database = client.getDB(DATABASE); GridFS gridfs = new GridFS(database, "WriteTest"); List<GridFSDBFile> files = gridfs.find("WriteTestData"); assertTrue(files.size() > 0); for (GridFSDBFile file : files) { assertEquals(100, file.getChunkSize()); int l = (int) file.getLength(); try (InputStream ins = file.getInputStream()) { DataInputStream dis = new DataInputStream(ins); byte[] b = new byte[l]; dis.readFully(b); results.append(new String(b, StandardCharsets.UTF_8)); } } String dataString = results.toString(); for (int x = 0; x < 1000; x++) { assertTrue(dataString.contains("Message " + x)); } files = gridfs.find("WriteTestIntData"); boolean[] intResults = new boolean[100]; for (GridFSDBFile file : files) { int l = (int) file.getLength(); try (InputStream ins = file.getInputStream()) { DataInputStream dis = new DataInputStream(ins); byte[] b = new byte[l]; dis.readFully(b); for (byte aB : b) { intResults[aB] = true; } } } for (int x = 0; x < 100; x++) { assertTrue("Did not get a result for " + x, intResults[x]); } } finally { if (client != null) { client.close(); } } }