Java 类com.google.protobuf.DynamicMessage 实例源码

/**
 * Handle all the logic leading to the decoding of a Protobuf-encoded binary given a schema file path.
 * @param schema  Schema used to decode the binary data
 * @param messageType   Type of Protobuf Message
 * @param encodedData   Encoded data source
 * @return  A JSON representation of the data, contained in a Java String
 * @throws InvalidProtocolBufferException   Thrown when an error occurs during the encoding of the decoded data into JSON
 * @throws Descriptors.DescriptorValidationException    Thrown when the schema is invalid
 * @throws UnknownMessageTypeException  Thrown when the given message type is not contained in the schema
 * @throws MessageDecodingException Thrown when an error occurs during the binary decoding
 * @throws SchemaLoadingException   Thrown when an error occurs while reading the schema file
 */
public static String decodeProtobuf(DynamicSchema schema, String messageType, InputStream encodedData) throws InvalidProtocolBufferException, Descriptors.DescriptorValidationException, UnknownMessageTypeException, MessageDecodingException, SchemaLoadingException {
    Descriptors.Descriptor descriptor;
    DynamicMessage message;

    descriptor = schema.getMessageDescriptor(messageType);

    if (descriptor == null) {
        throw new UnknownMessageTypeException(messageType);
    }

    try {
        message = DynamicMessage.parseFrom(descriptor, encodedData);
    } catch (IOException e) {
        throw new MessageDecodingException(e);
    }

    return JSONMapper.toJSON(message);
}

@Override
public DynamicMessage encode(RowData rowData) {
    RowDataValueSource value = new RowDataValueSource();
    DynamicMessage.Builder builder = DynamicMessage.newBuilder(messageType);
    for (int i = 0; i < fields.length; i++) {
        value.bind(rowDef.getFieldDef(i), rowData);
        if (value.isNull()) {
            if (nullFields != null) {
                FieldDescriptor nullField = nullFields[i];
                if (nullField != null) {
                    builder.setField(nullField, Boolean.TRUE);
                }
            }
        }
        else {
            conversions[i].setValue(builder, fields[i], value);
        }
    }
    return builder.build();
}

@Override
public void decode(DynamicMessage msg, RowData rowData) {
    Object[] objects = new Object[fields.length];
    for (FieldDescriptor field : msg.getAllFields().keySet()) {
        Integer columnIndex = columnIndexesByField.get(field);
        if (columnIndex != null) {
            objects[columnIndex] = conversions[columnIndex].getValue(msg, field);
        }
        else {
            Integer nullIndex = nullableIndexesByField.get(field);
            if (nullIndex != null) {
                // TODO: It's already null, because we aren't
                // handling defaults yet.
                objects[nullIndex] = null;
            }
        }
    }
    if (rowData.getBytes() == null) {
        rowData.reset(new byte[RowData.CREATE_ROW_INITIAL_SIZE]);
    }
    rowData.createRow(rowDef, objects, true);
}

@Override
public Row decode(DynamicMessage msg) {
    boolean first = true;
    Row row = null;
    for (FieldDescriptor field : msg.getAllFields().keySet()) {
        ProtobufRowConverter tableConverter = 
            tableConvertersByField.get(field);
        if (tableConverter != null) {
            assert first;
            first = false;
            row = tableConverter.decode((DynamicMessage)msg.getField(field));
        }
    }
    assert !first;
    return row;
}

@Override
public DynamicMessage encode(Row row) {
    DynamicMessage.Builder builder = DynamicMessage.newBuilder(messageType);
    for (int i = 0; i < fields.length; i++) {
        if (row.value(i).isNull()) {
            if (nullFields != null) {
                FieldDescriptor nullField = nullFields[i];
                if (nullField != null) {
                    builder.setField(nullField, Boolean.TRUE);
                }
            }
        } else {
            conversions[i].setValue(builder, fields[i], row.value(i));
        }
    }
    return builder.build();
}

