@Override
public Flux<Account> findByStatus(Int32Value status) {
    return Flux.just(Account.newBuilder().setId(new Random().nextInt())
                    .setEmail(faker.internet().emailAddress())
                    .setPhone(faker.phoneNumber().cellPhone())
                    .setNick(faker.name().name())
                    .setStatus(status.getValue())
                    .build(),
            Account.newBuilder().setId(new Random().nextInt())
                    .setEmail(faker.internet().emailAddress())
                    .setPhone(faker.phoneNumber().cellPhone())
                    .setNick(faker.name().name())
                    .setStatus(status.getValue())
                    .build()
    );
}
 

@Test
public void testEncodesLimits() {
  Query q = Query.atPath(ResourcePath.fromString("docs")).limitToFirst(26);
  Target actual = serializer.encodeTarget(wrapTargetData(q));

  StructuredQuery.Builder structuredQueryBuilder =
      StructuredQuery.newBuilder()
          .addFrom(CollectionSelector.newBuilder().setCollectionId("docs"))
          .addOrderBy(defaultKeyOrder())
          .setLimit(Int32Value.newBuilder().setValue(26));
  QueryTarget.Builder queryBuilder =
      QueryTarget.newBuilder()
          .setParent("projects/p/databases/d/documents")
          .setStructuredQuery(structuredQueryBuilder);
  Target expected =
      Target.newBuilder()
          .setQuery(queryBuilder)
          .setTargetId(1)
          .setResumeToken(ByteString.EMPTY)
          .build();

  assertEquals(expected, actual);
  assertEquals(
      serializer.decodeQueryTarget(serializer.encodeQueryTarget(q.toTarget())), q.toTarget());
}
 

/**
 * Adds L+ targeting to the Apk targeting of module split. If SDK targeting already exists, it's
 * not overridden but checked that it targets no L- devices.
 */
private ModuleSplit addLPlusApkTargeting(ModuleSplit split) {
  if (split.getApkTargeting().hasSdkVersionTargeting()) {
    checkState(
        split.getApkTargeting().getSdkVersionTargeting().getValue(0).getMin().getValue()
            >= ANDROID_L_API_VERSION,
        "Module Split should target SDK versions above L.");
    return split;
  }

  return split
      .toBuilder()
      .setApkTargeting(
          split
              .getApkTargeting()
              .toBuilder()
              .setSdkVersionTargeting(
                  SdkVersionTargeting.newBuilder()
                      .addValue(
                          SdkVersion.newBuilder()
                              .setMin(Int32Value.newBuilder().setValue(ANDROID_L_API_VERSION))))
              .build())
      .build();
}
 

private static ModuleSplit addLPlusApkTargeting(ModuleSplit split) {
  if (split.getApkTargeting().hasSdkVersionTargeting()) {
    checkState(
        split.getApkTargeting().getSdkVersionTargeting().getValue(0).getMin().getValue()
            >= ANDROID_L_API_VERSION,
        "Module Split should target SDK versions above L.");
    return split;
  }

  return split.toBuilder()
      .setApkTargeting(
          split.getApkTargeting().toBuilder()
              .setSdkVersionTargeting(
                  SdkVersionTargeting.newBuilder()
                      .addValue(
                          SdkVersion.newBuilder()
                              .setMin(Int32Value.newBuilder().setValue(ANDROID_L_API_VERSION))))
              .build())
      .build();
}
 

