org.joda.time.Instant#parse ( )源码实例Demo

/**
 * input - a tupel that contains the data element (TableRow), the window, the timestamp, and the pane
 */

@Override
public TableDestination apply(ValueInSingleWindow<TableRow> input) {
    
	String partition;
	
	if (this.isTimeField) {
     String sTime = (String) input.getValue().get(this.fieldName);
     Instant time = Instant.parse(sTime);
     partition = time.toString(partitionFormatter);
	} else {
		partition = ((Integer) input.getValue().get(this.fieldName)).toString();
	}
	
    TableReference reference = new TableReference();
    reference.setProjectId(this.projectId);
    reference.setDatasetId(this.datasetId);
    reference.setTableId(this.partitionPrefix + partition);
    return new TableDestination(reference, null);
}
 

/**
 * This is a change detector test for the sizes of encoded windows. Since these are present for
 * every element of every windowed PCollection, the size matters.
 *
 * <p>This test documents the expectation that encoding as a (endpoint, duration) pair using big
 * endian for the endpoint and variable length long for the duration should be about 25% smaller
 * than encoding two big endian Long values.
 */
@Test
public void testLengthsOfEncodingChoices() throws Exception {
  Instant start = Instant.parse("2015-04-01T00:00:00Z");
  Instant minuteEnd = Instant.parse("2015-04-01T00:01:00Z");
  Instant hourEnd = Instant.parse("2015-04-01T01:00:00Z");
  Instant dayEnd = Instant.parse("2015-04-02T00:00:00Z");

  Coder<Instant> instantCoder = InstantCoder.of();
  byte[] encodedStart = CoderUtils.encodeToByteArray(instantCoder, start);
  byte[] encodedMinuteEnd = CoderUtils.encodeToByteArray(instantCoder, minuteEnd);
  byte[] encodedHourEnd = CoderUtils.encodeToByteArray(instantCoder, hourEnd);
  byte[] encodedDayEnd = CoderUtils.encodeToByteArray(instantCoder, dayEnd);

  byte[] encodedMinuteWindow =
      CoderUtils.encodeToByteArray(TEST_CODER, new IntervalWindow(start, minuteEnd));
  byte[] encodedHourWindow =
      CoderUtils.encodeToByteArray(TEST_CODER, new IntervalWindow(start, hourEnd));
  byte[] encodedDayWindow =
      CoderUtils.encodeToByteArray(TEST_CODER, new IntervalWindow(start, dayEnd));

  assertThat(
      encodedMinuteWindow.length, equalTo(encodedStart.length + encodedMinuteEnd.length - 5));
  assertThat(encodedHourWindow.length, equalTo(encodedStart.length + encodedHourEnd.length - 4));
  assertThat(encodedDayWindow.length, equalTo(encodedStart.length + encodedDayEnd.length - 4));
}
 

protected DesignOptions createDesignOptions(UndesignedParameterSet parameterSet,
                                            ArrayList<TimeseriesProperties> props,
                                            boolean renderGrid) {
    Interval timespan = Interval.parse(parameterSet.getTimespan());
    long begin = timespan.getStartMillis();
    long end = timespan.getEndMillis();
    DesignOptions designOptions = new DesignOptions(props, begin, end, renderGrid);
    if (parameterSet.getResultTime() != null) {
        Instant resultTime = Instant.parse(parameterSet.getResultTime());
        designOptions.setResultTime(resultTime.getMillis());
    }
    return designOptions;
}
 

private static boolean isTimestamp(String value) {
  try {
    Instant.parse(value);
    return true;
  } catch (IllegalArgumentException e) {
    return false;
  }
}
 

@ProcessElement
public void processElement(@Element String input, OutputReceiver<KV<String, Integer>> output) {
    String[] parts = input.split(",");
    String team = parts[1].trim();
    Integer score = Integer.parseInt(parts[2].trim());
    Instant eventTime = Instant.parse(parts[3].trim());
    output.outputWithTimestamp(KV.of(team, score), eventTime);
}
 

