import com.google.protobuf.util.Durations; //导入依赖的package包/类
private CPSPublisherTask(StartRequest request) {
  super(request, "gcloud", MetricsHandler.MetricName.PUBLISH_ACK_LATENCY);
  try {
    this.publisher =
        Publisher.defaultBuilder(TopicName.create(request.getProject(), request.getTopic()))
            .setBatchingSettings(
                BatchingSettings.newBuilder()
                    .setElementCountThreshold(950L)
                    .setRequestByteThreshold(9500000L)
                    .setDelayThreshold(
                        Duration.ofMillis(Durations.toMillis(request.getPublishBatchDuration())))
                    .build())
            .build();
  } catch (Exception e) {
    throw new RuntimeException(e);
  }
  this.payload = ByteString.copyFromUtf8(LoadTestRunner.createMessage(request.getMessageSize()));
  this.batchSize = request.getPublishBatchSize();
  this.messageSize = request.getMessageSize();
  this.id = (new Random()).nextInt();
}
 

import com.google.protobuf.util.Durations; //导入依赖的package包/类
@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());
}
 

import com.google.protobuf.util.Durations; //导入依赖的package包/类
private KafkaPublisherTask(StartRequest request) {
  super(request, "kafka", MetricsHandler.MetricName.PUBLISH_ACK_LATENCY);
  this.topic = request.getTopic();
  this.payload = LoadTestRunner.createMessage(request.getMessageSize());
  this.batchSize = request.getPublishBatchSize();
  Properties props = new Properties();
  props.putAll(new ImmutableMap.Builder<>()
      .put("max.block.ms", "30000")
      .put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
      .put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
      .put("acks", "all")
      .put("bootstrap.servers", request.getKafkaOptions().getBroker())
      .put("buffer.memory", Integer.toString(1000 * 1000 * 1000)) // 1 GB
      // 10M, high enough to allow for duration to control batching
      .put("batch.size", Integer.toString(10 * 1000 * 1000))
      .put("linger.ms", Long.toString(Durations.toMillis(request.getPublishBatchDuration())))
      .build()
  );
  this.publisher = new KafkaProducer<>(props);
}
 

import com.google.protobuf.util.Durations; //导入依赖的package包/类
@Override
public void doMerge(JsonParser parser, int unused, Message.Builder messageBuilder)
    throws IOException {
  Duration.Builder builder = (Duration.Builder) messageBuilder;
  try {
    builder.mergeFrom(Durations.parse(ParseSupport.parseString(parser)));
  } catch (ParseException e) {
    throw new InvalidProtocolBufferException(
        "Failed to readValue duration: " + parser.getText());
  }
}
 

import com.google.protobuf.util.Durations; //导入依赖的package包/类
private void mergeDuration(JsonElement json, Message.Builder builder)
    throws InvalidProtocolBufferException {
  try {
    Duration value = Durations.parse(json.getAsString());
    builder.mergeFrom(value.toByteString());
  } catch (ParseException e) {
    throw new InvalidProtocolBufferException("Failed to parse duration: " + json);
  }
}
 

import com.google.protobuf.util.Durations; //导入依赖的package包/类
private KafkaSubscriberTask(StartRequest request) {
  super(request, "kafka", MetricsHandler.MetricName.END_TO_END_LATENCY);
  this.pollLength = Durations.toMillis(request.getKafkaOptions().getPollDuration());
  Properties props = new Properties();
  props.putAll(ImmutableMap.of(
      "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer",
      "group.id", "SUBSCRIBER_ID",
      "enable.auto.commit", "true",
      "session.timeout.ms", "30000"
  ));
  props.put("bootstrap.servers", request.getKafkaOptions().getBroker());
  subscriber = new KafkaConsumer<>(props);
  subscriber.subscribe(Collections.singletonList(request.getTopic()));
}
 

import com.google.protobuf.util.Durations; //导入依赖的package包/类
@Override
public Duration convert(String value) {
  try {
    if (value.isEmpty()) {
      return Durations.fromMillis(0);
    }
    long millis = 0;
    boolean negative = value.startsWith("-");
    int index = negative ? 1 : 0;
    Pattern unitPattern =
        Pattern.compile(
            "(?x) (?<whole>[0-9]+)? (?<frac>\\.[0-9]*)? (?<unit>d|h|ms?|s)",
            Pattern.CASE_INSENSITIVE);
    Matcher matcher = unitPattern.matcher(value);
    while (matcher.find(index) && matcher.start() == index) {
      Preconditions.checkArgument(CharMatcher.inRange('0', '9').matchesAnyOf(matcher.group(0)));
      long whole = Long.parseLong(MoreObjects.firstNonNull(matcher.group("whole"), "0"));
      double frac =
          Double.parseDouble("0" + MoreObjects.firstNonNull(matcher.group("frac"), ""));
      int millisPerUnit = millisPerUnit(matcher.group("unit"));
      millis += millisPerUnit * whole;
      millis += (long) (millisPerUnit * frac);
      index = matcher.end();
    }
    if (index < value.length()) {
      throw new IllegalArgumentException("Could not parse entire duration");
    }
    if (negative) {
      millis = -millis;
    }
    return Durations.fromMillis(millis);
  } catch (Exception e) {
    throw new ParameterException(
        getErrorString(value, "A duration string must include units (d|h|m|s|ms)."));
  }
}
 