@Test
public void anyInMaps() throws Exception {
  TestAny.Builder testAny = TestAny.newBuilder();
  testAny.putAnyMap("int32_wrapper", Any.pack(Int32Value.newBuilder().setValue(123).build()));
  testAny.putAnyMap("int64_wrapper", Any.pack(Int64Value.newBuilder().setValue(456).build()));
  testAny.putAnyMap("timestamp", Any.pack(Timestamps.parse("1969-12-31T23:59:59Z")));
  testAny.putAnyMap("duration", Any.pack(Durations.parse("12345.1s")));
  testAny.putAnyMap("field_mask", Any.pack(FieldMaskUtil.fromString("foo.bar,baz")));
  Value numberValue = Value.newBuilder().setNumberValue(1.125).build();
  Struct.Builder struct = Struct.newBuilder();
  struct.putFields("number", numberValue);
  testAny.putAnyMap("struct", Any.pack(struct.build()));
  Value nullValue = Value.newBuilder().setNullValue(NullValue.NULL_VALUE).build();
  testAny.putAnyMap(
      "list_value",
      Any.pack(ListValue.newBuilder().addValues(numberValue).addValues(nullValue).build()));
  testAny.putAnyMap("number_value", Any.pack(numberValue));
  testAny.putAnyMap("any_value_number", Any.pack(Any.pack(numberValue)));
  testAny.putAnyMap("any_value_default", Any.pack(Any.getDefaultInstance()));
  testAny.putAnyMap("default", Any.getDefaultInstance());

  assertMatchesUpstream(testAny.build(), TestAllTypes.getDefaultInstance());
}
 

private static StepAdjustments toStepAdjustments(StepAdjustment stepAdjustment) {
    StepAdjustments.Builder stepAdjustmentsBuilder = StepAdjustments.newBuilder()
            .setScalingAdjustment(Int32Value.newBuilder()
                    .setValue(stepAdjustment.getScalingAdjustment())
                    .build());
    stepAdjustment.getMetricIntervalLowerBound().ifPresent(
            metricIntervalLowerBound ->
                    stepAdjustmentsBuilder.setMetricIntervalLowerBound(DoubleValue.newBuilder()
                            .setValue(metricIntervalLowerBound)
                            .build())
    );
    stepAdjustment.getMetricIntervalUpperBound().ifPresent(
            metricIntervalUpperBound ->
                    stepAdjustmentsBuilder.setMetricIntervalUpperBound(DoubleValue.newBuilder()
                            .setValue(metricIntervalUpperBound)
                            .build())
    );

    return stepAdjustmentsBuilder.build();
}
 

/**
 * Cloud Datastore system tables with statistics are periodically updated. This method fetches
 * the latest timestamp (in microseconds) of statistics update using the {@code __Stat_Total__}
 * table.
 */
private static long queryLatestStatisticsTimestamp(
    Datastore datastore, @Nullable String namespace) throws DatastoreException {
  Query.Builder query = Query.newBuilder();
  // Note: namespace either being null or empty represents the default namespace, in which
  // case we treat it as not provided by the user.
  if (Strings.isNullOrEmpty(namespace)) {
    query.addKindBuilder().setName("__Stat_Total__");
  } else {
    query.addKindBuilder().setName("__Stat_Ns_Total__");
  }
  query.addOrder(makeOrder("timestamp", DESCENDING));
  query.setLimit(Int32Value.newBuilder().setValue(1));
  RunQueryRequest request = makeRequest(query.build(), namespace);

  RunQueryResponse response = datastore.runQuery(request);
  QueryResultBatch batch = response.getBatch();
  if (batch.getEntityResultsCount() == 0) {
    throw new NoSuchElementException("Datastore total statistics unavailable");
  }
  Entity entity = batch.getEntityResults(0).getEntity();
  return entity.getProperties().get("timestamp").getTimestampValue().getSeconds() * 1000000;
}
 

/** Tests that {@link ReadFn} retries after an error. */
@Test
public void testReadFnRetriesErrors() throws Exception {
  // An empty query to read entities.
  Query query = Query.newBuilder().setLimit(Int32Value.newBuilder().setValue(1)).build();

  // Use mockResponseForQuery to generate results.
  when(mockDatastore.runQuery(any(RunQueryRequest.class)))
      .thenThrow(new DatastoreException("RunQuery", Code.DEADLINE_EXCEEDED, "", null))
      .thenAnswer(
          invocationOnMock -> {
            Query q = ((RunQueryRequest) invocationOnMock.getArguments()[0]).getQuery();
            return mockResponseForQuery(q);
          });

  ReadFn readFn = new ReadFn(V_1_OPTIONS, mockDatastoreFactory);
  DoFnTester<Query, Entity> doFnTester = DoFnTester.of(readFn);
  doFnTester.setCloningBehavior(CloningBehavior.DO_NOT_CLONE);
  doFnTester.processBundle(query);
}
 

