vineetsaurabh commented on code in PR #18212:
URL: https://github.com/apache/camel/pull/18212#discussion_r2123170589
##########
components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsProducer.java:
##########
@@ -193,4 +195,65 @@ protected void doStop() throws Exception {
super.doStop();
}
+ private void publishWithJetStream(NatsConfiguration config, final byte[]
body, final Exchange exchange) {
+ LOG.debug("JetStream is available");
+ JetStreamManagement jsm;
+ JetStream js;
+ try {
+ jsm = this.connection.jetStreamManagement();
+ js = jsm.jetStream();
+ if (js == null) {
+ jsm.addStream(StreamConfiguration.builder()
+ .name(config.getJetstreamName())
+ .subjects(config.getTopic())
+ .build());
+ js = jsm.jetStream();
+ }
+ } catch (IOException | JetStreamApiException e) {
+ throw new RuntimeException("Failed to initialize JetStream: " +
e.getMessage(), e);
+ }
+
+ final NatsMessage.Builder builder = NatsMessage.builder()
+ .data(body)
+ .subject(config.getTopic())
+ .headers(this.buildHeaders(exchange));
+
+ if (ObjectHelper.isNotEmpty(config.getReplySubject())) {
+ final String replySubject = config.getReplySubject();
+ builder.replyTo(replySubject);
+ }
+ final Message jetStreamMessage = builder.build();
+
+ PublishAck pa;
+ if (config.isJetstreamAsync()) {
+ CompletableFuture<PublishAck> future =
js.publishAsync(jetStreamMessage);
+ try {
+ pa = future.get(1, TimeUnit.SECONDS);
Review Comment:
Sorry, it is not clear to me - are you suggesting a new timeout value? Or,
shall I proceed with the code mentioned above?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@camel.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