/**
 * List messages.
 *
 * @param hl7v2Store the HL7v2 store to list messages from
 * @throws IOException the io exception
 */
@ProcessElement
public void listMessages(
    @Element String hl7v2Store,
    RestrictionTracker<OffsetRange, Long> tracker,
    OutputReceiver<HL7v2Message> outputReceiver)
    throws IOException {
  OffsetRange currentRestriction = (OffsetRange) tracker.currentRestriction();
  Instant startRestriction = Instant.ofEpochMilli(currentRestriction.getFrom());
  Instant endRestriction = Instant.ofEpochMilli(currentRestriction.getTo());
  HttpHealthcareApiClient.HL7v2MessagePages pages =
      new HttpHealthcareApiClient.HL7v2MessagePages(
          client, hl7v2Store, startRestriction, endRestriction, filter.get(), "sendTime");
  Instant cursor;
  long lastClaimedMilliSecond = startRestriction.getMillis() - 1;
  for (HL7v2Message msg : FluentIterable.concat(pages)) {
    cursor = Instant.parse(msg.getSendTime());
    if (cursor.getMillis() > lastClaimedMilliSecond) {
      // Return early after the first claim failure preventing us from iterating
      // through the remaining messages.
      if (!tracker.tryClaim(cursor.getMillis())) {
        return;
      }
      lastClaimedMilliSecond = cursor.getMillis();
    }

    outputReceiver.output(msg);
  }

  // We've paginated through all messages for this restriction but the last message may be
  // before the end of the restriction
  tracker.tryClaim(currentRestriction.getTo());
}
 

@Override
public Instant getEarliestHL7v2SendTime(String hl7v2Store, @Nullable String filter)
    throws IOException {
  ListMessagesResponse response =
      client
          .projects()
          .locations()
          .datasets()
          .hl7V2Stores()
          .messages()
          .list(hl7v2Store)
          .setFilter(filter)
          .set("view", "full") // needed to retrieve the value for sendtime
          .setOrderBy("sendTime") // default order is ascending
          // https://cloud.google.com/apis/design/design_patterns#sorting_order
          .setPageSize(1) // Only interested in the earliest sendTime
          .execute();
  if (response.isEmpty()) {
    throw new IllegalArgumentException(
        String.format(
            "Could not find earliest send time. The filter %s  matched no results on "
                + "HL7v2 Store: %s",
            filter, hl7v2Store));
  }
  String sendTime = response.getHl7V2Messages().get(0).getSendTime();
  if (Strings.isNullOrEmpty(sendTime)) {
    LOG.warn(
        String.format(
            "Earliest message in %s has null or empty sendTime defaulting to Epoch.",
            hl7v2Store));
    return Instant.ofEpochMilli(0);
  }
  // sendTime is conveniently RFC3339 UTC "Zulu"
  // https://cloud.google.com/healthcare/docs/reference/rest/v1beta1/projects.locations.datasets.hl7V2Stores.messages#Message
  return Instant.parse(sendTime);
}
 

@Override
public Instant getLatestHL7v2SendTime(String hl7v2Store, @Nullable String filter)
    throws IOException {
  ListMessagesResponse response =
      client
          .projects()
          .locations()
          .datasets()
          .hl7V2Stores()
          .messages()
          .list(hl7v2Store)
          .setFilter(filter)
          .set("view", "full") // needed to retrieve the value for sendTime
          .setOrderBy("sendTime desc")
          // https://cloud.google.com/apis/design/design_patterns#sorting_order
          .setPageSize(1) // Only interested in the earliest sendTime
          .execute();
  if (response.isEmpty()) {
    throw new IllegalArgumentException(
        String.format(
            "Could not find latest send time. The filter %s  matched no results on "
                + "HL7v2 Store: %s",
            filter, hl7v2Store));
  }
  String sendTime = response.getHl7V2Messages().get(0).getSendTime();
  if (Strings.isNullOrEmpty(sendTime)) {
    LOG.warn(
        String.format(
            "Latest message in %s has null or empty sendTime defaulting to now.", hl7v2Store));
    return Instant.now();
  }
  // sendTime is conveniently RFC3339 UTC "Zulu"
  // https://cloud.google.com/healthcare/docs/reference/rest/v1beta1/projects.locations.datasets.hl7V2Stores.messages#Message
  return Instant.parse(sendTime);
}
 