private Iterator<EntityResult> getIteratorAndMoveCursor() throws DatastoreException {
  Query.Builder query = this.query.toBuilder();
  query.setLimit(Int32Value.newBuilder().setValue(QUERY_BATCH_LIMIT));
  if (currentBatch != null && !currentBatch.getEndCursor().isEmpty()) {
    query.setStartCursor(currentBatch.getEndCursor());
  }

  RunQueryRequest request = makeRequest(query.build(), namespace);
  RunQueryResponse response = datastore.runQuery(request);

  currentBatch = response.getBatch();

  int numFetch = currentBatch.getEntityResultsCount();
  // All indications from the API are that there are/may be more results.
  moreResults =
      (numFetch == QUERY_BATCH_LIMIT) || (currentBatch.getMoreResults() == NOT_FINISHED);

  // May receive a batch of 0 results if the number of records is a multiple
  // of the request limit.
  if (numFetch == 0) {
    return null;
  }

  return currentBatch.getEntityResultsList().iterator();
}
 

@Override
public void errorWithMessage(SimpleRequest request, StreamObserver<SimpleResponse> responseObserver) {
    final Metadata metadata = new Metadata();
    metadata.put(STRING_VALUE_KEY, StringValue.newBuilder().setValue("custom metadata").build());
    metadata.put(CUSTOM_VALUE_KEY, "custom value");

    final ServiceRequestContext ctx = ServiceRequestContext.current();
    // gRPC wire format allow comma-separated binary headers.
    ctx.mutateAdditionalResponseTrailers(
            mutator -> mutator.add(
                    INT_32_VALUE_KEY.name(),
                    Base64.getEncoder().encodeToString(
                            Int32Value.newBuilder().setValue(10).build().toByteArray()) +
                    ',' +
                    Base64.getEncoder().encodeToString(
                            Int32Value.newBuilder().setValue(20).build().toByteArray())));

    responseObserver.onError(Status.ABORTED.withDescription("aborted call").asException(metadata));
}
 

@ParameterizedTest
@ArgumentsSource(BlockingClientProvider.class)
void error_withMessage(UnitTestServiceBlockingStub blockingClient) throws Exception {
    final StatusRuntimeException t = (StatusRuntimeException) catchThrowable(
            () -> blockingClient.errorWithMessage(REQUEST_MESSAGE));
    assertThat(t.getStatus().getCode()).isEqualTo(Code.ABORTED);
    assertThat(t.getStatus().getDescription()).isEqualTo("aborted call");
    assertThat(t.getTrailers().getAll(STRING_VALUE_KEY))
            .containsExactly(StringValue.newBuilder().setValue("custom metadata").build());
    assertThat(t.getTrailers().getAll(INT_32_VALUE_KEY))
            .containsExactly(Int32Value.newBuilder().setValue(10).build(),
                             Int32Value.newBuilder().setValue(20).build());
    assertThat(t.getTrailers().get(CUSTOM_VALUE_KEY)).isEqualTo("custom value");

    checkRequestLog((rpcReq, rpcRes, grpcStatus) -> {
        assertThat(rpcReq.method()).isEqualTo("armeria.grpc.testing.UnitTestService/ErrorWithMessage");
        assertThat(rpcReq.params()).containsExactly(REQUEST_MESSAGE);
        assertThat(grpcStatus).isNotNull();
        assertThat(grpcStatus.getCode()).isEqualTo(Code.ABORTED);
        assertThat(grpcStatus.getDescription()).isEqualTo("aborted call");
    });
}
 