@Override
public Row decode(DynamicMessage msg) {
    Object[] objects = new Object[fields.length];
    for (FieldDescriptor field : msg.getAllFields().keySet()) {
        Integer columnIndex = columnIndexesByField.get(field);
        if (columnIndex != null) {
            objects[columnIndex] = conversions[columnIndex].getValue(msg, field);
        }
        else {
            Integer nullIndex = nullableIndexesByField.get(field);
            if (nullIndex != null) {
                // TODO: It's already null, because we aren't
                // handling defaults yet.
                objects[nullIndex] = null;
            }
        }
    }
    ValuesHolderRow row = new ValuesHolderRow (rowType, objects);
    return row;
}

@Override 
public Row expandRow (FDBStore store, Session session,
                        FDBStoreData storeData, Schema schema) {
    ensureRowConverter();
    DynamicMessage msg;
    try {
        msg = DynamicMessage.parseFrom(rowConverter.getMessageType(), storeData.rawValue);
    } catch (InvalidProtocolBufferException ex) {
        ProtobufReadException nex = new ProtobufReadException(rowDataConverter.getMessageType().getName(), ex.getMessage());
        nex.initCause(ex);
        throw nex;
    }
    Row row = rowConverter.decode(msg);
    row = overlayBlobData(row.rowType(), row, store, session);
    return row;
}

@Override
public void write(Record record) throws IOException, DataGeneratorException {
  if (closed) {
    throw new IOException("generator has been closed");
  }
  DynamicMessage message = ProtobufTypeUtil.sdcFieldToProtobufMsg(
      record,
      descriptor,
      messageTypeToExtensionMap,
      defaultValueMap
  );
  if (isDelimited) {
    message.writeDelimitedTo(outputStream);
  } else {
    message.writeTo(outputStream);
  }
}

public ProtobufDataParser(
    ProtoConfigurableEntity.Context context,
    String messageId,
    Descriptors.Descriptor descriptor,
    Map<String, Set<Descriptors.FieldDescriptor>> messageTypeToExtensionMap,
    ExtensionRegistry extensionRegistry,
    InputStream inputStream,
    String readerOffset,
    int maxObjectLength,
    boolean isDelimited
) throws IOException, Descriptors.DescriptorValidationException, DataParserException {
  this.context = context;
  this.inputStream = new OverrunInputStream(inputStream, maxObjectLength, true);
  this.messageId = messageId;
  this.messageTypeToExtensionMap = messageTypeToExtensionMap;
  this.extensionRegistry = extensionRegistry;
  this.descriptor = descriptor;
  this.builder = DynamicMessage.newBuilder(descriptor);
  this.isDelimited = isDelimited;

  // skip to the required location
  if (readerOffset != null && !readerOffset.isEmpty() && !readerOffset.equals("0")) {
    int offset = Integer.parseInt(readerOffset);
    this.inputStream.skip(offset);
  }
}

private static void handleUnknownFields(
    Record record,
    String fieldPath,
    DynamicMessage.Builder builder
) throws IOException {
  String path = fieldPath.isEmpty() ? FORWARD_SLASH : fieldPath;
  String attribute = record.getHeader().getAttribute(ProtobufTypeUtil.PROTOBUF_UNKNOWN_FIELDS_PREFIX + path);
  if (attribute != null) {
    UnknownFieldSet.Builder unknownFieldBuilder = UnknownFieldSet.newBuilder();
    unknownFieldBuilder.mergeDelimitedFrom(
        new ByteArrayInputStream(
            org.apache.commons.codec.binary.Base64.decodeBase64(attribute.getBytes(StandardCharsets.UTF_8))
        )
    );
    UnknownFieldSet unknownFieldSet = unknownFieldBuilder.build();
    builder.setUnknownFields(unknownFieldSet);
  }
}