import com.google.protobuf.util.Durations; //导入依赖的package包/类
private static LoadBalanceResponse buildInitialResponse(long loadReportIntervalMillis) {
  return LoadBalanceResponse.newBuilder()
      .setInitialResponse(
          InitialLoadBalanceResponse.newBuilder()
          .setClientStatsReportInterval(Durations.fromMillis(loadReportIntervalMillis)))
      .build();
}
 

import com.google.protobuf.util.Durations; //导入依赖的package包/类
@Override
public void doWrite(Duration message, JsonGenerator gen) throws IOException {
  gen.writeString(Durations.toString(message));
}
 

import com.google.protobuf.util.Durations; //导入依赖的package包/类
@Test
public void duration() throws Exception {
  TestDuration message =
      TestDuration.newBuilder().setDurationValue(Durations.parse("12345s")).build();
  assertMatchesUpstream(message);
}
 

import com.google.protobuf.util.Durations; //导入依赖的package包/类
@Test
public void anyFields() throws Exception {
  TestAllTypes content = TestAllTypes.newBuilder().setOptionalInt32(1234).build();
  TestAny message = TestAny.newBuilder().setAnyValue(Any.pack(content)).build();
  assertMatchesUpstream(message, TestAllTypes.getDefaultInstance());

  TestAny messageWithDefaultAnyValue =
      TestAny.newBuilder().setAnyValue(Any.getDefaultInstance()).build();
  assertMatchesUpstream(messageWithDefaultAnyValue);

  // Well-known types have a special formatting when embedded in Any.
  //
  // 1. Any in Any.
  Any anyMessage = Any.pack(Any.pack(content));
  assertMatchesUpstream(anyMessage, TestAllTypes.getDefaultInstance());

  // 2. Wrappers in Any.
  anyMessage = Any.pack(Int32Value.newBuilder().setValue(12345).build());
  assertMatchesUpstream(anyMessage, TestAllTypes.getDefaultInstance());
  anyMessage = Any.pack(UInt32Value.newBuilder().setValue(12345).build());
  assertMatchesUpstream(anyMessage, TestAllTypes.getDefaultInstance());
  anyMessage = Any.pack(Int64Value.newBuilder().setValue(12345).build());
  assertMatchesUpstream(anyMessage, TestAllTypes.getDefaultInstance());
  anyMessage = Any.pack(UInt64Value.newBuilder().setValue(12345).build());
  assertMatchesUpstream(anyMessage, TestAllTypes.getDefaultInstance());
  anyMessage = Any.pack(FloatValue.newBuilder().setValue(12345).build());
  assertMatchesUpstream(anyMessage, TestAllTypes.getDefaultInstance());
  anyMessage = Any.pack(DoubleValue.newBuilder().setValue(12345).build());
  assertMatchesUpstream(anyMessage, TestAllTypes.getDefaultInstance());
  anyMessage = Any.pack(BoolValue.newBuilder().setValue(true).build());
  assertMatchesUpstream(anyMessage, TestAllTypes.getDefaultInstance());
  anyMessage = Any.pack(StringValue.newBuilder().setValue("Hello").build());
  assertMatchesUpstream(anyMessage, TestAllTypes.getDefaultInstance());
  anyMessage =
      Any.pack(BytesValue.newBuilder().setValue(ByteString.copyFrom(new byte[] {1, 2})).build());
  assertMatchesUpstream(anyMessage, TestAllTypes.getDefaultInstance());

  // 3. Timestamp in Any.
  anyMessage = Any.pack(Timestamps.parse("1969-12-31T23:59:59Z"));
  assertMatchesUpstream(anyMessage, TestAllTypes.getDefaultInstance());

  // 4. Duration in Any
  anyMessage = Any.pack(Durations.parse("12345.10s"));
  assertMatchesUpstream(anyMessage, TestAllTypes.getDefaultInstance());

  // 5. FieldMask in Any
  anyMessage = Any.pack(FieldMaskUtil.fromString("foo.bar,baz"));
  assertMatchesUpstream(anyMessage, TestAllTypes.getDefaultInstance());

  // 6. Struct in Any
  Struct.Builder structBuilder = Struct.newBuilder();
  structBuilder.putFields("number", Value.newBuilder().setNumberValue(1.125).build());
  anyMessage = Any.pack(structBuilder.build());
  assertMatchesUpstream(anyMessage, TestAllTypes.getDefaultInstance());

  // 7. Value (number type) in Any
  Value.Builder valueBuilder = Value.newBuilder();
  valueBuilder.setNumberValue(1);
  anyMessage = Any.pack(valueBuilder.build());
  assertMatchesUpstream(anyMessage, TestAllTypes.getDefaultInstance());

  // 8. Value (null type) in Any
  anyMessage = Any.pack(Value.newBuilder().setNullValue(NullValue.NULL_VALUE).build());
  assertMatchesUpstream(anyMessage, TestAllTypes.getDefaultInstance());
}
 

