private void getAllExtensions(ServerReflectionRequest request) {
  String type = request.getAllExtensionNumbersOfType();
  Set<Integer> extensions = serverReflectionIndex.getExtensionNumbersOfType(type);
  if (extensions != null) {
    ExtensionNumberResponse.Builder builder =
        ExtensionNumberResponse.newBuilder()
            .setBaseTypeName(type)
            .addAllExtensionNumber(extensions);
    serverCallStreamObserver.onNext(
        ServerReflectionResponse.newBuilder()
            .setValidHost(request.getHost())
            .setOriginalRequest(request)
            .setAllExtensionNumbersResponse(builder)
            .build());
  } else {
    sendErrorResponse(request, Status.Code.NOT_FOUND, "Type not found.");
  }
}
 

private ServerReflectionResponse createServerReflectionResponse(
    ServerReflectionRequest request, FileDescriptor fd) {
  FileDescriptorResponse.Builder fdRBuilder = FileDescriptorResponse.newBuilder();

  Set<String> seenFiles = new HashSet<String>();
  Queue<FileDescriptor> frontier = new ArrayDeque<FileDescriptor>();
  seenFiles.add(fd.getName());
  frontier.add(fd);
  while (!frontier.isEmpty()) {
    FileDescriptor nextFd = frontier.remove();
    fdRBuilder.addFileDescriptorProto(nextFd.toProto().toByteString());
    for (FileDescriptor dependencyFd : nextFd.getDependencies()) {
      if (!seenFiles.contains(dependencyFd.getName())) {
        seenFiles.add(dependencyFd.getName());
        frontier.add(dependencyFd);
      }
    }
  }
  return ServerReflectionResponse.newBuilder()
      .setValidHost(request.getHost())
      .setOriginalRequest(request)
      .setFileDescriptorResponse(fdRBuilder)
      .build();
}
 

@Test
public void fileContainingNestedSymbol() throws Exception {
  ServerReflectionRequest request =
      ServerReflectionRequest.newBuilder()
          .setHost(TEST_HOST)
          .setFileContainingSymbol("grpc.reflection.testing.NestedTypeOuter.Middle.Inner")
          .build();

  ServerReflectionResponse goldenResponse =
      ServerReflectionResponse.newBuilder()
          .setValidHost(TEST_HOST)
          .setOriginalRequest(request)
          .setFileDescriptorResponse(
              FileDescriptorResponse.newBuilder()
                  .addFileDescriptorProto(
                      ReflectionTestDepthThreeProto.getDescriptor().toProto().toByteString())
                  .build())
          .build();

  StreamRecorder<ServerReflectionResponse> responseObserver = StreamRecorder.create();
  StreamObserver<ServerReflectionRequest> requestObserver =
      stub.serverReflectionInfo(responseObserver);
  requestObserver.onNext(request);
  requestObserver.onCompleted();
  assertEquals(goldenResponse, responseObserver.firstValue().get());
}
 

@Test
public void allExtensionNumbersOfType() throws Exception {
  ServerReflectionRequest request =
      ServerReflectionRequest.newBuilder()
          .setHost(TEST_HOST)
          .setAllExtensionNumbersOfType("grpc.reflection.testing.ThirdLevelType")
          .build();

  Set<Integer> goldenResponse = new HashSet<Integer>(Arrays.asList(100, 101));

  StreamRecorder<ServerReflectionResponse> responseObserver = StreamRecorder.create();
  StreamObserver<ServerReflectionRequest> requestObserver =
      stub.serverReflectionInfo(responseObserver);
  requestObserver.onNext(request);
  requestObserver.onCompleted();
  Set<Integer> extensionNumberResponseSet =
      new HashSet<Integer>(
          responseObserver
              .firstValue()
              .get()
              .getAllExtensionNumbersResponse()
              .getExtensionNumberList());
  assertEquals(goldenResponse, extensionNumberResponseSet);
}
 

@Test
public void testRetrievingFilesByFileName() {
    ServerReflectionRequest request = ServerReflectionRequest.newBuilder()
            .setHost("localhost")
            .setFileByFilename("reflection/reflection_test_depth_three.proto")
            .build();

    ServerReflectionResponse expected = ServerReflectionResponse.newBuilder()
            .setValidHost("localhost")
            .setOriginalRequest(request)
            .setFileDescriptorResponse(
                    FileDescriptorResponse.newBuilder()
                            .addFileDescriptorProto(
                                    ReflectionTestDepthThreeProto.getDescriptor().toProto().toByteString())
                            .build())
            .build();

    ServerReflectionResponse response = invoke(request);
    assertThat(response).isEqualTo(expected);
}
 

