-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #6 from SolaceCoEExt/failure-strategies
Failure strategies implementation
- Loading branch information
Showing
16 changed files
with
336 additions
and
141 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
41 changes: 41 additions & 0 deletions
41
pubsub-plus-connector/src/main/java/io/quarkiverse/solace/fault/SolaceDiscard.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| package io.quarkiverse.solace.fault; | ||
|
|
||
| import java.util.concurrent.CompletionStage; | ||
|
|
||
| import org.eclipse.microprofile.reactive.messaging.Metadata; | ||
|
|
||
| import com.solace.messaging.config.MessageAcknowledgementConfiguration; | ||
| import com.solace.messaging.receiver.AcknowledgementSupport; | ||
|
|
||
| import io.quarkiverse.solace.i18n.SolaceLogging; | ||
| import io.quarkiverse.solace.incoming.SettleMetadata; | ||
| import io.quarkiverse.solace.incoming.SolaceInboundMessage; | ||
| import io.smallrye.mutiny.Uni; | ||
|
|
||
| public class SolaceDiscard implements SolaceFailureHandler { | ||
| private final String channel; | ||
| private final AcknowledgementSupport ackSupport; | ||
|
|
||
| public SolaceDiscard(String channel, AcknowledgementSupport ackSupport) { | ||
| this.channel = channel; | ||
| this.ackSupport = ackSupport; | ||
| } | ||
|
|
||
| @Override | ||
| public CompletionStage<Void> handle(SolaceInboundMessage<?> msg, Throwable reason, Metadata metadata) { | ||
| MessageAcknowledgementConfiguration.Outcome outcome; | ||
| if (metadata != null) { | ||
| outcome = metadata.get(SettleMetadata.class) | ||
| .map(SettleMetadata::getOutcome) | ||
| .orElseGet(() -> MessageAcknowledgementConfiguration.Outcome.REJECTED /* TODO get outcome from reason */); | ||
| } else { | ||
| outcome = MessageAcknowledgementConfiguration.Outcome.REJECTED; | ||
| } | ||
|
|
||
| SolaceLogging.log.messageSettled(channel, outcome.toString().toLowerCase(), reason.getMessage()); | ||
| return Uni.createFrom().voidItem() | ||
| .invoke(() -> ackSupport.settle(msg.getMessage(), outcome)) | ||
| .runSubscriptionOn(msg::runOnMessageContext) | ||
| .subscribeAsCompletionStage(); | ||
| } | ||
| } |
73 changes: 73 additions & 0 deletions
73
pubsub-plus-connector/src/main/java/io/quarkiverse/solace/fault/SolaceErrorTopic.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,73 @@ | ||
| package io.quarkiverse.solace.fault; | ||
|
|
||
| import java.time.Duration; | ||
| import java.util.concurrent.CompletionStage; | ||
|
|
||
| import org.eclipse.microprofile.reactive.messaging.Metadata; | ||
|
|
||
| import com.solace.messaging.MessagingService; | ||
| import com.solace.messaging.config.MessageAcknowledgementConfiguration; | ||
| import com.solace.messaging.publisher.PersistentMessagePublisher; | ||
| import com.solace.messaging.receiver.AcknowledgementSupport; | ||
|
|
||
| import io.quarkiverse.solace.i18n.SolaceLogging; | ||
| import io.quarkiverse.solace.incoming.SolaceInboundMessage; | ||
| import io.smallrye.mutiny.Uni; | ||
|
|
||
| public class SolaceErrorTopic implements SolaceFailureHandler { | ||
| private final String channel; | ||
| private final AcknowledgementSupport ackSupport; | ||
| private final MessagingService solace; | ||
|
|
||
| private final SolaceErrorTopicPublisherHandler solaceErrorTopicPublisherHandler; | ||
| private long maxDeliveryAttempts; | ||
| private String errorTopic; | ||
| private boolean dmqEligible; | ||
| private Long timeToLive; | ||
|
|
||
| public SolaceErrorTopic(String channel, AcknowledgementSupport ackSupport, MessagingService solace) { | ||
| this.channel = channel; | ||
| this.ackSupport = ackSupport; | ||
| this.solace = solace; | ||
| this.solaceErrorTopicPublisherHandler = new SolaceErrorTopicPublisherHandler(solace); | ||
| } | ||
|
|
||
| public void setMaxDeliveryAttempts(long maxDeliveryAttempts) { | ||
| this.maxDeliveryAttempts = maxDeliveryAttempts; | ||
| } | ||
|
|
||
| public void setErrorTopic(String errorTopic) { | ||
| this.errorTopic = errorTopic; | ||
| } | ||
|
|
||
| public void setDmqEligible(boolean dmqEligible) { | ||
| this.dmqEligible = dmqEligible; | ||
| } | ||
|
|
||
| public void setTimeToLive(Long timeToLive) { | ||
| this.timeToLive = timeToLive; | ||
| } | ||
|
|
||
| @Override | ||
| public CompletionStage<Void> handle(SolaceInboundMessage<?> msg, Throwable reason, Metadata metadata) { | ||
| PersistentMessagePublisher.PublishReceipt publishReceipt = solaceErrorTopicPublisherHandler | ||
| .handle(msg, errorTopic, dmqEligible, timeToLive) | ||
| .onFailure().retry().withBackOff(Duration.ofSeconds(1)) | ||
| .atMost(maxDeliveryAttempts) | ||
| .subscribeAsCompletionStage().exceptionally((t) -> { | ||
| SolaceLogging.log.unsuccessfulToTopic(errorTopic, channel, | ||
| t.getMessage()); | ||
| return null; | ||
| }).join(); | ||
|
|
||
| if (publishReceipt != null) { | ||
| return Uni.createFrom().voidItem() | ||
| .invoke(() -> ackSupport.settle(msg.getMessage(), MessageAcknowledgementConfiguration.Outcome.ACCEPTED)) | ||
| .runSubscriptionOn(msg::runOnMessageContext) | ||
| .subscribeAsCompletionStage(); | ||
| } | ||
|
|
||
| return Uni.createFrom().<Void> failure(reason) | ||
| .emitOn(msg::runOnMessageContext).subscribeAsCompletionStage(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
22 changes: 9 additions & 13 deletions
22
...solace/incoming/SolaceFailureHandler.java → .../quarkiverse/solace/fault/SolaceFail.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
51 changes: 51 additions & 0 deletions
51
pubsub-plus-connector/src/main/java/io/quarkiverse/solace/fault/SolaceFailureHandler.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,51 @@ | ||
| package io.quarkiverse.solace.fault; | ||
|
|
||
| import static io.quarkiverse.solace.i18n.SolaceExceptions.ex; | ||
|
|
||
| import java.util.concurrent.CompletionStage; | ||
|
|
||
| import org.eclipse.microprofile.reactive.messaging.Metadata; | ||
|
|
||
| import io.quarkiverse.solace.incoming.SolaceInboundMessage; | ||
|
|
||
| public interface SolaceFailureHandler { | ||
|
|
||
| enum Strategy { | ||
| /** | ||
| * Mark the message as IGNORED, will continue processing with next message. | ||
| */ | ||
| IGNORE, | ||
| /** | ||
| * Mark the message as FAILED, broker will redeliver the message. | ||
| */ | ||
| FAIL, | ||
| /** | ||
| * Mark the message as REJECTED, broker will discard the message. The message will be moved to DMQ if DMQ is configured | ||
| * for queue and DMQ Eligible is set on message. | ||
| */ | ||
| DISCARD, | ||
| /** | ||
| * Will publish the message to configured error topic, on success the message will be acknowledged in the queue. | ||
| */ | ||
| ERROR_TOPIC; | ||
|
|
||
| public static Strategy from(String s) { | ||
| if (s == null || s.equalsIgnoreCase("ignore")) { | ||
| return IGNORE; | ||
| } | ||
| if (s.equalsIgnoreCase("fail")) { | ||
| return FAIL; | ||
| } | ||
| if (s.equalsIgnoreCase("discard")) { | ||
| return DISCARD; | ||
| } | ||
| if (s.equalsIgnoreCase("error_topic")) { | ||
| return ERROR_TOPIC; | ||
| } | ||
|
|
||
| throw ex.illegalArgumentUnknownFailureStrategy(s); | ||
| } | ||
| } | ||
|
|
||
| CompletionStage<Void> handle(SolaceInboundMessage<?> msg, Throwable reason, Metadata metadata); | ||
| } |
Oops, something went wrong.