@Test
public void itSetsFieldsWhenZeroInJson() throws IOException {
  String json = camelCase().writeValueAsString(defaultPopulatedJsonNode(camelCase()));
  HasWrappedPrimitives message = camelCase().readValue(json, HasWrappedPrimitives.class);
  assertThat(message.hasDoubleWrapper()).isTrue();
  assertThat(message.getDoubleWrapper()).isEqualTo(DoubleValue.getDefaultInstance());
  assertThat(message.hasFloatWrapper()).isTrue();
  assertThat(message.getFloatWrapper()).isEqualTo(FloatValue.getDefaultInstance());
  assertThat(message.hasInt64Wrapper()).isTrue();
  assertThat(message.getInt64Wrapper()).isEqualTo(Int64Value.getDefaultInstance());
  assertThat(message.hasUint64Wrapper()).isTrue();
  assertThat(message.getUint64Wrapper()).isEqualTo(UInt64Value.getDefaultInstance());
  assertThat(message.hasInt32Wrapper()).isTrue();
  assertThat(message.getInt32Wrapper()).isEqualTo(Int32Value.getDefaultInstance());
  assertThat(message.hasUint32Wrapper()).isTrue();
  assertThat(message.getUint32Wrapper()).isEqualTo(UInt32Value.getDefaultInstance());
  assertThat(message.hasBoolWrapper()).isTrue();
  assertThat(message.getBoolWrapper()).isEqualTo(BoolValue.getDefaultInstance());
  assertThat(message.hasStringWrapper()).isTrue();
  assertThat(message.getStringWrapper()).isEqualTo(StringValue.getDefaultInstance());
  assertThat(message.hasBytesWrapper()).isTrue();
  assertThat(message.getBytesWrapper()).isEqualTo(BytesValue.getDefaultInstance());
}
 

@Override
public Mono<Account> findById(Int32Value id) {
    return Mono.just(Account.newBuilder().setId(id.getValue())
            .setEmail(faker.internet().emailAddress())
            .setPhone(faker.phoneNumber().cellPhone())
            .setNick(faker.name().name())
            .build());
}
 

@Override
public Flux<Account> findByIdStream(Flux<Int32Value> idStream) {
    return idStream.map(id -> Account.newBuilder()
            .setId(id.getValue())
            .setEmail(faker.internet().emailAddress())
            .setPhone(faker.phoneNumber().cellPhone())
            .setNick(faker.name().name())
            .setStatus(1)
            .build());
}
 

@Test
public void testGrpcProxy() throws Exception {
    DemoInterceptor interceptor = new DemoInterceptor();
    Class<ReactorAccountServiceGrpc.AccountServiceImplBase> dynamicType = (Class<ReactorAccountServiceGrpc.AccountServiceImplBase>) new ByteBuddy()
            .subclass(ReactorAccountServiceGrpc.AccountServiceImplBase.class)
            .method(ElementMatchers.returns(target -> target.isAssignableFrom(Mono.class) || target.isAssignableFrom(Flux.class)))
            .intercept(MethodDelegation.to(interceptor))
            .make()
            .load(getClass().getClassLoader())
            .getLoaded();
    System.out.println(dynamicType.newInstance().findById(Mono.just(Int32Value.newBuilder().setValue(1).build())).block());
}
 

/**
 * Creates a new AdScheduleInfo with the specified attributes.
 *
 * @param day day of the week of the AdScheduleInfo.
 * @param startHour the starting hour of the AdScheduleInfo.
 * @param startMinute the starting minute of the AdScheduleInfo.
 * @param endHour the ending hour of the AdScheduleInfo.
 * @param endMinute ending minute of the AdScheduleInfo.
 */
private static AdScheduleInfo createAdScheduleInfo(
    DayOfWeek day, int startHour, MinuteOfHour startMinute, int endHour, MinuteOfHour endMinute) {
  return AdScheduleInfo.newBuilder()
      .setDayOfWeek(day)
      .setStartHour(Int32Value.of(startHour))
      .setStartMinute(startMinute)
      .setEndHour(Int32Value.of(endHour))
      .setEndMinute(endMinute)
      .build();
}
 