/** Tests that at least one result is emitted per element written in each window. */
@Test
public void testWritingEmitsResultsWhenDoneInFixedWindow() throws Exception {
  final String table = "table";
  final String key = "key";
  final String value = "value";

  service.createTable(table);

  Instant elementTimestamp = Instant.parse("2019-06-10T00:00:00");
  Duration windowDuration = Duration.standardMinutes(1);

  TestStream<Integer> input =
      TestStream.create(VarIntCoder.of())
          .advanceWatermarkTo(elementTimestamp)
          .addElements(1)
          .advanceWatermarkTo(elementTimestamp.plus(windowDuration))
          .addElements(2)
          .advanceWatermarkToInfinity();

  BoundedWindow expectedFirstWindow = new IntervalWindow(elementTimestamp, windowDuration);
  BoundedWindow expectedSecondWindow =
      new IntervalWindow(elementTimestamp.plus(windowDuration), windowDuration);

  PCollection<BigtableWriteResult> results =
      p.apply("rows", input)
          .apply("window", Window.into(FixedWindows.of(windowDuration)))
          .apply("expand", ParDo.of(new WriteGeneratorDoFn()))
          .apply("write", defaultWrite.withTableId(table).withWriteResults());
  PAssert.that(results)
      .inWindow(expectedFirstWindow)
      .containsInAnyOrder(BigtableWriteResult.create(1));
  PAssert.that(results)
      .inWindow(expectedSecondWindow)
      .containsInAnyOrder(BigtableWriteResult.create(2));

  p.run();
}
 

/**
 * Tests that serialising an instant with out milliseconds returns the expected ISO 8601 string
 */
@Test
public void testSerialiseWithOutMilliseconds()
{
  final Gson gson = Converters.registerInstant(new GsonBuilder()).create();
  final Instant instant = Instant.parse("2019-07-23T00:06:12Z");
  final String expectedJson = "\"2019-07-23T00:06:12.000Z\"";

  final String actualJson = gson.toJson(instant);

  assertThat(actualJson, is(expectedJson));
}
 

/**
 * Tests that deserialising an ISO 8601 string with milliseconds returns an instant with milliseconds
 */
@Test
public void testDeserialiseWithMilliseconds()
{
  final Gson gson = Converters.registerInstant(new GsonBuilder()).create();
  final String json = "\"2019-07-23T00:06:12.123Z\"";
  final String expectedStr = "2019-07-23T00:06:12.123Z";
  final Instant expected = Instant.parse("2019-07-23T00:06:12.123Z");

  final Instant reconstituted = gson.fromJson(json, Instant.class);

  assertThat(reconstituted.getMillis(), is(expected.getMillis()));
  assertThat(reconstituted.toString(), is(expectedStr));
  assertThat(reconstituted, is(expected));
}
 

/**
 * Tests that deserialising an ISO 8601 string without milliseconds returns an instant with zero milliseconds
 */
@Test
public void testDeserialiseWithoutMilliseconds()
{
  final Gson gson = Converters.registerInstant(new GsonBuilder()).create();
  final String json = "\"2019-07-23T00:06:12Z\"";
  final String expectedStr = "2019-07-23T00:06:12.000Z";
  final Instant expected = Instant.parse("2019-07-23T00:06:12.000Z");

  final Instant reconstituted = gson.fromJson(json, Instant.class);

  assertThat(reconstituted.getMillis(), is(expected.getMillis()));
  assertThat(reconstituted.toString(), is(expectedStr));
  assertThat(reconstituted, is(expected));
}
 

/**
 * Tests that deserialising an ISO 8601 string with microseconds returns an instant with milliseconds portion of string
 */