@Test
public void testRetrievingFilesContainingSymbol() {
    ServerReflectionRequest request = ServerReflectionRequest.newBuilder()
            .setHost("localhost")
            .setFileContainingSymbol("grpc.reflection.testing.ReflectableService.Method")
            .build();

    List<ByteString> responses = Arrays.asList(
            ReflectionTestProto.getDescriptor().toProto().toByteString(),
            ReflectionTestDepthTwoProto.getDescriptor().toProto().toByteString(),
            ReflectionTestDepthThreeProto.getDescriptor().toProto().toByteString());

    ServerReflectionResponse response = invoke(request);
    List<ByteString> list = response.getFileDescriptorResponse().getFileDescriptorProtoList();
    assertThat(list).containsExactlyInAnyOrderElementsOf(responses);
}
 

@Test
public void testRetrievingFilesContainingNestedSymbol() {
    ServerReflectionRequest request = ServerReflectionRequest.newBuilder()
            .setHost("localhost")
            .setFileContainingSymbol("grpc.reflection.testing.NestedTypeOuter.Middle.Inner")
            .build();
    ServerReflectionResponse expected = ServerReflectionResponse.newBuilder()
            .setValidHost("localhost")
            .setOriginalRequest(request)
            .setFileDescriptorResponse(
                    FileDescriptorResponse.newBuilder()
                            .addFileDescriptorProto(
                                    ReflectionTestDepthThreeProto.getDescriptor().toProto().toByteString())
                            .build())
            .build();
    ServerReflectionResponse resp = invoke(request);
    assertThat(resp).isEqualTo(expected);
}
 

@Test
public void testRetrievingFilesContainingExtension() {
    ServerReflectionRequest request = ServerReflectionRequest.newBuilder()
            .setHost("localhost")
            .setFileContainingExtension(
                    ExtensionRequest.newBuilder()
                            .setContainingType("grpc.reflection.testing.ThirdLevelType")
                            .setExtensionNumber(100)
                            .build())
            .build();

    List<ByteString> expected = Arrays.asList(
            ReflectionTestProto.getDescriptor().toProto().toByteString(),
            ReflectionTestDepthTwoProto.getDescriptor().toProto().toByteString(),
            ReflectionTestDepthThreeProto.getDescriptor().toProto().toByteString());

    ServerReflectionResponse response = invoke(request);
    assertThat(response.getFileDescriptorResponse().getFileDescriptorProtoList())
            .containsExactlyInAnyOrderElementsOf(expected);
}
 

@Test
public void testRetrievingFilesContainingNestedExtension() {
    ServerReflectionRequest request = ServerReflectionRequest.newBuilder()
            .setHost("localhost")
            .setFileContainingExtension(
                    ExtensionRequest.newBuilder()
                            .setContainingType("grpc.reflection.testing.ThirdLevelType")
                            .setExtensionNumber(101)
                            .build())
            .build();

    ServerReflectionResponse expected = ServerReflectionResponse.newBuilder()
            .setValidHost("localhost")
            .setOriginalRequest(request)
            .setFileDescriptorResponse(
                    FileDescriptorResponse.newBuilder()
                            .addFileDescriptorProto(
                                    ReflectionTestDepthTwoProto.getDescriptor().toProto().toByteString())
                            .addFileDescriptorProto(
                                    ReflectionTestDepthThreeProto.getDescriptor().toProto().toByteString())
                            .build())
            .build();

    ServerReflectionResponse response = invoke(request);
    assertThat(response).isEqualTo(expected);
}
 

