public BlockingDirectBinaryEncoder(OutputStream out) {
    super(out);
    this.buffers = new ArrayList<>();
    this.stashedBuffers = new ArrayDeque<>();
    this.blockItemCounts = new ArrayDeque<>();
  }
@Test
  void blockingDirectBinaryEncoder() throws IOException, NoSuchAlgorithmException {
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    BinaryEncoder encoder = EncoderFactory.get().blockingDirectBinaryEncoder(baos, null);

    // This is needed because there is no BlockingDirectBinaryEncoder
    // BinaryMessageWriter
    // available out of the box
    encoder.writeFixed(new byte[] { (byte) 0xC3, (byte) 0x01 });
    encoder.writeFixed(SchemaNormalization.parsingFingerprint("CRC-64-AVRO", TestRecordWithMapsAndArrays.SCHEMA$));

    // Array
    this.writeToArray(encoder, new int[] { 1, 2, 3, 4, 5 });

    // Map
    writeToMap(encoder, new long[] { 1L, 2L, 3L, 4L, 5L });

    // Nested Array

    encoder.writeArrayStart();
    encoder.setItemCount(2);
    this.writeToArray(encoder, new int[] { 1, 2 });
    this.writeToArray(encoder, new int[] { 3, 4, 5 });
    encoder.writeArrayEnd();

    // Nested Map

    encoder.writeMapStart();
    encoder.setItemCount(2);
    encoder.writeString("first");
    this.writeToMap(encoder, new long[] { 1L, 2L });
    encoder.writeString("second");
    this.writeToMap(encoder, new long[] { 3L, 4L, 5L });
    encoder.writeMapEnd();

    // Read

    encoder.flush();

    BinaryMessageDecoder<TestRecordWithMapsAndArrays> decoder = TestRecordWithMapsAndArrays.getDecoder();
    TestRecordWithMapsAndArrays r = decoder.decode(baos.toByteArray());

    assertThat(r.getArr(), is(Arrays.asList("1", "2", "3", "4", "5")));
    Map<String, Long> map = r.getMap();
    assertThat(map.size(), is(5));
    for (long i = 1; i <= 5; i++) {
      assertThat(map.get(Long.toString(i)), is(i));
    }

    assertThat(r.getNestedArr(), is(Arrays.asList(Arrays.asList("1", "2"), Arrays.asList("3", "4", "5"))));

    Map<String, Map<String, Long>> nestedMap = r.getNestedMap();
    assertThat(nestedMap.size(), is(2));

    assertThat(nestedMap.get("first").size(), is(2));
    assertThat(nestedMap.get("first").get("1"), is(1L));
    assertThat(nestedMap.get("first").get("2"), is(2L));

    assertThat(nestedMap.get("second").size(), is(3));
    assertThat(nestedMap.get("second").get("3"), is(3L));
    assertThat(nestedMap.get("second").get("4"), is(4L));
    assertThat(nestedMap.get("second").get("5"), is(5L));
  }