@Test
public void testDeserialiseWithMicroseconds()
{
  final Gson gson = Converters.registerInstant(new GsonBuilder()).create();
  final String json = "\"2019-07-23T00:06:12.123456Z\"";
  final String expectedStr = "2019-07-23T00:06:12.123Z";
  final Instant expected = Instant.parse("2019-07-23T00:06:12.123Z");

  final Instant reconstituted = gson.fromJson(json, Instant.class);

  assertThat(reconstituted.getMillis(), is(expected.getMillis()));
  assertThat(reconstituted.toString(), is(expectedStr));
  assertThat(reconstituted, is(expected));
}
 

/**
 * Tests that deserialising an ISO 8601 string with a timezone returns an instant (converted to utc)
 */
@Test
public void testDeserialiseWithTimezoneOffset()
{
  final Gson gson = Converters.registerInstant(new GsonBuilder()).create();
  final String json = "\"2019-07-23T01:06:12.123+01:00\"";
  final String expectedStr = "2019-07-23T00:06:12.123Z";
  final Instant expected = Instant.parse("2019-07-23T01:06:12.123+01:00");

  final Instant reconstituted = gson.fromJson(json, Instant.class);

  assertThat(reconstituted.toString(), is(expectedStr));
  assertThat(reconstituted, is(expected));
}
 

/**
 * Tests that serialising an instant with milliseconds returns the expected ISO 8601 string
 */
@Test
public void testSerialiseWithMilliseconds()
{
  final Gson gson = Converters.registerInstant(new GsonBuilder()).create();
  final Instant instant = Instant.parse("2019-07-23T00:06:12.123Z");
  final String expectedJson = "\"2019-07-23T00:06:12.123Z\"";

  final String actualJson = gson.toJson(instant);

  assertThat(actualJson, is(expectedJson));
}
 

public static Instant parseTime(String time) {
    return Instant.parse("T" + time + "Z");
}
 

private IntervalWindow createIntervalWindow(String startStr, String endStr) {
  return new IntervalWindow(Instant.parse(startStr), Instant.parse(endStr));
}
 

/**
 * Tests that the outputs of the Bigtable writer are correctly windowed, and can be used in a
 * Wait.on transform as the trigger.
 */
@Test
public void testWritingAndWaitingOnResults() throws Exception {
  final String table = "table";
  final String key = "key";
  final String value = "value";

  service.createTable(table);

  Instant elementTimestamp = Instant.parse("2019-06-10T00:00:00");
  Duration windowDuration = Duration.standardMinutes(1);

  TestStream<KV<ByteString, Iterable<Mutation>>> writeInputs =
      TestStream.create(bigtableCoder)
          .advanceWatermarkTo(elementTimestamp)
          .addElements(makeWrite(key, value))
          .advanceWatermarkToInfinity();

  TestStream<String> testInputs =
      TestStream.create(StringUtf8Coder.of())
          .advanceWatermarkTo(elementTimestamp)
          .addElements("done")
          .advanceWatermarkToInfinity();

  PCollection<BigtableWriteResult> writes =
      p.apply("rows", writeInputs)
          .apply(
              "window rows",
              Window.<KV<ByteString, Iterable<Mutation>>>into(FixedWindows.of(windowDuration))
                  .withAllowedLateness(Duration.ZERO))
          .apply("write", defaultWrite.withTableId(table).withWriteResults());

  PCollection<String> inputs =
      p.apply("inputs", testInputs)
          .apply("window inputs", Window.into(FixedWindows.of(windowDuration)))
          .apply("wait", Wait.on(writes));

  BoundedWindow expectedWindow = new IntervalWindow(elementTimestamp, windowDuration);

  PAssert.that(inputs).inWindow(expectedWindow).containsInAnyOrder("done");

  p.run();
}
 