@Override
public Multi<ServerReflectionResponse> serverReflectionInfo(Multi<ServerReflectionRequest> request) {
    return request
            .onItem().apply(new Function<ServerReflectionRequest, ServerReflectionResponse>() {
                @Override
                public ServerReflectionResponse apply(ServerReflectionRequest req) {
                    switch (req.getMessageRequestCase()) {
                        case LIST_SERVICES:
                            return ReflectionService.this.getServiceList(req);
                        case FILE_BY_FILENAME:
                            return ReflectionService.this.getFileByName(req);
                        case FILE_CONTAINING_SYMBOL:
                            return ReflectionService.this.getFileContainingSymbol(req);
                        case FILE_CONTAINING_EXTENSION:
                            return ReflectionService.this.getFileByExtension(req);
                        case ALL_EXTENSION_NUMBERS_OF_TYPE:
                            return ReflectionService.this.getAllExtensions(req);
                        default:
                            return ReflectionService.this.getErrorResponse(req, Status.Code.UNIMPLEMENTED,
                                    "not implemented " + req.getMessageRequestCase());

                    }
                }
            });
}
 

private ServerReflectionResponse getAllExtensions(ServerReflectionRequest request) {
    String type = request.getAllExtensionNumbersOfType();
    Set<Integer> extensions = index.getExtensionNumbersOfType(type);
    if (extensions != null) {
        ExtensionNumberResponse.Builder builder = ExtensionNumberResponse.newBuilder()
                .setBaseTypeName(type)
                .addAllExtensionNumber(extensions);
        return ServerReflectionResponse.newBuilder()
                .setValidHost(request.getHost())
                .setOriginalRequest(request)
                .setAllExtensionNumbersResponse(builder)
                .build();
    } else {
        return getErrorResponse(request, Status.Code.NOT_FOUND, "Type not found.");
    }
}
 

private ServerReflectionResponse getServerReflectionResponse(
        ServerReflectionRequest request, FileDescriptor fd) {
    FileDescriptorResponse.Builder fdRBuilder = FileDescriptorResponse.newBuilder();

    // Traverse the descriptors to get the full list of dependencies and add them to the builder
    Set<String> seenFiles = new HashSet<>();
    Queue<FileDescriptor> frontier = new ArrayDeque<>();
    seenFiles.add(fd.getName());
    frontier.add(fd);
    while (!frontier.isEmpty()) {
        FileDescriptor nextFd = frontier.remove();
        fdRBuilder.addFileDescriptorProto(nextFd.toProto().toByteString());
        for (FileDescriptor dependencyFd : nextFd.getDependencies()) {
            if (!seenFiles.contains(dependencyFd.getName())) {
                seenFiles.add(dependencyFd.getName());
                frontier.add(dependencyFd);
            }
        }
    }
    return ServerReflectionResponse.newBuilder()
            .setValidHost(request.getHost())
            .setOriginalRequest(request)
            .setFileDescriptorResponse(fdRBuilder)
            .build();
}
 

private void getAllExtensions(ServerReflectionRequest request) {
  String type = request.getAllExtensionNumbersOfType();
  Set<Integer> extensions = serverReflectionIndex.getExtensionNumbersOfType(type);
  if (extensions != null) {
    ExtensionNumberResponse.Builder builder =
        ExtensionNumberResponse.newBuilder()
            .setBaseTypeName(type)
            .addAllExtensionNumber(extensions);
    serverCallStreamObserver.onNext(
        ServerReflectionResponse.newBuilder()
            .setValidHost(request.getHost())
            .setOriginalRequest(request)
            .setAllExtensionNumbersResponse(builder)
            .build());
  } else {
    sendErrorResponse(request, Status.Code.NOT_FOUND, "Type not found.");
  }
}
 

private ServerReflectionResponse createServerReflectionResponse(
    ServerReflectionRequest request, FileDescriptor fd) {
  FileDescriptorResponse.Builder fdRBuilder = FileDescriptorResponse.newBuilder();

  Set<String> seenFiles = new HashSet<>();
  Queue<FileDescriptor> frontier = new ArrayDeque<>();
  seenFiles.add(fd.getName());
  frontier.add(fd);
  while (!frontier.isEmpty()) {
    FileDescriptor nextFd = frontier.remove();
    fdRBuilder.addFileDescriptorProto(nextFd.toProto().toByteString());
    for (FileDescriptor dependencyFd : nextFd.getDependencies()) {
      if (!seenFiles.contains(dependencyFd.getName())) {
        seenFiles.add(dependencyFd.getName());
        frontier.add(dependencyFd);
      }
    }
  }
  return ServerReflectionResponse.newBuilder()
      .setValidHost(request.getHost())
      .setOriginalRequest(request)
      .setFileDescriptorResponse(fdRBuilder)
      .build();
}
 