@Test
public void testProtoToSdcMessageFields() throws Exception {

  List<DynamicMessage> messages = ProtobufTestUtil.getMessages(
    md,
    extensionRegistry,
    ProtobufTestUtil.getProtoBufData()
  );

  for (int i = 0; i < messages.size(); i++) {
    DynamicMessage m = messages.get(i);
    Record record = RecordCreator.create();
    Field field = ProtobufTypeUtil.protobufToSdcField(record, "", md, typeToExtensionMap, m);
    Assert.assertNotNull(field);
    ProtobufTestUtil.checkProtobufRecords(field, i);
  }
}

@Test
public void testProtoToSdcExtensionFields() throws Exception {

  List<DynamicMessage> messages = ProtobufTestUtil.getMessages(
    md,
    extensionRegistry,
    ProtobufTestUtil.getProtoBufData()
  );

  for (int i = 0; i < messages.size(); i++) {
    DynamicMessage m = messages.get(i);
    Record record = RecordCreator.create();
    Field field = ProtobufTypeUtil.protobufToSdcField(record, "", md, typeToExtensionMap, m);
    Assert.assertNotNull(field);
    ProtobufTestUtil.checkProtobufRecordsForExtensions(field, i);
  }
}

@Test
public void testProtoToSdcUnknownFields() throws Exception {

  List<DynamicMessage> messages = ProtobufTestUtil.getMessages(
    md,
    extensionRegistry,
    ProtobufTestUtil.getProtoBufData()
  );

  for (int i = 0; i < messages.size(); i++) {
    DynamicMessage m = messages.get(i);
    Record record = RecordCreator.create();
    ProtobufTypeUtil.protobufToSdcField(record, "", md, typeToExtensionMap, m);
    ProtobufTestUtil.checkRecordForUnknownFields(record, i);
  }
}

@Test
public void testSdcToProtobufFields() throws Exception {

  List<Record> protobufRecords = ProtobufTestUtil.getProtobufRecords();
  ByteArrayOutputStream bOut = new ByteArrayOutputStream();
  BufferedOutputStream bufferedOutputStream = new BufferedOutputStream(bOut);
  for (int i = 0; i < protobufRecords.size(); i++) {
    DynamicMessage dynamicMessage = ProtobufTypeUtil.sdcFieldToProtobufMsg(
        protobufRecords.get(i),
        md,
        typeToExtensionMap,
        defaultValueMap
    );

    dynamicMessage.writeDelimitedTo(bufferedOutputStream);
  }
  bufferedOutputStream.flush();
  bufferedOutputStream.close();
  ProtobufTestUtil.checkProtobufDataFields(bOut.toByteArray());
}

@Test
public void testSdcToProtobufExtensions() throws Exception {

  List<Record> protobufRecords = ProtobufTestUtil.getProtobufRecords();
  ByteArrayOutputStream bOut = new ByteArrayOutputStream();
  BufferedOutputStream bufferedOutputStream = new BufferedOutputStream(bOut);
  for (int i = 0; i < protobufRecords.size(); i++) {
    DynamicMessage dynamicMessage = ProtobufTypeUtil.sdcFieldToProtobufMsg(
      protobufRecords.get(i),
      md,
      typeToExtensionMap,
      defaultValueMap
    );

    dynamicMessage.writeDelimitedTo(bufferedOutputStream);
  }
  bufferedOutputStream.flush();
  bufferedOutputStream.close();
  ProtobufTestUtil.checkProtobufDataExtensions(bOut.toByteArray());
}

@Test
public void testSdcToProtobufUnknownFields() throws Exception {

  List<Record> protobufRecords = ProtobufTestUtil.getProtobufRecords();
  ByteArrayOutputStream bOut = new ByteArrayOutputStream();
  BufferedOutputStream bufferedOutputStream = new BufferedOutputStream(bOut);
  for (int i = 0; i < protobufRecords.size(); i++) {
    DynamicMessage dynamicMessage = ProtobufTypeUtil.sdcFieldToProtobufMsg(
      protobufRecords.get(i),
      md,
      typeToExtensionMap,
      defaultValueMap
    );

    dynamicMessage.writeDelimitedTo(bufferedOutputStream);
  }
  bufferedOutputStream.flush();
  bufferedOutputStream.close();
  ProtobufTestUtil.checkProtobufDataUnknownFields(bOut.toByteArray());
}

