/**
 * <p>parseDate.</p>
 *
 * @param value a {@link java.lang.String} object.
 * @param pos   a {@link java.text.ParsePosition} object.
 * @return a {@link java.util.Date} object.
 */
public static Date parseDate(String value, ParsePosition pos) {
    Long timestamp = parseTimestamp(value);
    if (timestamp != null) {
        return new Date(timestamp);
    }
    if (value.contains(" ")) {
        value = value.replace(" ", "+");
    }
    if (!(value.contains("-") || value.contains("+")) && !value.endsWith("Z")) {
        value += SYS_TZ;
    }
    try {
        return ISO8601Utils.parse(value, pos);
    } catch (ParseException e) {
        throw new ExtractorException(e);
    }
}
 

/**
 * <p>
 * Parses the ISO-8601 date from the image JSON.
 * </p>
 * <p>
 * Handles the bug in the NASA JSON where not all timestamps comply with ISO-8601 (some are missing the 'Z' for UTC
 * time at the end of the timestamp).
 * </p>
 * Uses Jackson ISO8601Utils to convert the timestamp.
 *
 * @param image
 *            JSON representation of the image
 * @return Java Date object containing the creation time stamp
 */
protected static Date getTimestamp(final JsonNode image) {
    Date date = null;
    String iso8601 = image.get(IMAGE_TIME_KEY).get(CREATION_TIME_STAMP_KEY).asText();
    try {
        date = ISO8601Utils.parse(iso8601);
    } catch (final IllegalArgumentException e) {
        // Don't like this, but not all times have the Z at the end for
        // ISO-8601
        if (iso8601.charAt(iso8601.length() - 1) != 'Z') {
            iso8601 = iso8601 + "Z";
            date = ISO8601Utils.parse(iso8601);
        } else {
            throw e;
        }
    }
    return date;
}
 

private void handleDueBefore(TaskInfoQueryWrapper taskInfoQueryWrapper, JsonNode dueBeforeNode) {
  String date = dueBeforeNode.asText();
  Date d = null;
  try {
    d = ISO8601Utils.parse(date, new ParsePosition(0));
  } catch (ParseException e) {
    e.printStackTrace();
  }
  taskInfoQueryWrapper.getTaskInfoQuery().taskDueBefore(d);
}
 

private void handleDueAfter(TaskInfoQueryWrapper taskInfoQueryWrapper, JsonNode dueAfterNode) {
  String date = dueAfterNode.asText();
  Date d = null;
  try {
    d = ISO8601Utils.parse(date, new ParsePosition(0));
  } catch (ParseException e) {
    e.printStackTrace();
  }
  taskInfoQueryWrapper.getTaskInfoQuery().taskDueAfter(d);
}
 

/** Parses an ISO 8601 date+time string into a Java Date object. */
public static Date parseTimestamp(@Nullable String string) {
    Date date = null;
    try {
        if (string != null) {
            date = ISO8601Utils.parse(string, new ParsePosition(0));
        }
    } catch (ParseException e) {
        throw Throwables.propagate(e);
    }
    return date;
}
 

@Test
public void testTopologyResolvingOneRef() {
    final String testTableName = "tableA";
    final String testId = "id1";
    final Map<String, Object> testContents = new HashMap<String, Object>() {{
        put("~table", testTableName);
        put("~id", testId);
        put("~version", 0);
        put("~signature", "abc123");
        put("~deleted", false);
        put("~firstUpdateAt", ISO8601Utils.format(new Date()));
        put("~lastUpdateAt", ISO8601Utils.format(new Date()));
        put("~lastMutateAt", ISO8601Utils.format(new Date()));
    }};

    MegabusRef megabusRef = new MegabusRef(testTableName, testId, TimeUUIDs.newUUID(), Instant.now(), MegabusRef.RefType.NORMAL);

    ((TestDataProvider) dataProvider).addTable(testTableName, new InMemoryTable(testTableName, new TableOptionsBuilder().setPlacement("app_global").build(), new HashMap<>()));
    ((TestDataProvider) dataProvider).add(testContents);

    List<MegabusRef> megabusRefs = Collections.singletonList(megabusRef);

    // push the megabusref to the input topic
    testDriver.pipeInput(recordFactory.create(refTopic.getName(), "eventId1", megabusRefs));

    // read the result
    final ProducerRecord<String, Map<String, Object>> output = testDriver.readOutput(resolvedTopic.getName(), stringDeserializer, jsonPOJOSerdeForMaps.deserializer());

    // ensure ref was resolved successfully and had the correct value and was written to the correct topic
    OutputVerifier.compareKeyValue(output,  String.format("%s/%s", testTableName, testId), testContents);

    // ensure no more records left unread
    assertNull(testDriver.readOutput(resolvedTopic.getName(), stringDeserializer, jsonPOJOSerdeForMaps.deserializer()));
}
 