@Test
public void fileContainingNestedSymbol() throws Exception {
  ServerReflectionRequest request =
      ServerReflectionRequest.newBuilder()
          .setHost(TEST_HOST)
          .setFileContainingSymbol("grpc.reflection.testing.NestedTypeOuter.Middle.Inner")
          .build();

  ServerReflectionResponse goldenResponse =
      ServerReflectionResponse.newBuilder()
          .setValidHost(TEST_HOST)
          .setOriginalRequest(request)
          .setFileDescriptorResponse(
              FileDescriptorResponse.newBuilder()
                  .addFileDescriptorProto(
                      ReflectionTestDepthThreeProto.getDescriptor().toProto().toByteString())
                  .build())
          .build();

  StreamRecorder<ServerReflectionResponse> responseObserver = StreamRecorder.create();
  StreamObserver<ServerReflectionRequest> requestObserver =
      stub.serverReflectionInfo(responseObserver);
  requestObserver.onNext(request);
  requestObserver.onCompleted();
  assertEquals(goldenResponse, responseObserver.firstValue().get());
}
 

@Test
public void allExtensionNumbersOfType() throws Exception {
  ServerReflectionRequest request =
      ServerReflectionRequest.newBuilder()
          .setHost(TEST_HOST)
          .setAllExtensionNumbersOfType("grpc.reflection.testing.ThirdLevelType")
          .build();

  Set<Integer> goldenResponse = new HashSet<>(Arrays.asList(100, 101));

  StreamRecorder<ServerReflectionResponse> responseObserver = StreamRecorder.create();
  StreamObserver<ServerReflectionRequest> requestObserver =
      stub.serverReflectionInfo(responseObserver);
  requestObserver.onNext(request);
  requestObserver.onCompleted();
  Set<Integer> extensionNumberResponseSet =
      new HashSet<>(
          responseObserver
              .firstValue()
              .get()
              .getAllExtensionNumbersResponse()
              .getExtensionNumberList());
  assertEquals(goldenResponse, extensionNumberResponseSet);
}
 

@Override
public StreamObserver<ServerReflectionRequest> serverReflectionInfo(
    final StreamObserver<ServerReflectionResponse> responseObserver) {
  final ServerCallStreamObserver<ServerReflectionResponse> serverCallStreamObserver =
      (ServerCallStreamObserver<ServerReflectionResponse>) responseObserver;
  ProtoReflectionStreamObserver requestObserver =
      new ProtoReflectionStreamObserver(updateIndexIfNecessary(), serverCallStreamObserver);
  serverCallStreamObserver.setOnReadyHandler(requestObserver);
  serverCallStreamObserver.disableAutoInboundFlowControl();
  serverCallStreamObserver.request(1);
  return requestObserver;
}
 

private void listServices(ServerReflectionRequest request) {
  ListServiceResponse.Builder builder = ListServiceResponse.newBuilder();
  for (String serviceName : serverReflectionIndex.getServiceNames()) {
    builder.addService(ServiceResponse.newBuilder().setName(serviceName));
  }
  serverCallStreamObserver.onNext(
      ServerReflectionResponse.newBuilder()
          .setValidHost(request.getHost())
          .setOriginalRequest(request)
          .setListServicesResponse(builder)
          .build());
}
 

private void sendErrorResponse(
    ServerReflectionRequest request, Status.Code code, String message) {
  ServerReflectionResponse response =
      ServerReflectionResponse.newBuilder()
          .setValidHost(request.getHost())
          .setOriginalRequest(request)
          .setErrorResponse(
              ErrorResponse.newBuilder()
                  .setErrorCode(code.value())
                  .setErrorMessage(message))
          .build();
  serverCallStreamObserver.onNext(response);
}
 