@Test
public void testNullRepeated() throws DataGeneratorException {
  Record r = RecordCreator.create();
  Map<String, Field> repeated = new HashMap<>();
  repeated.put("samples", Field.create(Field.Type.LIST, null));
  r.set(Field.create(repeated));
  Descriptors.Descriptor descriptor = RepeatedProto.getDescriptor().findMessageTypeByName("Repeated");
  // repeated field samples is null and ignored
  DynamicMessage dynamicMessage = ProtobufTypeUtil.sdcFieldToProtobufMsg(
    r,
    descriptor,
    typeToExtensionMap,
    defaultValueMap
  );
  // null repeated fields are treated as empty arrays
  Object samples = dynamicMessage.getField(descriptor.findFieldByName("samples"));
  Assert.assertNotNull(samples);
  Assert.assertTrue(samples instanceof List);
  Assert.assertEquals(0, ((List) samples).size());
}

@Test
public void testEmptyRepeated() throws DataGeneratorException {
  Record r = RecordCreator.create();
  Map<String, Field> repeated = new HashMap<>();
  repeated.put("samples", Field.create(Field.Type.LIST, new ArrayList<>()));
  r.set(Field.create(repeated));
  Descriptors.Descriptor descriptor = RepeatedProto.getDescriptor().findMessageTypeByName("Repeated");
  // repeated field samples is null and ignored
  DynamicMessage dynamicMessage = ProtobufTypeUtil.sdcFieldToProtobufMsg(
    r,
    descriptor,
    typeToExtensionMap,
    defaultValueMap
  );
  // null repeated fields are treated as empty arrays
  Object samples = dynamicMessage.getField(descriptor.findFieldByName("samples"));
  Assert.assertNotNull(samples);
  Assert.assertTrue(samples instanceof List);
  Assert.assertEquals(0, ((List)samples).size());
}

@Test
public void testGetDescriptor() throws IOException {
    Descriptor descriptor2 = AddressBookProtos.AddressBook.getDescriptor();

    FieldDescriptor stringMapFD = descriptor2.findFieldByName("person");
    byte[] bytes = getProtoBytes2();
    DynamicMessage parseFrom = DynamicMessage.parseFrom(descriptor2, bytes);
    Object field = parseFrom.getField(stringMapFD);
    Assert.assertTrue(field instanceof List);

    Codec<AddressBookProtosPOJO> codec = ProtobufProxy.create(AddressBookProtosPOJO.class, false);
    Descriptor descriptor = codec.getDescriptor();

    stringMapFD = descriptor.findFieldByName("list");

    bytes = getProtoBytes2();

    parseFrom = DynamicMessage.parseFrom(descriptor, bytes);

    Object field2 = parseFrom.getField(stringMapFD);
    Assert.assertTrue(field2 instanceof List);

}

/**
 * Test get descriptor.
 *
 * @throws IOException Signals that an I/O exception has occurred.
 */
@Test
public void testGetDescriptor() throws IOException {
    Descriptor descriptor2 = AddressBookProtos.AddressBook.getDescriptor();

    FieldDescriptor stringMapFD = descriptor2.findFieldByName("person");
    byte[] bytes = getProtoBytes2();
    DynamicMessage parseFrom = DynamicMessage.parseFrom(descriptor2, bytes);
    Object field = parseFrom.getField(stringMapFD);
    Assert.assertTrue(field instanceof List);

    Codec<AddressBookProtosPOJO> codec = ProtobufProxy.create(AddressBookProtosPOJO.class);
    Descriptor descriptor = codec.getDescriptor();

    stringMapFD = descriptor.findFieldByName("list");

    bytes = getProtoBytes2();

    parseFrom = DynamicMessage.parseFrom(descriptor, bytes);

    Object field2 = parseFrom.getField(stringMapFD);
    Assert.assertTrue(field2 instanceof List);

}

