-
Notifications
You must be signed in to change notification settings - Fork 219
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Extract RequestedAcksFilter from InboundDispatchingActor; switch to e…
…xternal message dispatcher. Reason: typical QoS headers are available in external message headers and not in the signal headers. Signed-off-by: Yufei Cai <yufei.cai@bosch.io>
- Loading branch information
Showing
5 changed files
with
206 additions
and
95 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
90 changes: 90 additions & 0 deletions
90
.../src/main/java/org/eclipse/ditto/services/connectivity/messaging/RequestedAcksFilter.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,90 @@ | ||
| /* | ||
| * Copyright (c) 2020 Contributors to the Eclipse Foundation | ||
| * | ||
| * See the NOTICE file(s) distributed with this work for additional | ||
| * information regarding copyright ownership. | ||
| * | ||
| * This program and the accompanying materials are made available under the | ||
| * terms of the Eclipse Public License 2.0 which is available at | ||
| * http://www.eclipse.org/legal/epl-2.0 | ||
| * | ||
| * SPDX-License-Identifier: EPL-2.0 | ||
| */ | ||
| package org.eclipse.ditto.services.connectivity.messaging; | ||
|
|
||
| import java.util.Collections; | ||
| import java.util.Optional; | ||
|
|
||
| import javax.annotation.Nullable; | ||
|
|
||
| import org.eclipse.ditto.model.base.headers.DittoHeaderDefinition; | ||
| import org.eclipse.ditto.model.base.headers.DittoHeaders; | ||
| import org.eclipse.ditto.model.connectivity.ConnectionId; | ||
| import org.eclipse.ditto.model.placeholders.ExpressionResolver; | ||
| import org.eclipse.ditto.services.models.connectivity.ExternalMessage; | ||
| import org.eclipse.ditto.signals.base.Signal; | ||
|
|
||
| /** | ||
| * Execute filter for requested-acks configured in connection sources. | ||
| */ | ||
| final class RequestedAcksFilter { | ||
|
|
||
| /** | ||
| * Apply the configured requested-acks filter to a signal mapped from an external message. | ||
| * - If the filter is tripped, set requested-acks to [] | ||
| * - If the filter is not tripped and requested-acks is defined, leave it | ||
| * - If the filter is not tripped, requested-acks is not defined, fallback to the resolved value of the filter. | ||
| * | ||
| * @param signal signal to filter requested acknowledges for | ||
| * @param externalMessage incoming external message that mapped to the signal | ||
| * @param filter the filter string | ||
| * @param connectionId the connection ID receiving the signal. | ||
| * @return the filtered signal. | ||
| */ | ||
| static Signal<?> filterAcknowledgements(final Signal<?> signal, | ||
| final ExternalMessage externalMessage, | ||
| final @Nullable String filter, | ||
| final ConnectionId connectionId) { | ||
| if (filter != null) { | ||
| final String requestedAcks = DittoHeaderDefinition.REQUESTED_ACKS.getKey(); | ||
| final boolean headerDefined = signal.getDittoHeaders().containsKey(requestedAcks); | ||
| final String requestedAcksValue = getDefaultRequestedAcks(headerDefined, signal); | ||
| final String fullFilter = "fn:default('" + requestedAcksValue + "')|" + filter; | ||
| final ExpressionResolver resolver = Resolvers.forExternalMessage(externalMessage, connectionId); | ||
| final Optional<String> resolverResult = resolver.resolveAsPipelineElement(fullFilter).toOptional(); | ||
| if (resolverResult.isEmpty()) { | ||
| // filter tripped: set requested-acks to [] | ||
| return signal.setDittoHeaders(DittoHeaders.newBuilder(signal.getDittoHeaders()) | ||
| .acknowledgementRequests(Collections.emptySet()) | ||
| .build()); | ||
| } else if (headerDefined) { | ||
| // filter not tripped, header defined | ||
| return signal.setDittoHeaders(DittoHeaders.newBuilder(signal.getDittoHeaders()) | ||
| .putHeader(requestedAcks, resolverResult.orElseThrow()) | ||
| .build()); | ||
| } else { | ||
| // filter not tripped, header not defined: | ||
| // - evaluate filter again against unresolved and set requested-acks accordingly | ||
| // - if filter is not resolved, then keep requested-acks undefined for the default behavior | ||
| final Optional<String> unsetFilterResult = | ||
| resolver.resolveAsPipelineElement(filter).toOptional(); | ||
| return unsetFilterResult.<Signal<?>>map(newAckRequests -> | ||
| signal.setDittoHeaders(DittoHeaders.newBuilder(signal.getDittoHeaders()) | ||
| .putHeader(requestedAcks, newAckRequests) | ||
| .build())) | ||
| .orElse(signal); | ||
| } | ||
| } | ||
| return signal; | ||
| } | ||
|
|
||
| private static String getDefaultRequestedAcks(final boolean headerDefined, final Signal<?> signal) { | ||
| if (headerDefined) { | ||
| final String headerValue = signal.getDittoHeaders().get(DittoHeaderDefinition.REQUESTED_ACKS.getKey()); | ||
| if (!headerValue.contains("'")) { | ||
| return headerValue; | ||
| } | ||
| } | ||
| return "[]"; | ||
| } | ||
| } |
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
Oops, something went wrong.