как сохранить значения в некотором постоянном объекте при использовании конечных точек файла - PullRequest
Новая
       14

как сохранить значения в некотором постоянном объекте при использовании конечных точек файла

0 голосов
/ 24 июня 2019

Я не уверен, как сохранить значения в некотором постоянном объекте для извлечения / использования в процессоре метрик, поскольку свойства и заголовки недоступны в конечных точках файла.

Используя Camel Core 2.22.1, Spring Boot starter 2.22.1 под Spring Boot 2.0.3, я построил процессор, который может анализировать метрики из файлов, но не может получить доступ к таймеру (или любым значениям) из свойств или заголовки. Я могу запустить процессор как встроенный или как перехватчик, и он будет выполняться, но выдает ошибки в значениях, которые он пытается получить из заголовков или реквизитов (теперь я знаю, почему)

RouteBuilder.java 

    @Override
    public void configure() throws Exception {
        from("file:func/input")
        .unmarshal().json(JsonLibrary.Jackson, Options.class)
        .process(optionsToHeaders)
        .process(functionEngineProcessor)
        .to("file:func/output?tempPrefix=.inprogress/");

FileReceiverRouteBuilder.java 

    @Override
    public void configure() throws Exception {
        rest("/func")
            .post("/process")
            .consumes(MediaType.TEXT_PLAIN_VALUE)
            .route().routeId("func-upload-route")
            .setHeader(Exchange.FILE_NAME, () -> UUID.randomUUID().toString())  // use a UUID for inprocess filename
            .to("file:uploads")
            .setBody((ex) -> buildOptions(ex))   // appends  json as function.options
            .marshal().json(JsonLibrary.Jackson, true)
            .log(LoggingLevel.INFO, "function.options: ${in.body}")
            .to("file:func/input")
            .setBody((ex) -> new ProcessResponse((String) ex.getIn().getHeader(Exchange.FILE_NAME)))
            .marshal().json(JsonLibrary.Jackson, true);

        rest("/func")
            .get("status/{token}")
            .route().routeId("get-generated-func-file")
            .choice()
                .when((ex) -> Paths.get("function/output", (String) ex.getIn().getHeader("token")).toFile().exists())
                    .to("direct:func-return-file-content")
                .otherwise()
                    .setHeader(Exchange.HTTP_RESPONSE_CODE, constant("202"));    /// send 202 until file complete

        from("direct:func-return-file-content")
            .log("returning file content")
            .setBody((ex) -> Paths.get("func/output", (String) ex.getIn().getHeader("token")).toFile())
            .setHeader(Exchange.CONTENT_TYPE, constant(MediaType.TEXT_PLAIN_VALUE));

OptionsToHeadersProcessor помещает имя файла и параметры конфигурации в json. 

public class OptionsToHeadersProcessor implements Processor{

    public static final String PROCESS_PROFILE = "PROCESS_PROFILE";
    public static final String FILE_NAME = "FILE_NAME";

    @Override
    public void process(Exchange exchange) throws Exception {
        Options options = exchange.getIn().getBody(Options.class);
        exchange.getIn().setHeader(FILE_NAME, options.getInputFile());
        exchange.getIn().setHeader(PROCESS_PROFILE, options.getConfiguration());  //as header

    }

Обработка работает нормально, файлы возвращаются и т. Д. У нас есть log4j, выполняющий добавление журнала (скользящий файл) для службы, также ОК.

Однако теперь мы хотим улучшить, создав статистический набор данных о каждом обработанном файле. Статистика должна включать в себя: - время начала процесса, время окончания и продолжительность, время одного или нескольких методов - также начало, конец, продолжительность (изнутри основной функции EngineProcessor) данные о составе входного файла (например, байты / строки) в целом и для определенных разделов в файле.

Нам также нужно скомпилировать всю эту статистику в один (json) текстовый пакет, чтобы он представлял собой одну строку json, прежде чем мы отправим их в файл.

Последнее требование вызывает наибольшую проблему; Я думаю, вы можете видеть, что если мне нужно соединить, скажем, начало, время окончания и продолжительность, чтобы записать их (даже используя log4j), Мне нужно сохранить значение времени начала для всех процессоров, чтобы мой процессор статистики мог рассчитать конечное время и продолжительность окончания.

Я пытался использовать перехват перед самым определением первой конечной точки в FileReceiverRouteBuilder.java:

        interceptSendToEndpoint("direct:func*")
        .log(LoggingLevel.INFO, "intercepted sendto output for metrics")
        .process(metricsProcessor);

Однако процессор метрик не может извлечь что-либо из заголовков (даже PROCESS_PROFILE или FILE_NAME), поскольку, очевидно, конечные точки файла не сохраняют свойства или заголовки.
«Маркер» может быть извлечен, но он привязан к подписи REST клиента, и я не могу расширить его без особых проблем.

Поэтому я не уверен, как сохранить значения в каком-либо постоянном объекте для извлечения / использования в процессоре метрик, поскольку свойства и заголовки недоступны. Мое текущее чтение, кажется, подразумевает, что JMS - способ достигнуть этого? Я новичок в верблюде и т. Д., И я хотел посмотреть, упускаю ли я что-то очевидное.

Любая помощь высоко ценится!

...