/**
 * Creates a new ad schedule info with the specified parameters.
 *
 * @param dayOfWeek the day of week for which the schedule is enabled.
 * @param startHour the hour at which the schedule takes effect.
 * @param startMinute the minute past the hour at which the schedule takes effect.
 * @param endHour the hour at which the schedule stops running.
 * @param endMinute the minute past the hour at which the schedule stops running.
 * @return a newly created ad schedule object.
 */
private AdScheduleInfo createAdSchedule(
    DayOfWeek dayOfWeek,
    int startHour,
    MinuteOfHour startMinute,
    int endHour,
    MinuteOfHour endMinute) {
  return AdScheduleInfo.newBuilder()
      .setDayOfWeek(dayOfWeek)
      .setStartHour(Int32Value.of(startHour))
      .setStartMinute(startMinute)
      .setEndHour(Int32Value.of(endHour))
      .setEndMinute(endMinute)
      .build();
}
 

@Test
public void execute_skipUnsupportedSdks() throws Exception {
  // GIVEN a .apks containing containing only targets that are greater than the device SDK...
  BuildApksResult apexTableOfContents =
      BuildApksResult.newBuilder()
          .setPackageName(PKG_NAME_1)
          .setBundletool(
              Bundletool.newBuilder()
                  .setVersion(BundleToolVersion.getCurrentVersion().toString()))
          .addVariant(
              apexVariant(
                  VariantTargeting.newBuilder()
                      .setSdkVersionTargeting(
                          SdkVersionTargeting.newBuilder()
                              .addValue(
                                  SdkVersion.newBuilder()
                                      .setMin(Int32Value.of(ANDROID_Q_API_VERSION + 3)))
                              .build())
                      .build(),
                  ApkTargeting.getDefaultInstance(),
                  ZipPath.create(PKG_NAME_1 + "base.apex")))
          .build();
  Path package1Apks = createApksArchiveFile(apexTableOfContents, tmpDir.resolve("package1.apks"));

  InstallMultiApksCommand command =
      InstallMultiApksCommand.builder()
          .setAdbServer(fakeServerOneDevice(device))
          .setDeviceId(DEVICE_ID)
          .setAdbPath(adbPath)
          .setApksArchivePaths(ImmutableList.of(package1Apks))
          .setAapt2Command(createFakeAapt2Command(ImmutableMap.of(PKG_NAME_1, 1L)))
          .build();

  // EXPECT to check the existing list of packages...
  givenEmptyListPackages(device);

  // THEN the command executes without triggering any other shell commands.
  command.execute();
}
 

@BeforeEach
void setUp() {
  cache =
      new ProtobufRedisLoadingCache<>(
          StringValue.getDefaultInstance(),
          Int32Value.getDefaultInstance(),
          TTL,
          null,
          remoteCache);
}
 

private static AlarmConfiguration toAlarmConfiguration(com.netflix.titus.api.appscale.model.AlarmConfiguration alarmConfiguration) {
    AlarmConfiguration.Builder alarmConfigBuilder = AlarmConfiguration.newBuilder();
    alarmConfiguration.getActionsEnabled().ifPresent(
            actionsEnabled ->
                    alarmConfigBuilder.setActionsEnabled(BoolValue.newBuilder()
                            .setValue(actionsEnabled)
                            .build())
    );

    AlarmConfiguration.ComparisonOperator comparisonOperator =
            toComparisonOperator(alarmConfiguration.getComparisonOperator());
    AlarmConfiguration.Statistic statistic =
            toStatistic(alarmConfiguration.getStatistic());
    return AlarmConfiguration.newBuilder()
            .setComparisonOperator(comparisonOperator)
            .setEvaluationPeriods(Int32Value.newBuilder()
                    .setValue(alarmConfiguration.getEvaluationPeriods())
                    .build())
            .setPeriodSec(Int32Value.newBuilder()
                    .setValue(alarmConfiguration.getPeriodSec())
                    .build())
            .setThreshold(DoubleValue.newBuilder()
                    .setValue(alarmConfiguration.getThreshold())
                    .build())
            .setMetricNamespace(alarmConfiguration.getMetricNamespace())
            .setMetricName(alarmConfiguration.getMetricName())
            .setStatistic(statistic)
            .build();
}
 