@Test
public void testTextFormatWithDescriptor() throws Exception {
  //TestUtil.getAllSet();
  String allSet = TextFormat.printToString(TestUtil.getAllSet());
  final DescriptorProto expectedAllSetProto = TestAllTypes.getDescriptor().toProto();
  String allSetProto = TextFormat.printToString(expectedAllSetProto);
  log.debug("the message: {}", allSet);
  log.debug("the proto: {}", allSetProto);

  DynamicMessage.Builder builder = DynamicMessage.newBuilder(DescriptorProto.getDescriptor());
  TextFormat.merge(allSetProto, builder);
  Message actualAllSetProto = builder.build();

  assertThat(actualAllSetProto).isEqualTo(expectedAllSetProto);

  FieldDescriptor field = FileDescriptorProto.getDescriptor()
      .findFieldByNumber(FileDescriptorProto.MESSAGE_TYPE_FIELD_NUMBER);
  FileDescriptorProto fileProto = FileDescriptorProto.newBuilder().setName("my file1")
      .addRepeatedField(field, actualAllSetProto)
      .build();
  FileDescriptor fileDescriptor = FileDescriptor.buildFrom(fileProto, new FileDescriptor[0]);
  Descriptor actualAllTypesDescriptor = fileDescriptor.findMessageTypeByName(
      TestAllTypes.getDescriptor().getFullName());

  assertThat(actualAllTypesDescriptor, equalTo(TestAllTypes.getDescriptor()));
}

public Message callBlockingMethod(MethodDescriptorContainer methodDescriptor, Message request) throws ServiceException {
    try {
        dataOutputStream.writeUTF(methodDescriptor.getServiceDescriptorContainer().getName());
        dataOutputStream.writeUTF(methodDescriptor.getName());
        dataOutputStream.writeUTF(tokenHolder.getToken() == null ? "" : tokenHolder.getToken());
        request.writeDelimitedTo(dataOutputStream);
        dataOutputStream.flush();
        DynamicMessage response = DynamicMessage.getDefaultInstance(methodDescriptor.getOutputDescriptor());
        Builder responseBuilder = response.newBuilderForType();
        responseBuilder.mergeDelimitedFrom(inputStream);
        return responseBuilder.build();
    } catch (IOException e) {
        LOGGER.error("", e);
    }
    return null;
}

/**
 * {@inheritDoc}
 *
 * This variation allows for the inclusion of schemas for serializing
 * sub-objects that may appear in {@code message}. If no suitable schema
 * is found in the registry, a schema with default settings is generated
 * on the fly using {@link
 * SchemaSource#get(com.google.protobuf.Descriptors.Descriptor)}.
 *
 */
public Message parse(JsonNode node, ReadableSchemaRegistry registry) throws ParsingException {
    Message.Builder builder = DynamicMessage.newBuilder(schema.getDescriptor());

    for (Map.Entry<String, FieldDescriptor> fieldEntry : schema.getFields().entrySet()) {
        String fieldName = schema.getPropertyName(fieldEntry.getKey());
        FieldDescriptor field = fieldEntry.getValue();
        if (node.has(fieldName) && !node.get(fieldName).isNull()) {
            JsonNode valueNode = node.get(fieldName);
            if (field.isRepeated()) {
                if (schema.getMappings().containsKey(field.getName())) {
                    parseMappedField(registry, builder, fieldName, field, valueNode);
                } else {
                    parseRepeatedField(registry, builder, fieldName, field, valueNode);
                }
            } else {
                Object value = parseValue(valueNode, field, registry);
                if (value != null) {
                    builder.setField(field, value);
                }
            }
        }
    }

    return builder.build();
}