@Test
public void fileByFilename() throws Exception {
  ServerReflectionRequest request =
      ServerReflectionRequest.newBuilder()
          .setHost(TEST_HOST)
          .setFileByFilename("io/grpc/reflection/testing/reflection_test_depth_three.proto")
          .build();

  ServerReflectionResponse goldenResponse =
      ServerReflectionResponse.newBuilder()
          .setValidHost(TEST_HOST)
          .setOriginalRequest(request)
          .setFileDescriptorResponse(
              FileDescriptorResponse.newBuilder()
                  .addFileDescriptorProto(
                      ReflectionTestDepthThreeProto.getDescriptor().toProto().toByteString())
                  .build())
          .build();

  StreamRecorder<ServerReflectionResponse> responseObserver = StreamRecorder.create();
  StreamObserver<ServerReflectionRequest> requestObserver =
      stub.serverReflectionInfo(responseObserver);
  requestObserver.onNext(request);
  requestObserver.onCompleted();

  assertEquals(goldenResponse, responseObserver.firstValue().get());
}
 

@Test
public void fileContainingSymbol() throws Exception {
  ServerReflectionRequest request =
      ServerReflectionRequest.newBuilder()
          .setHost(TEST_HOST)
          .setFileContainingSymbol("grpc.reflection.testing.ReflectableService.Method")
          .build();

  List<ByteString> goldenResponse =
      Arrays.asList(
          ReflectionTestProto.getDescriptor().toProto().toByteString(),
          ReflectionTestDepthTwoProto.getDescriptor().toProto().toByteString(),
          ReflectionTestDepthTwoAlternateProto.getDescriptor().toProto().toByteString(),
          ReflectionTestDepthThreeProto.getDescriptor().toProto().toByteString());

  StreamRecorder<ServerReflectionResponse> responseObserver = StreamRecorder.create();
  StreamObserver<ServerReflectionRequest> requestObserver =
      stub.serverReflectionInfo(responseObserver);
  requestObserver.onNext(request);
  requestObserver.onCompleted();

  List<ByteString> response =
      responseObserver
          .firstValue()
          .get()
          .getFileDescriptorResponse()
          .getFileDescriptorProtoList();
  assertEquals(goldenResponse.size(), response.size());
  assertEquals(new HashSet<ByteString>(goldenResponse), new HashSet<ByteString>(response));
}
 

@Test
public void fileContainingExtension() throws Exception {
  ServerReflectionRequest request =
      ServerReflectionRequest.newBuilder()
          .setHost(TEST_HOST)
          .setFileContainingExtension(
              ExtensionRequest.newBuilder()
                  .setContainingType("grpc.reflection.testing.ThirdLevelType")
                  .setExtensionNumber(100)
                  .build())
          .build();

  List<ByteString> goldenResponse =
      Arrays.asList(
          ReflectionTestProto.getDescriptor().toProto().toByteString(),
          ReflectionTestDepthTwoProto.getDescriptor().toProto().toByteString(),
          ReflectionTestDepthTwoAlternateProto.getDescriptor().toProto().toByteString(),
          ReflectionTestDepthThreeProto.getDescriptor().toProto().toByteString());

  StreamRecorder<ServerReflectionResponse> responseObserver = StreamRecorder.create();
  StreamObserver<ServerReflectionRequest> requestObserver =
      stub.serverReflectionInfo(responseObserver);
  requestObserver.onNext(request);
  requestObserver.onCompleted();

  List<ByteString> response =
      responseObserver
          .firstValue()
          .get()
          .getFileDescriptorResponse()
          .getFileDescriptorProtoList();
  assertEquals(goldenResponse.size(), response.size());
  assertEquals(new HashSet<ByteString>(goldenResponse), new HashSet<ByteString>(response));
}
 

@Test
public void fileContainingNestedExtension() throws Exception {
  ServerReflectionRequest request =
      ServerReflectionRequest.newBuilder()
          .setHost(TEST_HOST)
          .setFileContainingExtension(
              ExtensionRequest.newBuilder()
                  .setContainingType("grpc.reflection.testing.ThirdLevelType")
                  .setExtensionNumber(101)
                  .build())
          .build();

  ServerReflectionResponse goldenResponse =
      ServerReflectionResponse.newBuilder()
          .setValidHost(TEST_HOST)
          .setOriginalRequest(request)
          .setFileDescriptorResponse(
              FileDescriptorResponse.newBuilder()
                  .addFileDescriptorProto(
                      ReflectionTestDepthTwoProto.getDescriptor().toProto().toByteString())
                  .addFileDescriptorProto(
                      ReflectionTestDepthThreeProto.getDescriptor().toProto().toByteString())
                  .build())
          .build();

  StreamRecorder<ServerReflectionResponse> responseObserver = StreamRecorder.create();
  StreamObserver<ServerReflectionRequest> requestObserver =
      stub.serverReflectionInfo(responseObserver);
  requestObserver.onNext(request);
  requestObserver.onCompleted();
  assertEquals(goldenResponse, responseObserver.firstValue().get());
}
 