private static StepScalingPolicy toStepScalingPolicy(StepScalingPolicyConfiguration stepScalingPolicyConfiguration) {
    StepScalingPolicy.Builder stepScalingPolicyBuilder = StepScalingPolicy.newBuilder();

    stepScalingPolicyConfiguration.getCoolDownSec().ifPresent(
            coolDown ->
                    stepScalingPolicyBuilder.setCooldownSec(Int32Value.newBuilder().setValue(coolDown).build())
    );
    stepScalingPolicyConfiguration.getMetricAggregationType().ifPresent(
            metricAggregationType ->
                    stepScalingPolicyBuilder.setMetricAggregationType(toMetricAggregationType(metricAggregationType))
    );
    stepScalingPolicyConfiguration.getAdjustmentType().ifPresent(
            stepAdjustmentType ->
                    stepScalingPolicyBuilder.setAdjustmentType(toAdjustmentType(stepAdjustmentType))
    );
    stepScalingPolicyConfiguration.getMinAdjustmentMagnitude().ifPresent(
            minAdjustmentMagnitude ->
                    stepScalingPolicyBuilder.setMinAdjustmentMagnitude(
                            Int64Value.newBuilder()
                                    .setValue(minAdjustmentMagnitude)
                                    .build())
    );
    stepScalingPolicyBuilder.addAllStepAdjustments(toStepAdjustmentsList(stepScalingPolicyConfiguration.getSteps()));

    return stepScalingPolicyBuilder
            .build();
}
 

private static TargetTrackingPolicyDescriptor toTargetTrackingPolicyDescriptor(TargetTrackingPolicy targetTrackingPolicy) {
    TargetTrackingPolicyDescriptor.Builder targetTrackingPolicyDescBuilder = TargetTrackingPolicyDescriptor.newBuilder();
    targetTrackingPolicyDescBuilder.setTargetValue(DoubleValue.newBuilder()
            .setValue(targetTrackingPolicy.getTargetValue())
            .build());

    targetTrackingPolicy.getScaleOutCooldownSec().ifPresent(
            scaleOutCoolDownSec -> targetTrackingPolicyDescBuilder.setScaleOutCooldownSec(
                    Int32Value.newBuilder().setValue(scaleOutCoolDownSec).build())
    );
    targetTrackingPolicy.getScaleInCooldownSec().ifPresent(
            scaleInCoolDownSec -> targetTrackingPolicyDescBuilder.setScaleInCooldownSec(
                    Int32Value.newBuilder().setValue(scaleInCoolDownSec).build())
    );
    targetTrackingPolicy.getDisableScaleIn().ifPresent(
            disableScaleIn -> targetTrackingPolicyDescBuilder.setDisableScaleIn(
                    BoolValue.newBuilder().setValue(disableScaleIn).build())
    );
    targetTrackingPolicy.getPredefinedMetricSpecification().ifPresent(
            predefinedMetricSpecification ->
                    targetTrackingPolicyDescBuilder.setPredefinedMetricSpecification(
                            toPredefinedMetricSpecification(targetTrackingPolicy.getPredefinedMetricSpecification().get()))
    );
    targetTrackingPolicy.getCustomizedMetricSpecification().ifPresent(
            customizedMetricSpecification ->
                    targetTrackingPolicyDescBuilder.setCustomizedMetricSpecification(
                            toCustomizedMetricSpecification(targetTrackingPolicy.getCustomizedMetricSpecification().get()))
    );

    return targetTrackingPolicyDescBuilder.build();
}
 