@Test(groups = {"unit", "server"})
public void testTimestamp() {
    String insertCql = "insert into test_drive.basic_data_type(id_uuid, date_timestamp) values(?, ?)";
    String queryCql = "select date_timestamp from test_drive.basic_data_type where id_uuid = ?";
    UUID id = UUID.randomUUID();
    String timestamp = "2019-02-01T04:12:21.330Z";
    String timestamp1 = "2019-02-01 04:12:21.330";
    Instant instant = Instant.parse(timestamp);
    long ts = instant.getMillis();
    Timestamp t = new Timestamp(ts);

    try {
        // set time by string
        java.sql.PreparedStatement s = conn.prepareStatement(insertCql);
        s.setObject(1, id);
        s.setString(2, timestamp);
        s.execute();
        s.close();

        s = conn.prepareStatement(queryCql);
        s.setObject(1, id);
        ResultSet rs = s.executeQuery();
        rs.next();
        assertEquals(rs.getString(1), timestamp);
        assertEquals(rs.getObject(1), instant.toDate());
        assertEquals(rs.getTimestamp(1), t);
        rs.close();
        s.close();

        // set time by string in the other format
        s = conn.prepareStatement(insertCql);
        s.setObject(1, id);
        s.setString(2, timestamp1);
        s.execute();
        s.close();

        s = conn.prepareStatement(queryCql);
        s.setObject(1, id);
        rs = s.executeQuery();
        rs.next();
        assertEquals(rs.getString(1), timestamp);
        assertEquals(rs.getObject(1), instant.toDate());
        assertEquals(rs.getTimestamp(1), t);
        rs.close();
        s.close();

        // by long
        s = conn.prepareStatement(insertCql);
        s.setObject(1, id);
        s.setObject(2, ts);
        s.execute();
        s.close();

        s = conn.prepareStatement(queryCql);
        s.setObject(1, id);
        rs = s.executeQuery();
        rs.next();
        assertEquals(rs.getString(1), timestamp);
        assertEquals(rs.getObject(1), instant.toDate());
        assertEquals(rs.getTimestamp(1), t);
        rs.close();
        s.close();

        // by timestamp
        s = conn.prepareStatement(insertCql);
        s.setObject(1, id);
        s.setObject(2, t);
        s.execute();
        s.close();

        s = conn.prepareStatement(queryCql);
        s.setObject(1, id);
        rs = s.executeQuery();
        rs.next();
        assertEquals(rs.getString(1), timestamp);
        assertEquals(rs.getObject(1), instant.toDate());
        assertEquals(rs.getTimestamp(1), t);
        rs.close();
        s.close();

        // by Joda Instant
        s = conn.prepareStatement(insertCql);
        s.setObject(1, id);
        s.setObject(2, instant);
        s.execute();
        s.close();

        s = conn.prepareStatement(queryCql);
        s.setObject(1, id);
        rs = s.executeQuery();
        rs.next();
        assertEquals(rs.getString(1), timestamp);
        assertEquals(rs.getObject(1), instant.toDate());
        assertEquals(rs.getTimestamp(1), t);
        rs.close();
        s.close();
    } catch (Exception e) {
        e.printStackTrace();
        fail("Error occurred during testing: " + e.getMessage());
    }
}
 

/**
 * Gson invokes this call-back method during deserialization when it encounters a field of the
 * specified type. <p>
 *
 * In the implementation of this call-back method, you should consider invoking
 * {@link JsonDeserializationContext#deserialize(JsonElement, Type)} method to create objects
 * for any non-trivial field of the returned object. However, you should never invoke it on the
 * the same type passing {@code json} since that will cause an infinite loop (Gson will call your
 * call-back method again).
 * @param json The Json data being deserialized
 * @param typeOfT The type of the Object to deserialize to
 * @return a deserialized object of the specified type typeOfT which is a subclass of {@code T}
 * @throws JsonParseException if json is not in the expected format of {@code typeOfT}
 */
@Override
public Instant deserialize(JsonElement json, Type typeOfT, JsonDeserializationContext context)
    throws JsonParseException
{
  if (json.getAsString() == null || json.getAsString().isEmpty())
  {
    return null;
  }

  return Instant.parse(json.getAsString(), formatter);
}