Marmelator/layout-parser
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 289 Wiki Activity

Merge branch 'clari-30' into 'main'

Browse Source 
CLARI-30 - forward analysis headers

See merge request fforesight/layout-parser!112
This commit is contained in:
Timo Bejan 2024-03-08 15:51:53 +01:00
commit bef23e38b5
1 changed files with 17 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
layoutparser-service/layoutparser-service-server/src/main/java/com/knecon/fforesight/service/layoutparser/server/queue/MessageHandler.java
View File

@ -1,6 +1,9 @@
package com.knecon.fforesight.service.layoutparser.server.queue; package com.knecon.fforesight.service.layoutparser.server.queue;
import java.util.Arrays; import java.util.Arrays;
import java.util.Locale;
import java.util.Map;
import java.util.stream.Collectors;
import org.springframework.amqp.AmqpRejectAndDontRequeueException; import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.core.Message; import org.springframework.amqp.core.Message;
@ -28,6 +31,7 @@ public class MessageHandler {
private final LayoutParsingPipeline layoutParsingPipeline; private final LayoutParsingPipeline layoutParsingPipeline;
private final ObjectMapper objectMapper; private final ObjectMapper objectMapper;
private final RabbitTemplate rabbitTemplate; private final RabbitTemplate rabbitTemplate;
private final static String X_PIPELINE_PREFIX = "X-PIPE-";
@RabbitHandler @RabbitHandler
@ -46,14 +50,24 @@ public class MessageHandler {
layoutParsingRequest.identifier())); layoutParsingRequest.identifier()));
} }
LayoutParsingFinishedEvent layoutParsingFinishedEvent = layoutParsingPipeline.parseLayoutAndSaveFilesToStorage(layoutParsingRequest); LayoutParsingFinishedEvent layoutParsingFinishedEvent = layoutParsingPipeline.parseLayoutAndSaveFilesToStorage(layoutParsingRequest);
sendLayoutParsingFinishedEvent(layoutParsingFinishedEvent); sendLayoutParsingFinishedEvent(layoutParsingFinishedEvent, message);
} }
public void sendLayoutParsingFinishedEvent(LayoutParsingFinishedEvent layoutParsingFinishedEvent) { public void sendLayoutParsingFinishedEvent(LayoutParsingFinishedEvent layoutParsingFinishedEvent, Message message) {
Arrays.stream(layoutParsingFinishedEvent.message().split("\n")).forEach(log::info); Arrays.stream(layoutParsingFinishedEvent.message().split("\n")).forEach(log::info);
rabbitTemplate.convertAndSend(LayoutParsingQueueNames.LAYOUT_PARSING_FINISHED_EVENT_QUEUE, layoutParsingFinishedEvent); rabbitTemplate.convertAndSend(LayoutParsingQueueNames.LAYOUT_PARSING_FINISHED_EVENT_QUEUE, layoutParsingFinishedEvent, m -> {
var forwardHeaders = message.getMessageProperties()
.getHeaders()
.entrySet()
.stream()
.filter(e -> e.getKey().toUpperCase(Locale.ROOT).startsWith(X_PIPELINE_PREFIX))
.collect(Collectors.toMap(Map.Entry::getKey,
Map.Entry::getValue));
m.getMessageProperties().getHeaders().putAll(forwardHeaders);
return m;
});
} }
} }