/**
 * Parses a repeated mapped field.
 *
 * @param registry a registry of schemas, used for parsing enclosed objects
 * @param builder the builder in which the parsed field should be set
 * @param field the descriptor of the repeated field being parsed
 * @param fieldName the JSON name of the field
 * @param valueNode the JSON node being parsed
 * @throws ParsingException
 * @see NamedSchema#mapRepeatedField(String,String)
 */
private void parseMappedField(ReadableSchemaRegistry registry, Message.Builder builder,
        String fieldName, FieldDescriptor field, JsonNode valueNode) throws ParsingException {
    if (!valueNode.isObject()) {
        throw new IllegalArgumentException(
                "Field '" + fieldName +
                "' is expected to be an object, but was " +
                valueNode.asToken());
    }
    ObjectNode objectNode = (ObjectNode) valueNode;
    Iterator<Map.Entry<String, JsonNode>> subObjectsIterator =
            objectNode.fields();
    while (subObjectsIterator.hasNext()) {
        Map.Entry<String, JsonNode> subObject = subObjectsIterator.next();
        Message message = (Message) parseValue(
                subObject.getValue(), field, registry);
        DynamicMessage.Builder dynamicMessage = DynamicMessage.newBuilder(
                field.getMessageType());
        dynamicMessage.mergeFrom(message);
        dynamicMessage.setField(
                schema.getMappings().get(field.getName()), subObject.getKey());
        builder.addRepeatedField(field, dynamicMessage.build());
    }
}

/**
 * Parses a repeated mapped field.
 *
 * @param registry a registry of schemas, used for parsing enclosed objects
 * @param builder the builder in which the parsed field should be set
 * @param field the descriptor of the repeated field being parsed
 * @param fieldName the JSON name of the field
 * @param valueNode the JSON node being parsed
 * @throws ParsingException
 * @see {@link #mapRepeatedField(String, String)}
 */
private void parseMappedField(ReadableSchemaRegistry registry, Message.Builder builder,
        String fieldName, FieldDescriptor field, JsonNode valueNode) throws ParsingException {
    if (!valueNode.isObject()) {
        throw new IllegalArgumentException(
                "Field '" + fieldName +
                "' is expected to be an object, but was " +
                valueNode.asToken());
    }
    ObjectNode objectNode = (ObjectNode) valueNode;
    Iterator<Map.Entry<String, JsonNode>> subObjectsIterator =
            objectNode.fields();
    while (subObjectsIterator.hasNext()) {
        Map.Entry<String, JsonNode> subObject = subObjectsIterator.next();
        Message message = (Message) parseValue(
                subObject.getValue(), field, registry);
        DynamicMessage.Builder dynamicMessage = DynamicMessage.newBuilder(
                field.getMessageType());
        dynamicMessage.mergeFrom(message);
        dynamicMessage.setField(
                schema.getMappings().get(field.getName()), subObject.getKey());
        builder.addRepeatedField(field, dynamicMessage.build());
    }
}

/**
 * {@inheritDoc}
 */
@Override
public Message parse(JsonNode node, ReadableSchemaRegistry registry) throws ParsingException {
    Message.Builder builder = DynamicMessage.newBuilder(descriptor);

    for (FieldDescriptor field : descriptor.getFields()) {
        String fieldName = AutoSchema.PROTO_FIELD_CASE_FORMAT.to(
                AutoSchema.JSON_FIELD_CASE_FORMAT, field.getName());
        if (node.has(fieldName) && !node.get(fieldName).isNull()) {
            JsonNode valueNode = node.get(fieldName);
            if (field.isRepeated()) {
                parseRepeatedField(builder, fieldName, field, valueNode, registry);
            } else {
                Object value = parseValue(valueNode, field, registry);
                if (value != null) {
                    builder.setField(field, value);
                }
            }
        }
    }

    return builder.build();
}

