[eclipse-ditto#964] Add subject expiry notification duration to Activ…

…ateTokenIntegration.

Signed-off-by: Yufei Cai <yufei.cai@bosch.io>

model/base/src/main/java/org/eclipse/ditto/model/base/headers/DittoDuration.java

@@ -142,6 +142,28 @@ public ChronoUnit getChronoUnit() {
         return dittoTimeUnit.getChronoUnit();
     }
 
+    /**
+     * Set the duration according to a Java duration keeping the time unit.
+     *
+     * @param duration the duration.
+     * @return the new duration with adjusted amount.
+     */
+    public DittoDuration setAmount(final Duration duration) {
+        final Duration unit = dittoTimeUnit.getChronoUnit().getDuration();
+        final long seconds = duration.getSeconds();
+        final long nanoseconds = duration.getNano();
+        final long unitSeconds = unit.getSeconds();
+        final long unitNanoseconds = unit.getNano();
+        final long amount;
+        if (unitSeconds != 0) {
+            amount = Math.max(1L, seconds / unitSeconds);
+        } else {
+            final long withOverflow = seconds * (1_000_000_000L / unitNanoseconds) + (nanoseconds / unitNanoseconds);
+            amount = Math.max(1L, withOverflow);
+        }
+        return new DittoDuration(amount, dittoTimeUnit);
+    }
+
     @Override
     public int length() {
         return toString().length();

model/base/src/test/java/org/eclipse/ditto/model/base/headers/DittoDurationTest.java

@@ -191,4 +191,19 @@ public void tryToParseDurationWithNegativeAmountWithoutSuffix() {
                 .withNoCause();
     }
 
-}
+    @Test
+    public void testSetAmount() {
+        final DittoDuration hours = DittoDuration.parseDuration("1h");
+        final DittoDuration minutes = DittoDuration.parseDuration("1m");
+        final DittoDuration seconds = DittoDuration.parseDuration("1s");
+        final DittoDuration millis = DittoDuration.parseDuration("1ms");
+
+        final Duration duration = Duration.ofHours(10).plus(Duration.ofMillis(2030));
+
+        assertThat(hours.setAmount(duration)).isEqualTo(DittoDuration.parseDuration("10h"));
+        assertThat(minutes.setAmount(duration)).isEqualTo(DittoDuration.parseDuration("600m"));
+        assertThat(seconds.setAmount(duration)).isEqualTo(DittoDuration.parseDuration("36002s"));
+        assertThat(millis.setAmount(duration)).isEqualTo(DittoDuration.parseDuration("36002030ms"));
+    }
+
+}

model/policies/src/main/java/org/eclipse/ditto/model/policies/ImmutableSubjectExpiry.java

@@ -19,6 +19,7 @@
 import java.time.temporal.ChronoUnit;
 import java.util.EnumSet;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Set;
 
 import javax.annotation.Nonnull;