import com.google.protobuf.util.Durations; //导入依赖的package包/类
/** Prints google.protobuf.Duration */
private void printDuration(MessageOrBuilder message) throws IOException {
  Duration value = Duration.parseFrom(toByteString(message));
  generator.print("\"" + Durations.toString(value) + "\"");
}
 

import com.google.protobuf.util.Durations; //导入依赖的package包/类
/**
 * Converts a {@link WindowFn} into a {@link RunnerApi.MessageWithComponents} where {@link
 * RunnerApi.MessageWithComponents#getFunctionSpec()} is a {@link RunnerApi.FunctionSpec} for the
 * input {@link WindowFn}.
 */
public static SdkFunctionSpec toProto(
    WindowFn<?, ?> windowFn, @SuppressWarnings("unused") SdkComponents components) {
  // TODO: Set environment IDs
  ByteString serializedFn = ByteString.copyFrom(SerializableUtils.serializeToByteArray(windowFn));
  if (windowFn instanceof GlobalWindows) {
    return SdkFunctionSpec.newBuilder()
        .setSpec(FunctionSpec.newBuilder().setUrn(GLOBAL_WINDOWS_FN))
        .build();
  } else if (windowFn instanceof FixedWindows) {
    FixedWindowsPayload fixedWindowsPayload =
        FixedWindowsPayload.newBuilder()
            .setSize(Durations.fromMillis(((FixedWindows) windowFn).getSize().getMillis()))
            .setOffset(Timestamps.fromMillis(((FixedWindows) windowFn).getOffset().getMillis()))
            .build();
    return SdkFunctionSpec.newBuilder()
        .setSpec(
            FunctionSpec.newBuilder()
                .setUrn(FIXED_WINDOWS_FN)
                .setPayload(fixedWindowsPayload.toByteString()))
        .build();
  } else if (windowFn instanceof SlidingWindows) {
    SlidingWindowsPayload slidingWindowsPayload = SlidingWindowsPayload.newBuilder()
        .setSize(Durations.fromMillis(((SlidingWindows) windowFn).getSize().getMillis()))
        .setOffset(Timestamps.fromMillis(((SlidingWindows) windowFn).getOffset().getMillis()))
        .setPeriod(Durations.fromMillis(((SlidingWindows) windowFn).getPeriod().getMillis()))
        .build();
    return SdkFunctionSpec.newBuilder()
        .setSpec(
            FunctionSpec.newBuilder()
                .setUrn(SLIDING_WINDOWS_FN)
                .setPayload(slidingWindowsPayload.toByteString()))
        .build();
  } else if (windowFn instanceof Sessions) {
    SessionsPayload sessionsPayload =
        SessionsPayload.newBuilder()
            .setGapSize(Durations.fromMillis(((Sessions) windowFn).getGapDuration().getMillis()))
            .build();
    return SdkFunctionSpec.newBuilder()
        .setSpec(
            FunctionSpec.newBuilder()
                .setUrn(SESSION_WINDOWS_FN)
                .setPayload(sessionsPayload.toByteString()))
        .build();
  } else {
    return SdkFunctionSpec.newBuilder()
        .setSpec(
            FunctionSpec.newBuilder()
                .setUrn(SERIALIZED_JAVA_WINDOWFN_URN)
                .setPayload(serializedFn))
        .build();
  }
}
 