@Override
public DynamicMessage parse(InputStream inputStream) {
  try {
    return DynamicMessage.newBuilder(messageDescriptor)
        .mergeFrom(inputStream, ExtensionRegistryLite.getEmptyRegistry())
        .build();
  } catch (IOException e) {
    throw new RuntimeException("Unable to merge from the supplied input stream", e);
  }
}

/**
 * Makes an rpc to the remote endpoint and respects the supplied callback. Returns a future which
 * terminates once the call has ended. For calls which are single-request, this throws
 * {@link IllegalArgumentException} if the size of {@code requests} is not exactly 1.
 */
public ListenableFuture<Void> call(
    ImmutableList<DynamicMessage> requests,
    StreamObserver<DynamicMessage> responseObserver,
    CallOptions callOptions) {
  Preconditions.checkArgument(!requests.isEmpty(), "Can't make call without any requests");
  MethodType methodType = getMethodType();
  long numRequests = requests.size();
  if (methodType == MethodType.UNARY) {
    logger.info("Making unary call");
    Preconditions.checkArgument(numRequests == 1,
        "Need exactly 1 request for unary call, but got: " + numRequests);
    return callUnary(requests.get(0), responseObserver, callOptions);
  } else if (methodType == MethodType.SERVER_STREAMING) {
    logger.info("Making server streaming call");
    Preconditions.checkArgument(numRequests == 1,
        "Need exactly 1 request for server streaming call, but got: " + numRequests);
    return callServerStreaming(requests.get(0), responseObserver, callOptions);
  } else if (methodType == MethodType.CLIENT_STREAMING) {
    logger.info("Making client streaming call with " + requests.size() + " requests");
    return callClientStreaming(requests, responseObserver, callOptions);
  } else {
    // Bidi streaming.
    logger.info("Making bidi streaming call with " + requests.size() + " requests");
    return callBidiStreaming(requests, responseObserver, callOptions);
  }
}

private ListenableFuture<Void> callBidiStreaming(
    ImmutableList<DynamicMessage> requests,
    StreamObserver<DynamicMessage> responseObserver,
    CallOptions callOptions) {
  DoneObserver<DynamicMessage> doneObserver = new DoneObserver<>();
  StreamObserver<DynamicMessage> requestObserver = ClientCalls.asyncBidiStreamingCall(
      createCall(callOptions),
      CompositeStreamObserver.of(responseObserver, doneObserver));
  requests.forEach(requestObserver::onNext);
  requestObserver.onCompleted();
  return doneObserver.getCompletionFuture();
}

private ListenableFuture<Void> callClientStreaming(
    ImmutableList<DynamicMessage> requests,
    StreamObserver<DynamicMessage> responseObserver,
    CallOptions callOptions) {
  DoneObserver<DynamicMessage> doneObserver = new DoneObserver<>();
  StreamObserver<DynamicMessage> requestObserver = ClientCalls.asyncClientStreamingCall(
      createCall(callOptions),
      CompositeStreamObserver.of(responseObserver, doneObserver));
  requests.forEach(requestObserver::onNext);
  requestObserver.onCompleted();
  return doneObserver.getCompletionFuture();
}

private ListenableFuture<Void> callServerStreaming(
    DynamicMessage request,
    StreamObserver<DynamicMessage> responseObserver,
    CallOptions callOptions) {
  DoneObserver<DynamicMessage> doneObserver = new DoneObserver<>();
  ClientCalls.asyncServerStreamingCall(
      createCall(callOptions),
      request,
      CompositeStreamObserver.of(responseObserver, doneObserver));
  return doneObserver.getCompletionFuture();
}

private ListenableFuture<Void> callUnary(
    DynamicMessage request,
    StreamObserver<DynamicMessage> responseObserver,
    CallOptions callOptions) {
  DoneObserver<DynamicMessage> doneObserver = new DoneObserver<>();
  ClientCalls.asyncUnaryCall(
      createCall(callOptions),
      request,
      CompositeStreamObserver.of(responseObserver, doneObserver));
  return doneObserver.getCompletionFuture();
}