@Test
public void testTopologyMissingOneRef() {
    final String testTableName = "tableA";
    final String testId = "id1";
    final Map<String, Object> testContents = new HashMap<String, Object>() {{
        put("~table", testTableName);
        put("~id", testId);
        put("~version", 0);
        put("~signature", "abc123");
        put("~deleted", false);
        put("~firstUpdateAt", ISO8601Utils.format(new Date()));
        put("~lastUpdateAt", ISO8601Utils.format(new Date()));
        put("~lastMutateAt", ISO8601Utils.format(new Date()));
    }};

    MegabusRef megabusRef = new MegabusRef(testTableName, testId, TimeUUIDs.newUUID(), Instant.now(), MegabusRef.RefType.NORMAL);

    ((TestDataProvider) dataProvider).addTable(testTableName, new InMemoryTable(testTableName, new TableOptionsBuilder().setPlacement("app_global").build(), new HashMap<>()));
    ((TestDataProvider) dataProvider).addPending(testContents);

    List<MegabusRef> megabusRefs = Collections.singletonList(megabusRef);

    // push the megabusref to the input topic
    testDriver.pipeInput(recordFactory.create(refTopic.getName(), "eventId1", megabusRefs));

    // ensure the resolved topic does _NOT_ have the ref (it should not have resolved)
    assertNull(testDriver.readOutput(resolvedTopic.getName(), stringDeserializer, jsonPOJOSerdeForMaps.deserializer()));

    // read the result  from the "missing" topic
    final ProducerRecord<String, MissingRefCollection> output = testDriver.readOutput(missingRefsTopic.getName(), stringDeserializer, jsonPOJOSerdeForMissingRefCollection.deserializer());

    // ensure ref was not resolved successfully and had the correct key, and had exactly the missing reference and was written to the correct topic
    assertEquals(output.key(), "eventId1");
    assertEquals(output.value().getMissingRefs(), Collections.singletonList(megabusRef));

    // ensure no more records left unread in the missing refs topic
    assertNull(testDriver.readOutput(missingRefsTopic.getName(), stringDeserializer, jsonPOJOSerdeForMaps.deserializer()));
}
 

@Test
public void testTopologyMergesRetries() {
    final String testTableName = "tableA";
    final String testId = "id1";
    final Map<String, Object> testContents = new HashMap<String, Object>() {{
        put("~table", testTableName);
        put("~id", testId);
        put("~version", 0);
        put("~signature", "abc123");
        put("~deleted", false);
        put("~firstUpdateAt", ISO8601Utils.format(new Date()));
        put("~lastUpdateAt", ISO8601Utils.format(new Date()));
        put("~lastMutateAt", ISO8601Utils.format(new Date()));
    }};

    MegabusRef megabusRef = new MegabusRef(testTableName, testId, TimeUUIDs.newUUID(), Instant.now(), MegabusRef.RefType.NORMAL);

    ((TestDataProvider) dataProvider).addTable(testTableName, new InMemoryTable(testTableName, new TableOptionsBuilder().setPlacement("app_global").build(), new HashMap<>()));
    ((TestDataProvider) dataProvider).add(testContents);

    List<MegabusRef> megabusRefs = Collections.singletonList(megabusRef);

    // push the megabusref to the input topic
    testDriver.pipeInput(recordFactory.create(retryTopic.getName(), "eventId1", megabusRefs));

    // read the result
    final ProducerRecord<String, Map<String, Object>> output = testDriver.readOutput(resolvedTopic.getName(), stringDeserializer, jsonPOJOSerdeForMaps.deserializer());

    // ensure ref was resolved successfully and had the correct value and was written to the correct topic
    OutputVerifier.compareKeyValue(output,  String.format("%s/%s", testTableName, testId), testContents);

    // ensure no more records left unread
    assertNull(testDriver.readOutput(resolvedTopic.getName(), stringDeserializer, jsonPOJOSerdeForMaps.deserializer()));
}
 

public static Date parseDate(String s) {
    if (s != null) {
        try {
            return ISO8601Utils.parse(s, new ParsePosition(0));
        } catch (ParseException e) {
            return null;
        }
    }
    return null;
}
 

@Override
public StringBuffer format(Date date, StringBuffer toAppendTo, FieldPosition fieldPosition)
{
	final String value = ISO8601Utils.format(date, true);
	toAppendTo.append(value);
	return toAppendTo;
}
 

/**
 * @throws ParseException 
 */
private RequestMetric request(String requestStart, long requestDuration, String url, String resource,
        String method, String apiOrgId, String apiId, String apiVersion, String planId,
        String clientOrgId, String clientId, String clientVersion, String contractId, String user,
        int responseCode, String responseMessage, boolean failure, int failureCode, String failureReason,
        boolean error, String errorMessage, long bytesUploaded, long bytesDownloaded) throws ParseException {
    Date start = ISO8601Utils.parse(requestStart, new ParsePosition(0));
    RequestMetric rval = new RequestMetric();
    rval.setRequestStart(start);
    rval.setRequestEnd(new Date(start.getTime() + requestDuration));
    rval.setApiStart(start);
    rval.setApiEnd(rval.getRequestEnd());
    rval.setApiDuration(requestDuration);
    rval.setUrl(url);
    rval.setResource(resource);
    rval.setMethod(method);
    rval.setApiOrgId(apiOrgId);
    rval.setApiId(apiId);
    rval.setApiVersion(apiVersion);
    rval.setPlanId(planId);
    rval.setClientOrgId(clientOrgId);
    rval.setClientId(clientId);
    rval.setClientVersion(clientVersion);
    rval.setContractId(contractId);
    rval.setUser(user);
    rval.setResponseCode(responseCode);
    rval.setResponseMessage(responseMessage);
    rval.setFailure(failure);
    rval.setFailureCode(failureCode);
    rval.setFailureReason(failureReason);
    rval.setError(error);
    rval.setErrorMessage(errorMessage);
    rval.setBytesUploaded(bytesUploaded);
    rval.setBytesDownloaded(bytesDownloaded);
    return rval;
}
 

