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();
  }
@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://localhost:" + port)
                .withDatabase(DATABASE)
                .withChunkSize(100L)
                .withBucket("WriteTest")
                .withFilename("WriteTestData"));

    pipeline
        .apply("WithWriteFn", Create.of(intData))
        .apply(
            "WithWriteFnInternal",
            MongoDbGridFSIO.<Integer>write(
                    (output, outStream) -> {
                      // one byte per output
                      outStream.write(output.byteValue());
                    })
                .withUri("mongodb://localhost:" + port)
                .withDatabase(DATABASE)
                .withBucket("WriteTest")
                .withFilename("WriteTestIntData"));

    pipeline.run();

    MongoClient client = null;
    try {
      StringBuilder results = new StringBuilder();
      client = new MongoClient("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();
      }
    }
  }