@@ -78,7 +79,11 @@ static ImmutableSubjectExpiry parseAndValidate(final CharSequence expiry,
      * @throws NullPointerException if {@code expiry} is {@code null}.
      */
     public static SubjectExpiry of(final Instant expiry) {
-        return new ImmutableSubjectExpiry(checkNotNull(expiry, "expiry"), null);
+        return of(expiry, null);
+    }
+
+    static SubjectExpiry of(final Instant expiry, @Nullable final DittoDuration notifyBefore) {
+        return new ImmutableSubjectExpiry(checkNotNull(expiry, "expiry"), notifyBefore);
     }
 
     static SubjectExpiry fromJson(final JsonValue jsonValue) {
@@ -102,6 +107,11 @@ public boolean isExpired() {
         return timestamp.isBefore(Instant.now());
     }
 
+    @Override
+    public Optional<DittoDuration> getNotifyBefore() {
+        return Optional.ofNullable(notifyBefore);
+    }
+
     @Override
     public JsonValue toJson() {
         if (notifyBefore == null) {

model/policies/src/main/java/org/eclipse/ditto/model/policies/SubjectExpiry.java

@@ -13,12 +13,15 @@
 package org.eclipse.ditto.model.policies;
 
 import java.time.Instant;
+import java.util.Optional;
 
+import javax.annotation.Nullable;
 import javax.annotation.concurrent.Immutable;
 
 import org.eclipse.ditto.json.JsonFactory;
 import org.eclipse.ditto.json.JsonFieldDefinition;
 import org.eclipse.ditto.json.JsonValue;
+import org.eclipse.ditto.model.base.headers.DittoDuration;
 import org.eclipse.ditto.model.base.json.FieldType;
 import org.eclipse.ditto.model.base.json.JsonSchemaVersion;
 
@@ -54,6 +57,18 @@ static SubjectExpiry newInstance(final Instant expiry) {
         return PoliciesModelFactory.newSubjectExpiry(expiry);
     }
 
+    /**
+     * Returns a new {@link SubjectExpiry} with the specified {@code expiry} Instant.
+     *
+     * @param expiry the expiry Instant.
+     * @param notifyBefore the duration before the subject expiration when a notification should be sent.
+     * @return the new {@link SubjectExpiry}.
+     * @throws NullPointerException if {@code expiry} is {@code null}.
+     */
+    static SubjectExpiry newInstance(final Instant expiry, @Nullable final DittoDuration notifyBefore) {
+        return ImmutableSubjectExpiry.of(expiry, notifyBefore);
+    }
+
     static SubjectExpiry fromJson(final JsonValue jsonValue) {
         return ImmutableSubjectExpiry.fromJson(jsonValue);
     }
@@ -72,6 +87,13 @@ static SubjectExpiry fromJson(final JsonValue jsonValue) {
      */
     boolean isExpired();
 
+    /**
+     * Returns the duration before expiration when a notification should be sent.
+     *
+     * @return the notify-before duration.
+     */
+    Optional<DittoDuration> getNotifyBefore();
+
     /**
      * Returns a JSON representation of the subject expiry.
      *

services/policies/common/src/main/java/org/eclipse/ditto/services/policies/common/config/DefaultPolicyConfig.java

@@ -40,14 +40,17 @@ public final class DefaultPolicyConfig implements PolicyConfig {
     private final ActivityCheckConfig activityCheckConfig;
     private final SnapshotConfig snapshotConfig;
     private final Duration policySubjectExpiryGranularity;
+    private final Duration policySubjectExpiryNotificationGranularity;
     private final String subjectIdResolver;
 
     private DefaultPolicyConfig(final ScopedConfig scopedConfig) {
         supervisorConfig = DefaultSupervisorConfig.of(scopedConfig);
         activityCheckConfig = DefaultActivityCheckConfig.of(scopedConfig);
         snapshotConfig = DefaultSnapshotConfig.of(scopedConfig);
-        policySubjectExpiryGranularity = scopedConfig.getDuration(
-                PolicyConfigValue.SUBJECT_EXPIRY_GRANULARITY.getConfigPath());
+        policySubjectExpiryGranularity =
+                scopedConfig.getDuration(PolicyConfigValue.SUBJECT_EXPIRY_GRANULARITY.getConfigPath());
+        policySubjectExpiryNotificationGranularity =
+                scopedConfig.getDuration(PolicyConfigValue.SUBJECT_EXPIRY_NOTIFICATION_GRANULARITY.getConfigPath());
         subjectIdResolver = scopedConfig.getString(PolicyConfigValue.SUBJECT_ID_RESOLVER.getConfigPath());
     }
 
@@ -84,6 +87,11 @@ public Duration getSubjectExpiryGranularity() {
         return policySubjectExpiryGranularity;
     }
 
+    @Override
+    public Duration getSubjectExpiryNotificationGranularity() {
+        return policySubjectExpiryNotificationGranularity;
+    }
+
     @Override
     public String getSubjectIdResolver() {
         return subjectIdResolver;
@@ -102,13 +110,15 @@ public boolean equals(final Object o) {
                 Objects.equals(activityCheckConfig, that.activityCheckConfig) &&
                 Objects.equals(snapshotConfig, that.snapshotConfig) &&
                 Objects.equals(policySubjectExpiryGranularity, that.policySubjectExpiryGranularity) &&
+                Objects.equals(policySubjectExpiryNotificationGranularity,
+                        that.policySubjectExpiryNotificationGranularity) &&
                 Objects.equals(subjectIdResolver, that.subjectIdResolver);
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(supervisorConfig, activityCheckConfig, snapshotConfig, policySubjectExpiryGranularity,
-                subjectIdResolver);
+                policySubjectExpiryNotificationGranularity, subjectIdResolver);
     }
 
     @Override
@@ -118,6 +128,7 @@ public String toString() {
                 ", activityCheckConfig=" + activityCheckConfig +
                 ", snapshotConfig=" + snapshotConfig +
                 ", policySubjectExpiryGranularity=" + policySubjectExpiryGranularity +
+                ", policySubjectExpiryNotificationGranularity=" + policySubjectExpiryNotificationGranularity +
                 ", subjectIdResolver=" + subjectIdResolver +
                 "]";
     }

services/policies/common/src/main/java/org/eclipse/ditto/services/policies/common/config/PolicyConfig.java

@@ -44,6 +44,13 @@ public interface PolicyConfig extends WithSupervisorConfig, WithActivityCheckCon
      */
     Duration getSubjectExpiryGranularity();
 
+    /**
+     * Returns the configuration to which duration the notify-before duration of each subject-expiry is rounded up.
+     *
+     * @return the granularity.
+     */
+    Duration getSubjectExpiryNotificationGranularity();
+
     /**
      * Return the class responsible for placeholder resolution in the subject ID of policy action commands.
      *
@@ -61,6 +68,11 @@ enum PolicyConfigValue implements KnownConfigValue {
          */
         SUBJECT_EXPIRY_GRANULARITY("subject-expiry-granularity", Duration.ofHours(1L)),
 
+        /**
+         * The granularity to round up notify-before duration of subject-expiry.
+         */
+        SUBJECT_EXPIRY_NOTIFICATION_GRANULARITY("subject-expiry-notification-granularity", Duration.ofMinutes(1L)),
+
         SUBJECT_ID_RESOLVER("subject-id-resolver",
                 "org.eclipse.ditto.services.policies.persistence.actors.resolvers.DefaultSubjectIdFromActionResolver");
 

services/policies/common/src/test/java/org/eclipse/ditto/services/policies/common/config/DefaultPolicyConfigTest.java

@@ -66,6 +66,10 @@ public void underTestReturnsDefaultValuesIfBaseConfigWasEmpty() {
                 .as(PolicyConfig.PolicyConfigValue.SUBJECT_EXPIRY_GRANULARITY.getConfigPath())
                 .isEqualTo(Duration.ofHours(1L));
 
+        softly.assertThat(underTest.getSubjectExpiryNotificationGranularity())
+                .as(PolicyConfig.PolicyConfigValue.SUBJECT_EXPIRY_NOTIFICATION_GRANULARITY.getConfigPath())
+                .isEqualTo(Duration.ofMinutes(1L));
+
         softly.assertThat(underTest.getSubjectIdResolver())
                 .as(PolicyConfig.PolicyConfigValue.SUBJECT_ID_RESOLVER.getConfigPath())
                 .isEqualTo(PolicyConfig.PolicyConfigValue.SUBJECT_ID_RESOLVER.getDefaultValue());
@@ -79,6 +83,10 @@ public void underTestReturnsValuesOfBaseConfig() {
                 .as(PolicyConfig.PolicyConfigValue.SUBJECT_EXPIRY_GRANULARITY.getConfigPath())
                 .isEqualTo(Duration.ofSeconds(10));
 
+        softly.assertThat(underTest.getSubjectExpiryNotificationGranularity())
+                .as(PolicyConfig.PolicyConfigValue.SUBJECT_EXPIRY_NOTIFICATION_GRANULARITY.getConfigPath())
+                .isEqualTo(Duration.ofSeconds(11L));
+
         softly.assertThat(underTest.getSubjectIdResolver())
                 .as(PolicyConfig.PolicyConfigValue.SUBJECT_ID_RESOLVER.getConfigPath())
                 .isEqualTo("IrredeemableSubjectIdResolver");

services/policies/common/src/test/resources/policy-test.conf

@@ -1,5 +1,6 @@
 policy {
   subject-expiry-granularity = 10s
+  subject-expiry-notification-granularity = 11s
 
   subject-id-resolver = "IrredeemableSubjectIdResolver"
 

services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/AbstractPolicyCommandStrategy.java

@@ -27,6 +27,7 @@
 
 import org.eclipse.ditto.model.base.entity.metadata.Metadata;
 import org.eclipse.ditto.model.base.exceptions.DittoRuntimeException;
+import org.eclipse.ditto.model.base.headers.DittoDuration;
 import org.eclipse.ditto.model.base.headers.DittoHeaders;
 import org.eclipse.ditto.model.policies.Label;
 import org.eclipse.ditto.model.policies.Policy;
@@ -62,10 +63,12 @@ abstract class AbstractPolicyCommandStrategy<C extends Command<C>, E extends Pol
         extends AbstractConditionHeaderCheckingCommandStrategy<C, Policy, PolicyId, E> {
 
     private final PolicyExpiryGranularity policyExpiryGranularity;
+    private final Duration policyExpiryNotificationGranularity;
 
     AbstractPolicyCommandStrategy(final Class<C> theMatchingClass, final PolicyConfig policyConfig) {
         super(theMatchingClass);
         policyExpiryGranularity = calculateTemporalUnitAndAmount(policyConfig);
+        policyExpiryNotificationGranularity = policyConfig.getSubjectExpiryNotificationGranularity();
     }
 
     static PolicyExpiryGranularity calculateTemporalUnitAndAmount(final PolicyConfig policyConfig) {
@@ -222,7 +225,15 @@ protected SubjectExpiry roundPolicySubjectExpiry(final SubjectExpiry expiry) {
         final long toAdd = policyExpiryGranularity.amount - deltaModulo;
         final Instant roundedUp = truncated.plus(toAdd, policyExpiryGranularity.temporalUnit);
 
-        return SubjectExpiry.newInstance(roundedUp);
+        final var roundedUpNotifyBefore = expiry.getNotifyBefore()
+                .map(notifyBefore -> {
+                    final var roundedUpDuration =
+                            roundUpDuration(notifyBefore.getDuration(), policyExpiryNotificationGranularity);
+                    return notifyBefore.setAmount(roundedUpDuration);
+                })
+                .orElse(null);
+
+        return SubjectExpiry.newInstance(roundedUp, roundedUpNotifyBefore);
     }
 
     /**
@@ -297,6 +308,14 @@ static DittoRuntimeException policyEntryInvalid(final PolicyId policyId, final L
                 .build();
     }
 
+    private static Duration roundUpDuration(final Duration duration, final Duration granularity) {
+        final long granularitySeconds = Math.max(1L, granularity.getSeconds());
+        final long durationSeconds = Math.max(granularitySeconds, duration.getSeconds());
+        final long roundedUpSeconds =
+                ((durationSeconds + granularitySeconds - 1L) / granularitySeconds) * granularitySeconds;
+        return Duration.ofSeconds(roundedUpSeconds);
+    }
+
     private static class PolicyExpiryGranularity {
 
         private final ChronoUnit temporalUnit;

services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ActivateTokenIntegrationStrategy.java

@@ -73,7 +73,7 @@ protected Result<PolicyActionEvent<?>> doApply(final Context<PolicyId> context,
         final Policy nonNullPolicy = checkNotNull(policy, "policy");
         final PolicyId policyId = context.getState();
         final Label label = command.getLabel();
-        final SubjectExpiry commandSubjectExpiry = SubjectExpiry.newInstance(command.getExpiry());
+        final SubjectExpiry commandSubjectExpiry = command.getSubjectExpiry();
         final DittoHeaders dittoHeaders = command.getDittoHeaders();
 
         final Optional<PolicyEntry> optionalEntry = nonNullPolicy.getEntryFor(label)

services/policies/persistence/src/test/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/AbstractPolicyCommandStrategyTest.java

@@ -18,6 +18,8 @@
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.verify;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.function.Consumer;
 
 import javax.annotation.Nullable;
@@ -158,15 +160,39 @@ private static <T extends PolicyEvent<?>> T assertModificationResult(final Resul
         return event.getValue();
     }
 
-    private static <C extends Command<?>> Result<?> applyStrategy(
-            final CommandStrategy<C, Policy, PolicyId, ?> underTest,
+    static <C extends Command<?>, E extends Event<?>> Result<E> applyStrategy(
+            final CommandStrategy<C, Policy, PolicyId, E> underTest,
             final CommandStrategy.Context<PolicyId> context,
             @Nullable final Policy policy,
             final C command) {
 
         return underTest.apply(context, policy, NEXT_REVISION, command);
     }
 
+    static <E extends Event<?>> E getEvent(final Result<E> result) {
+        final List<E> box = new ArrayList<>(1);
+        result.accept(new ResultVisitor<>() {
+            @Override
+            public void onMutation(final Command<?> command, final E event, final WithDittoHeaders<?> response,
+                    final boolean becomeCreated,
+                    final boolean becomeDeleted) {
+
+                box.add(event);
+            }
+
+            @Override
+            public void onQuery(final Command<?> command, final WithDittoHeaders<?> response) {
+                throw new AssertionError("Expect mutation result, got query response: " + response);
+            }
+
+            @Override
+            public void onError(final DittoRuntimeException error, final Command<?> errorCausingCommand) {
+                throw new AssertionError("Expect mutation result, got error: " + error);
+            }
+        });
+        return box.get(0);
+    }
+
     @SuppressWarnings({"unchecked", "rawtypes"})
     private static <T extends Event<?>> Dummy<T> cast(final Dummy<?> dummy) {
         return (Dummy) dummy;