private JsonNode toJsonNode(String rawJson) {
    ObjectNode parse = JSONUtils.getInstance().parseObj(rawJson);
    parse.put("timestamp", ISO8601Utils.format(new Date()));
    return parse;
}
 

@Override
public StringBuffer format(Date date, StringBuffer toAppendTo, FieldPosition fieldPosition) {
  String value = ISO8601Utils.format(date, true);
  toAppendTo.append(value);
  return toAppendTo;
}
 

@Override
public StringBuffer format(Date date, StringBuffer toAppendTo, FieldPosition fieldPosition) {
  String value = ISO8601Utils.format(date, true);
  toAppendTo.append(value);
  return toAppendTo;
}
 

@Override
public StringBuffer format(Date date, StringBuffer toAppendTo, FieldPosition fieldPosition) {
  String value = ISO8601Utils.format(date, true);
  toAppendTo.append(value);
  return toAppendTo;
}
 

@Override
public StringBuffer format(Date date, StringBuffer toAppendTo, FieldPosition fieldPosition) {
  String value = ISO8601Utils.format(date, true);
  toAppendTo.append(value);
  return toAppendTo;
}
 

@Override
public StringBuffer format(Date date, StringBuffer toAppendTo, FieldPosition fieldPosition) {
  String value = ISO8601Utils.format(date, true);
  toAppendTo.append(value);
  return toAppendTo;
}
 

@Override
public StringBuffer format(Date date, StringBuffer toAppendTo, FieldPosition fieldPosition) {
  String value = ISO8601Utils.format(date, true);
  toAppendTo.append(value);
  return toAppendTo;
}
 

@Override
public StringBuffer format(Date date, StringBuffer toAppendTo, FieldPosition fieldPosition) {
  String value = ISO8601Utils.format(date, true);
  toAppendTo.append(value);
  return toAppendTo;
}
 

@Override
public StringBuffer format(Date date, StringBuffer toAppendTo, FieldPosition fieldPosition) {
  String value = ISO8601Utils.format(date, true);
  toAppendTo.append(value);
  return toAppendTo;
}
 

@Override
public StringBuffer format(Date date, StringBuffer toAppendTo, FieldPosition fieldPosition) {
  String value = ISO8601Utils.format(date, true);
  toAppendTo.append(value);
  return toAppendTo;
}
 

@Override
public StringBuffer format(Date date, StringBuffer toAppendTo, FieldPosition fieldPosition) {
  String value = ISO8601Utils.format(date, true);
  toAppendTo.append(value);
  return toAppendTo;
}
 

@Override
public StringBuffer format(Date date, StringBuffer toAppendTo, FieldPosition fieldPosition) {
  String value = ISO8601Utils.format(date, true);
  toAppendTo.append(value);
  return toAppendTo;
}
 

@Override
public StringBuffer format(Date date, StringBuffer toAppendTo, FieldPosition fieldPosition) {
  String value = ISO8601Utils.format(date, true);
  toAppendTo.append(value);
  return toAppendTo;
}
 

@Override
public StringBuffer format(Date date, StringBuffer toAppendTo, FieldPosition fieldPosition) {
  String value = ISO8601Utils.format(date, true);
  toAppendTo.append(value);
  return toAppendTo;
}
 

@Override
public StringBuffer format(Date date, StringBuffer toAppendTo, FieldPosition fieldPosition) {
    String value = ISO8601Utils.format(date, true);
    toAppendTo.append(value);
    return toAppendTo;
}
 

@Override
public StringBuffer format(Date date, StringBuffer toAppendTo, FieldPosition fieldPosition) {
  String value = ISO8601Utils.format(date, true);
  toAppendTo.append(value);
  return toAppendTo;
}
 

@Override
public StringBuffer format(Date date, StringBuffer toAppendTo, FieldPosition fieldPosition) {
  String value = ISO8601Utils.format(date, true);
  toAppendTo.append(value);
  return toAppendTo;
}
 

@Override
public StringBuffer format(Date date, StringBuffer toAppendTo, FieldPosition fieldPosition) {
    String value = ISO8601Utils.format(date, true);
    toAppendTo.append(value);
    return toAppendTo;
}
 

@Override
public StringBuffer format(Date date, StringBuffer toAppendTo, FieldPosition fieldPosition) {
    String value = ISO8601Utils.format(date, true);
    toAppendTo.append(value);
    return toAppendTo;
}