public static ScalingPolicy generateTargetPolicy() {
    CustomizedMetricSpecification customizedMetricSpec = CustomizedMetricSpecification.newBuilder()
            .addDimensions(MetricDimension.newBuilder()
                    .setName("testName")
                    .setValue("testValue")
                    .build())
            .setMetricName("testMetric")
            .setNamespace("NFLX/EPIC")
            .setStatistic(AlarmConfiguration.Statistic.Sum)
            .setMetricName("peanuts")
            .build();

    TargetTrackingPolicyDescriptor targetTrackingPolicyDescriptor = TargetTrackingPolicyDescriptor.newBuilder()
            .setTargetValue(DoubleValue.newBuilder()
                    .setValue(ThreadLocalRandom.current().nextDouble())
                    .build())
            .setScaleInCooldownSec(Int32Value.newBuilder()
                    .setValue(ThreadLocalRandom.current().nextInt())
                    .build())
            .setScaleOutCooldownSec(Int32Value.newBuilder()
                    .setValue(ThreadLocalRandom.current().nextInt())
                    .build())
            .setDisableScaleIn(BoolValue.newBuilder()
                    .setValue(false)
                    .build())
            .setCustomizedMetricSpecification(customizedMetricSpec)
            .build();
    return ScalingPolicy.newBuilder().setTargetPolicyDescriptor(targetTrackingPolicyDescriptor).build();
}
 

@Override
public Object execute(Object[] parameters) {
	StructuredQuery.Builder builder = createBuilderWithFilter(parameters);

	// Handle Pageable parameters.
	if (!getQueryMethod().getParameters().isEmpty()) {
		ParameterAccessor paramAccessor = new ParametersParameterAccessor(getQueryMethod().getParameters(),
				parameters);
		Pageable pageable = paramAccessor.getPageable();
		if (pageable != null && pageable.isPaged()) {
			builder.setOffset((int) Math.min(Integer.MAX_VALUE, pageable.getOffset()));
			builder.setLimit(Int32Value.newBuilder().setValue(pageable.getPageSize()));
		}

		Sort sort = paramAccessor.getSort();
		if (sort != null) {
			builder.addAllOrderBy(createFirestoreSortOrders(sort));
		}
	}

	if (this.tree.isCountProjection()) {
		return this.reactiveOperations.count(this.persistentEntity.getType(), builder);
	}
	else {
		return this.reactiveOperations.execute(builder, this.persistentEntity.getType());
	}
}
 

/**
 * Creates additional types (Value, Struct and ListValue) to be added to the Service config.
 * TODO (guptasu): Fix this hack. Find a better way to add the predefined types.
 * TODO (guptasu): Add them only when required and not in all cases.
 */

static Iterable<Type> createAdditionalServiceTypes() {
  Map<String, DescriptorProto> additionalMessages = Maps.newHashMap();
  additionalMessages.put(Struct.getDescriptor().getFullName(),
      Struct.getDescriptor().toProto());
  additionalMessages.put(Value.getDescriptor().getFullName(),
      Value.getDescriptor().toProto());
  additionalMessages.put(ListValue.getDescriptor().getFullName(),
      ListValue.getDescriptor().toProto());
  additionalMessages.put(Empty.getDescriptor().getFullName(),
      Empty.getDescriptor().toProto());
  additionalMessages.put(Int32Value.getDescriptor().getFullName(),
      Int32Value.getDescriptor().toProto());
  additionalMessages.put(DoubleValue.getDescriptor().getFullName(),
      DoubleValue.getDescriptor().toProto());
  additionalMessages.put(BoolValue.getDescriptor().getFullName(),
      BoolValue.getDescriptor().toProto());
  additionalMessages.put(StringValue.getDescriptor().getFullName(),
      StringValue.getDescriptor().toProto());

  for (Descriptor descriptor : Struct.getDescriptor().getNestedTypes()) {
    additionalMessages.put(descriptor.getFullName(), descriptor.toProto());
  }

  // TODO (guptasu): Remove this hard coding. Without this, creation of Model from Service throws.
  // Needs investigation.
  String fileName = "struct.proto";
  List<Type> additionalTypes = Lists.newArrayList();
  for (String typeName : additionalMessages.keySet()) {
    additionalTypes.add(TypesBuilderFromDescriptor.createType(typeName,
        additionalMessages.get(typeName), fileName));
  }
  return additionalTypes;
}
 