private io.grpc.MethodDescriptor<DynamicMessage, DynamicMessage> createGrpcMethodDescriptor() {
  return io.grpc.MethodDescriptor.<DynamicMessage, DynamicMessage>create(
      getMethodType(),
      getFullMethodName(),
      new DynamicMessageMarshaller(protoMethodDescriptor.getInputType()),
      new DynamicMessageMarshaller(protoMethodDescriptor.getOutputType()));
}

/** Attempts to read a response proto from the supplied file. */
public static ImmutableList<TestResponse> readResponseFile(Path file)
    throws InvalidProtocolBufferException {
  MessageReader reader = MessageReader.forFile(file, TestResponse.getDescriptor());
  ImmutableList<DynamicMessage> responses = reader.read();

  ImmutableList.Builder<TestResponse> resultBuilder = ImmutableList.builder();
  for (DynamicMessage response : responses) {
    resultBuilder.add(TestResponse.parseFrom(response.toByteString()));
  }
  return resultBuilder.build();
}

/** Parses all the messages and returns them in a list. */
public ImmutableList<DynamicMessage> read() {
  ImmutableList.Builder<DynamicMessage> resultBuilder = ImmutableList.builder();
  try {
    String line;
    boolean wasLastLineEmpty = false;
    while (true) {
      line = bufferedReader.readLine();

      // Two consecutive empty lines mark the end of the stream.
      if (Strings.isNullOrEmpty(line)) {
        if (wasLastLineEmpty) {
          return resultBuilder.build();
        }
        wasLastLineEmpty = true;
        continue;
      } else {
        wasLastLineEmpty = false;
      }

      // Read the next full message.
      StringBuilder stringBuilder = new StringBuilder();
      while (!Strings.isNullOrEmpty(line)) {
        stringBuilder.append(line);
        line = bufferedReader.readLine();
      }
      wasLastLineEmpty = true;

      DynamicMessage.Builder nextMessage = DynamicMessage.newBuilder(descriptor);
      jsonParser.merge(stringBuilder.toString(), nextMessage);

      // Clean up and prepare for next message.
      resultBuilder.add(nextMessage.build());
    }
  } catch (Exception e) {
    throw new IllegalArgumentException("Unable to read messages from: " + source, e);
  }
}

private static DynamicMessage makeProto(String content) {
  return DynamicMessage.newBuilder(
      TestRequest.newBuilder()
          .setMessage(content)
          .build())
      .build();
}

@Before
public void setUp() {
  when(mockChannel.newCall(
      Matchers.<io.grpc.MethodDescriptor<DynamicMessage, DynamicMessage>>any(),
      Matchers.any()))
          .thenReturn(mockClientCall);
}

@Test
public void passesCallOptions() {
  client = new DynamicGrpcClient(UNARY_METHOD, mockChannel);
  client.call(ImmutableList.of(REQUEST), mockStreamObserver, CALL_OPTIONS);

  verify(mockChannel).newCall(
      Matchers.<io.grpc.MethodDescriptor<DynamicMessage, DynamicMessage>>any(),
      callOptionsCaptor.capture());
  assertThat(callOptionsCaptor.getValue()).isEqualTo(CALL_OPTIONS);
}

@Test
public void emptyStream() throws Throwable {
  MessageWriter<Message> writer = makeWriter();
  writer.onCompleted();

  ImmutableList<DynamicMessage> results = makeReader().read();
  assertThat(results).isEmpty();
}

@Test
public void singleMessage() throws Throwable {
  MessageWriter<Message> writer = makeWriter();
  writer.onNext(TestData.REQUEST);
  writer.onCompleted();

  ImmutableList<DynamicMessage> results = makeReader().read();
  assertThat(results).containsExactly(TestData.REQUEST);
}