private void assertServiceResponseEquals(Set<ServiceResponse> goldenResponse) throws Exception {
  ServerReflectionRequest request =
      ServerReflectionRequest.newBuilder().setHost(TEST_HOST).setListServices("services").build();
  StreamRecorder<ServerReflectionResponse> responseObserver = StreamRecorder.create();
  StreamObserver<ServerReflectionRequest> requestObserver =
      stub.serverReflectionInfo(responseObserver);
  requestObserver.onNext(request);
  requestObserver.onCompleted();
  List<ServiceResponse> response =
      responseObserver.firstValue().get().getListServicesResponse().getServiceList();
  assertEquals(goldenResponse.size(), response.size());
  assertEquals(goldenResponse, new HashSet<ServiceResponse>(response));
}
 

@Override
public void onNext(ServerReflectionResponse serverReflectionResponse) {
    MessageResponseCase responseCase = serverReflectionResponse.getMessageResponseCase();
    switch (responseCase) {
        case LIST_SERVICES_RESPONSE:
            handleListServiceResponse(serverReflectionResponse.getListServicesResponse());
            break;
        default:
            logger.warn("Got unknown reflection response type: " + responseCase);
            break;
    }
}
 

@Override
public void onNext(ServerReflectionResponse response) {
    MessageResponseCase responseCase = response.getMessageResponseCase();
    switch (responseCase) {
        case FILE_DESCRIPTOR_RESPONSE:
            ImmutableSet<FileDescriptorProto> descriptors =
                    parseDescriptors(response.getFileDescriptorResponse().getFileDescriptorProtoList());
            descriptors.forEach(d -> resolvedDescriptors.put(d.getName(), d));
            descriptors.forEach(this::processDependencies);
            break;
        default:
            logger.warn("Got unknown reflection response type: " + responseCase);
            break;
    }
}
 

@Override
public void onNext(ServerReflectionResponse serverReflectionResponse) {
  MessageResponseCase responseCase = serverReflectionResponse.getMessageResponseCase();
  switch (responseCase) {
    case LIST_SERVICES_RESPONSE:
      handleListServiceRespones(serverReflectionResponse.getListServicesResponse());
      break;
    default:
      logger.warn("Got unknown reflection response type: " + responseCase);
      break;
  }
}
 

@Override
public void onNext(ServerReflectionResponse response) {
  MessageResponseCase responseCase = response.getMessageResponseCase();
  switch (responseCase) {
    case FILE_DESCRIPTOR_RESPONSE:
      ImmutableSet<FileDescriptorProto> descriptors =
          parseDescriptors(response.getFileDescriptorResponse().getFileDescriptorProtoList());
      descriptors.forEach(d -> resolvedDescriptors.put(d.getName(), d));
      descriptors.forEach(d -> processDependencies(d));
      break;
    default:
      logger.warn("Got unknown reflection response type: " + responseCase);
      break;
  }
}
 

@Override
public void onNext(ServerReflectionResponse serverReflectionResponse) {
    MessageResponseCase responseCase = serverReflectionResponse.getMessageResponseCase();
    switch (responseCase) {
        case LIST_SERVICES_RESPONSE:
            handleListServiceResponse(serverReflectionResponse.getListServicesResponse());
            break;
        default:
            logger.warn("Got unknown reflection response type: " + responseCase);
            break;
    }
}
 

@Override
public void onNext(ServerReflectionResponse response) {
    MessageResponseCase responseCase = response.getMessageResponseCase();
    switch (responseCase) {
        case FILE_DESCRIPTOR_RESPONSE:
            ImmutableSet<FileDescriptorProto> descriptors =
                    parseDescriptors(response.getFileDescriptorResponse().getFileDescriptorProtoList());
            descriptors.forEach(d -> resolvedDescriptors.put(d.getName(), d));
            descriptors.forEach(this::processDependencies);
            break;
        default:
            logger.warn("Got unknown reflection response type: " + responseCase);
            break;
    }
}