@Test
public void testReadValidationFailsQueryLimitZero() throws Exception {
  Query invalidLimit = Query.newBuilder().setLimit(Int32Value.newBuilder().setValue(0)).build();
  thrown.expect(IllegalArgumentException.class);
  thrown.expectMessage("Invalid query limit 0: must be positive");

  DatastoreIO.v1().read().withQuery(invalidLimit);
}
 

@Test
public void testReadValidationFailsQueryLimitNegative() throws Exception {
  Query invalidLimit = Query.newBuilder().setLimit(Int32Value.newBuilder().setValue(-5)).build();
  thrown.expect(IllegalArgumentException.class);
  thrown.expectMessage("Invalid query limit -5: must be positive");

  DatastoreIO.v1().read().withQuery(invalidLimit);
}
 

/** Tests {@link DatastoreV1.Read.SplitQueryFn} when the query has a user specified limit. */
@Test
public void testSplitQueryFnWithQueryLimit() throws Exception {
  Query queryWithLimit = QUERY.toBuilder().setLimit(Int32Value.newBuilder().setValue(1)).build();

  SplitQueryFn splitQueryFn = new SplitQueryFn(V_1_OPTIONS, 10, mockDatastoreFactory);
  DoFnTester<Query, Query> doFnTester = DoFnTester.of(splitQueryFn);
  doFnTester.setCloningBehavior(CloningBehavior.DO_NOT_CLONE);
  List<Query> queries = doFnTester.processBundle(queryWithLimit);

  assertEquals(1, queries.size());
  verifyNoMoreInteractions(mockDatastore);
  verifyNoMoreInteractions(mockQuerySplitter);
}
 

/** Helper function to run a test reading from a {@link ReadFn}. */
private void readFnTest(int numEntities) throws Exception {
  // An empty query to read entities.
  Query query =
      Query.newBuilder().setLimit(Int32Value.newBuilder().setValue(numEntities)).build();

  // Use mockResponseForQuery to generate results.
  when(mockDatastore.runQuery(any(RunQueryRequest.class)))
      .thenAnswer(
          invocationOnMock -> {
            Query q = ((RunQueryRequest) invocationOnMock.getArguments()[0]).getQuery();
            return mockResponseForQuery(q);
          });

  ReadFn readFn = new ReadFn(V_1_OPTIONS, mockDatastoreFactory);
  DoFnTester<Query, Entity> doFnTester = DoFnTester.of(readFn);
  /**
   * Although Datastore client is marked transient in {@link ReadFn}, when injected through mock
   * factory using a when clause for unit testing purposes, it is not serializable because it
   * doesn't have a no-arg constructor. Thus disabling the cloning to prevent the test object from
   * being serialized.
   */
  doFnTester.setCloningBehavior(CloningBehavior.DO_NOT_CLONE);
  List<Entity> entities = doFnTester.processBundle(query);

  int expectedNumCallsToRunQuery = (int) Math.ceil((double) numEntities / QUERY_BATCH_LIMIT);
  verify(mockDatastore, times(expectedNumCallsToRunQuery)).runQuery(any(RunQueryRequest.class));
  // Validate the number of results.
  assertEquals(numEntities, entities.size());
}
 

/** Builds a latest timestamp statistics query. */
private static Query makeLatestTimestampQuery(String namespace) {
  Query.Builder timestampQuery = Query.newBuilder();
  if (namespace == null) {
    timestampQuery.addKindBuilder().setName("__Stat_Total__");
  } else {
    timestampQuery.addKindBuilder().setName("__Stat_Ns_Total__");
  }
  timestampQuery.addOrder(makeOrder("timestamp", DESCENDING));
  timestampQuery.setLimit(Int32Value.newBuilder().setValue(1));
  return timestampQuery.build();
}