Core { public static Func1<JsonEvent, Observable<JsonEvent>> date(String field, String... formats) { final List<DateTimeFormatter> formatters = asList(formats).stream() .map(format -> DateTimeFormatter.ofPattern(format).withZone(ZoneId.of("UTC"))) .collect(toList()); return jsonEvent -> { String dateString = jsonEvent.valueAsString(field); for (DateTimeFormatter formatter: formatters) { try { ZonedDateTime date = ZonedDateTime.parse(dateString, formatter); String isoDate = DateTimeFormatter.ISO_OFFSET_DATE_TIME.format(date); jsonEvent.putMetaData("_@date", date); jsonEvent.remove(field); return jsonEvent.put("@timestamp", isoDate).toObservable(); } catch (RuntimeException e) { if (LOG.isTraceEnabled()) { LOG.trace("Failed to parse date {}", e.getMessage()); } continue; } } throw new IllegalStateException("Failed to parse date with any of the supplied formats"); }; }  static Func1<AnyJsonEvent, Observable<JsonEvent>> splitIfArray(); static Func1<List<T>,Observable<T>> flatten(); static Func1<List<T>,Observable<List<T>>> parallelize(Func1<T,T> func); static Func1<List<T>,Observable<List<T>>> sequence(Func1<T, T> func); static Func1<JsonEvent, Observable<JsonEvent>> params(Map conf); static Func1<E, Observable<E>> computeIfTagExists(String tag, Func1<E,?> cmd); static Func1<JsonEvent, Observable<JsonEvent>> computeIfTagAbsent(String tag, Func1<JsonEvent,?> cmd); static Func1<JsonEvent, Observable<JsonEvent>> computeIfNotMatch(String field, String value, Func1<JsonEvent,?> cmd); static Func1<JsonEvent, Observable<JsonEvent>> computeIfMatch(String field, String value, Func1<JsonEvent,?> cmd); static Func1<E, Observable<E>> computeIfExists(String field, Func1<E,?> cmd); static Func1<E, Observable<E>> computeIfAbsent(String field, Func1<E,?> cmd); @Deprecated static ConditionalFunc1 ifExists(String field); @Deprecated static ConditionalFunc1 ifNotExists(String field); @Deprecated static ConditionalFunc1 ifMatch(String field, String value); @Deprecated static ConditionalFunc1 ifNotMatch(String field, String value); static Func1<JsonEvent, Observable<JsonEvent>> remove(String... field); static Func1<JsonEvent, Observable<JsonEvent>> rename(Map map); static Func1<JsonEvent, Observable<JsonEvent>> copy(Map map); static Func1<JsonEvent, Observable<JsonEvent>> timestampNow(); static Func1<JsonEvent, Observable<JsonEvent>> timestampFromMs(String from); static Func1<JsonEvent,Observable<JsonEvent>> timestampFromSecs(String from); static Func1<JsonEvent, Observable<JsonEvent>> timestampFromMs(); static Func1<JsonEvent, Observable<JsonEvent>> timestampFromSecs(); static Func1<JsonEvent, Observable<JsonEvent>> timestampFromSecs(String from, String to); static Func1<JsonEvent, Observable<JsonEvent>> timestampFromMs(String from, String to); static Func1<JsonEvent, Observable<JsonEvent>> date(String field, String... formats); static RetryStrategy exceptionOfType(List<Class<? extends Throwable>> retryOn); static RetryStrategy exceptionOfType(Class<? extends Throwable> retryOn); static RetryStrategy exceptionOfAnyType(); static Func1<E, Boolean> hasField(String field); static Func1<JsonEvent, Observable<JsonEvent>> addField(String field, String value); static Func1<JsonEvent, Observable<JsonEvent>> addField(String field, int value); static Func1<JsonEvent, Observable<JsonEvent>> addField(String field, boolean value); static Func1<JsonEvent, Observable<JsonEvent>> addField(String field, float value); static Func1<List<JsonEvent>, Observable<AnyJsonEvent>> toJsonArray(); static Func1<E, Observable<AnyJsonEvent>> json(); static Func1<E, Observable<JsonEvent>> toJsonObject(Map map); static Func1<E, Observable<JsonEvent>> toJsonObject(); static Func1<JsonEvent, Observable<JsonEvent>> extractJsonObject(Map map); static Func1<JsonEvent, Observable<JsonEvent>> extractJsonObject(String field); static Func1<E, Observable<JsonEvent>> textToJson(); static Func1<JsonEvent, Observable<AnyJsonEvent>> jsonOfField(String field); static Func1<Event, Boolean> keepWhen(String strExpression); static Func1<Event, Boolean> skipWhen(String strExpression); @Deprecated static Subscriber<T> wrap(Subscriber<T>... s); @Deprecated static Func1<T,T> wrap(Func1<T, T> target); static Console console; static Grok grok; static GZip gzip; static Zlib zlib; static Base64 base64; static File file; static Fingerprint fingerprint; }
@Test public void testDate() { Codecs.JSON_OBJECT.from("{\"time\":\"2016-01-01T22\"}") .<JsonEvent>toObservable() .flatMap ( date("time", "yyyy-MM-dd'T'HH")) .doOnNext(jsonEvent -> assertThat(jsonEvent.valueAsString("@timestamp")).isEqualTo("2016-01-01T22:00:00Z")) .subscribe(); }