import com.google.protobuf.util.Durations; //导入依赖的package包/类
public static WindowFn<?, ?> windowFnFromProto(SdkFunctionSpec windowFnSpec) {
  try {
    switch (windowFnSpec.getSpec().getUrn()) {
      case GLOBAL_WINDOWS_FN:
        return new GlobalWindows();
      case FIXED_WINDOWS_FN:
        StandardWindowFns.FixedWindowsPayload fixedParams = null;
        fixedParams =
            StandardWindowFns.FixedWindowsPayload.parseFrom(
                windowFnSpec.getSpec().getPayload());
        return FixedWindows.of(Duration.millis(Durations.toMillis(fixedParams.getSize())))
            .withOffset(Duration.millis(Timestamps.toMillis(fixedParams.getOffset())));
      case SLIDING_WINDOWS_FN:
        StandardWindowFns.SlidingWindowsPayload slidingParams =
            StandardWindowFns.SlidingWindowsPayload.parseFrom(
                windowFnSpec.getSpec().getPayload());
        return SlidingWindows.of(Duration.millis(Durations.toMillis(slidingParams.getSize())))
            .every(Duration.millis(Durations.toMillis(slidingParams.getPeriod())))
            .withOffset(Duration.millis(Timestamps.toMillis(slidingParams.getOffset())));
      case SESSION_WINDOWS_FN:
        StandardWindowFns.SessionsPayload sessionParams =
            StandardWindowFns.SessionsPayload.parseFrom(windowFnSpec.getSpec().getPayload());
        return Sessions.withGapDuration(
            Duration.millis(Durations.toMillis(sessionParams.getGapSize())));
      case SERIALIZED_JAVA_WINDOWFN_URN:
      case OLD_SERIALIZED_JAVA_WINDOWFN_URN:
        return (WindowFn<?, ?>)
            SerializableUtils.deserializeFromByteArray(
                windowFnSpec.getSpec().getPayload().toByteArray(), "WindowFn");
      default:
        throw new IllegalArgumentException(
            "Unknown or unsupported WindowFn: " + windowFnSpec.getSpec().getUrn());
    }
  } catch (InvalidProtocolBufferException e) {
    throw new IllegalArgumentException(
        String.format(
            "%s for %s with URN %s did not contain expected proto message for payload",
            FunctionSpec.class.getSimpleName(),
            WindowFn.class.getSimpleName(),
            windowFnSpec.getSpec().getUrn()),
        e);
  }
}
 

import com.google.protobuf.util.Durations; //导入依赖的package包/类
private void handleResponse(LoadBalanceResponse response) {
  if (closed) {
    return;
  }
  logger.log(Level.FINE, "[{0}] Got an LB response: {1}", new Object[] {logId, response});

  LoadBalanceResponseTypeCase typeCase = response.getLoadBalanceResponseTypeCase();
  if (!initialResponseReceived) {
    if (typeCase != LoadBalanceResponseTypeCase.INITIAL_RESPONSE) {
      logger.log(
          Level.WARNING,
          "[{0}] : Did not receive response with type initial response: {1}",
          new Object[] {logId, response});
      return;
    }
    initialResponseReceived = true;
    InitialLoadBalanceResponse initialResponse = response.getInitialResponse();
    loadReportIntervalMillis =
        Durations.toMillis(initialResponse.getClientStatsReportInterval());
    scheduleNextLoadReport();
    return;
  }

  if (typeCase != LoadBalanceResponseTypeCase.SERVER_LIST) {
    logger.log(
        Level.WARNING,
        "[{0}] : Ignoring unexpected response type: {1}",
        new Object[] {logId, response});
    return;
  }

  balancerWorking = true;
  // TODO(zhangkun83): handle delegate from initialResponse
  ServerList serverList = response.getServerList();
  List<DropEntry> newDropList = new ArrayList<DropEntry>();
  List<BackendAddressGroup> newBackendAddrList = new ArrayList<BackendAddressGroup>();
  // Construct the new collections. Create new Subchannels when necessary.
  for (Server server : serverList.getServersList()) {
    String token = server.getLoadBalanceToken();
    if (server.getDrop()) {
      newDropList.add(new DropEntry(loadRecorder, token));
    } else {
      newDropList.add(null);
      InetSocketAddress address;
      try {
        address = new InetSocketAddress(
            InetAddress.getByAddress(server.getIpAddress().toByteArray()), server.getPort());
      } catch (UnknownHostException e) {
        propagateError(
            Status.UNAVAILABLE
                .withDescription("Host for server not found: " + server)
                .withCause(e));
        continue;
      }
      EquivalentAddressGroup eag = new EquivalentAddressGroup(address);
      newBackendAddrList.add(new BackendAddressGroup(eag, token));
    }
  }
  // Stop using fallback backends as soon as a new server list is received from the balancer.
  usingFallbackBackends = false;
  cancelFallbackTimer();
  useRoundRobinLists(newDropList, newBackendAddrList, loadRecorder);
  maybeUpdatePicker();
}