Performance optimizations for Multi-Assignments

hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpMessageDispatcherService.java

@@ -16,6 +16,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import org.eclipse.hawkbit.api.ApiType;
@@ -65,8 +66,6 @@
 import org.springframework.data.domain.PageRequest;
 import org.springframework.util.CollectionUtils;
 
-import com.google.common.collect.Maps;
-
 /**
  * {@link AmqpMessageDispatcherService} create all outgoing AMQP messages and
  * delegate the messages to a {@link AmqpMessageSenderService}.
@@ -172,30 +171,29 @@ protected void onMultiAction(final MultiActionEvent e) {
 
     protected void sendMultiActionRequestMessages(final String tenant, final List<String> controllerIds) {
 
-        final Map<Long, Map<SoftwareModule, List<SoftwareModuleMetadata>>> softwareModuleMetadata = new HashMap<>();
+        final Map<SoftwareModule, List<SoftwareModuleMetadata>> softwareModuleMetadata = new HashMap<>();
         targetManagement.getByControllerID(controllerIds).stream()
                 .filter(target -> IpUtil.isAmqpUri(target.getAddress())).forEach(target -> {
 
                     final List<Action> activeActions = deploymentManagement
                             .findActiveActionsByTarget(PageRequest.of(0, MAX_ACTION_COUNT), target.getControllerId())
                             .getContent();
 
-                    activeActions.forEach(action -> {
-                        final DistributionSet distributionSet = action.getDistributionSet();
-                        softwareModuleMetadata.computeIfAbsent(distributionSet.getId(),
-                                id -> getSoftwareModulesWithMetadata(distributionSet));
-                    });
+                    activeActions.forEach(action -> action.getDistributionSet().getModules().forEach(
+                            module -> softwareModuleMetadata.computeIfAbsent(module, this::getSoftwareModuleMetadata)));
 
                     if (!activeActions.isEmpty()) {
-                        sendMultiActionRequestToTarget(tenant, target, activeActions, softwareModuleMetadata);
+                        sendMultiActionRequestToTarget(tenant, target, activeActions,
+                                action -> action.getDistributionSet().getModules().stream()
+                                        .collect(Collectors.toMap(m -> m, softwareModuleMetadata::get)));
                     }
 
                 });
 
     }
 
     protected void sendMultiActionRequestToTarget(final String tenant, final Target target, final List<Action> actions,
-            final Map<Long, Map<SoftwareModule, List<SoftwareModuleMetadata>>> softwareModulesPerDistributionSet) {
+            final Function<Action, Map<SoftwareModule, List<SoftwareModuleMetadata>>> getSoftwareModuleMetaData) {
 
         final URI targetAdress = target.getAddress();
         if (!IpUtil.isAmqpUri(targetAdress) || CollectionUtils.isEmpty(actions)) {
@@ -205,7 +203,7 @@ protected void sendMultiActionRequestToTarget(final String tenant, final Target
         final DmfMultiActionRequest multiActionRequest = new DmfMultiActionRequest();
         actions.forEach(action -> {
             final DmfActionRequest actionRequest = createDmfActionRequest(target, action,
-                    softwareModulesPerDistributionSet.get(action.getDistributionSet().getId()));
+                    getSoftwareModuleMetaData.apply(action));
             multiActionRequest.addElement(getEventTypeForAction(action), actionRequest);
         });
 
@@ -471,15 +469,12 @@ private DmfArtifact convertArtifact(final Target target, final Artifact localArt
 
     private Map<SoftwareModule, List<SoftwareModuleMetadata>> getSoftwareModulesWithMetadata(
             final DistributionSet distributionSet) {
-        final Map<SoftwareModule, List<SoftwareModuleMetadata>> moduleMetadata = Maps
-                .newHashMapWithExpectedSize(distributionSet.getModules().size());
-        distributionSet.getModules()
-                .forEach(
-                        module -> moduleMetadata.put(module,
-                                softwareModuleManagement.findMetaDataBySoftwareModuleIdAndTargetVisible(
-                                        PageRequest.of(0, RepositoryConstants.MAX_META_DATA_COUNT), module.getId())
-                                        .getContent()));
-        return moduleMetadata;
+        return distributionSet.getModules().stream().collect(Collectors.toMap(m -> m, this::getSoftwareModuleMetadata));
+    }
+
+    private List<SoftwareModuleMetadata> getSoftwareModuleMetadata(final SoftwareModule module) {
+        return softwareModuleManagement.findMetaDataBySoftwareModuleIdAndTargetVisible(
+                PageRequest.of(0, RepositoryConstants.MAX_META_DATA_COUNT), module.getId()).getContent();
     }
 
 }

hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpMessageHandlerService.java

@@ -227,7 +227,7 @@ private void sendCurrentActionsAsMultiActionToTarget(final Target target) {
                 .stream().collect(Collectors.toMap(DistributionSet::getId, this::getSoftwareModulesWithMetadata));
 
         amqpMessageDispatcherService.sendMultiActionRequestToTarget(target.getTenant(), target, actions,
-                softwareModulesPerDistributionSet);
+                action -> softwareModulesPerDistributionSet.get(action.getDistributionSet().getId()));
     }
 
     private void sendOldestActionToTarget(final Target target) {

hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/integration/AmqpMessageDispatcherServiceIntegrationTest.java

@@ -135,21 +135,21 @@ public void sendDownloadAndInstallStatusMessageDuringMaintenanceWindow() {
     @Test
     @Description("Verify that a distribution assignment multiple times send cancel and assign events with right softwaremodules")
     @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1),
-            @Expect(type = TargetAssignDistributionSetEvent.class, count = 1),
+            @Expect(type = TargetAssignDistributionSetEvent.class, count = 2),
             @Expect(type = CancelTargetAssignmentEvent.class, count = 1),
             @Expect(type = ActionCreatedEvent.class, count = 2), @Expect(type = ActionUpdatedEvent.class, count = 1),
             @Expect(type = SoftwareModuleCreatedEvent.class, count = 6),
-            @Expect(type = SoftwareModuleUpdatedEvent.class, count = 6),
+            @Expect(type = SoftwareModuleUpdatedEvent.class, count = 12),
             @Expect(type = DistributionSetCreatedEvent.class, count = 2),
-            @Expect(type = TargetUpdatedEvent.class, count = 2), @Expect(type = TargetPollEvent.class, count = 3) })
+            @Expect(type = TargetUpdatedEvent.class, count = 2), @Expect(type = TargetPollEvent.class, count = 2) })
     public void assignDistributionSetMultipleTimes() {
         final String controllerId = TARGET_PREFIX + "assignDistributionSetMultipleTimes";
 
         final DistributionSetAssignmentResult assignmentResult = registerTargetAndAssignDistributionSet(controllerId);
-
         final DistributionSet distributionSet2 = testdataFactory.createDistributionSet(UUID.randomUUID().toString());
-        registerTargetAndAssignDistributionSet(distributionSet2.getId(), TargetUpdateStatus.PENDING,
-                getDistributionSet().getModules(), controllerId);
+        testdataFactory.addSoftwareModuleMetadata(distributionSet2);
+        assignDistributionSet(distributionSet2.getId(), controllerId);
+        assertDownloadAndInstallMessage(distributionSet2.getModules(), controllerId);
         assertCancelActionMessage(assignmentResult.getActionIds().get(0), controllerId);
 
         createAndSendThingCreated(controllerId, TENANT_EXIST);

hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/matcher/SoftwareModuleJsonMatcher.java

@@ -57,50 +57,27 @@ public SoftwareModulesMatcher(final List<DmfSoftwareModule> expectedModules) {
             this.expectedModules = expectedModules;
         }
 
-        static boolean containsExactly(final Object actual, final List<DmfSoftwareModule> expected) {
+        boolean containsExactly(final Object actual) {
             if (actual == null) {
-                return expected == null;
+                return expectedModules == null;
             }
 
             @SuppressWarnings("unchecked")
             final Collection<SoftwareModule> modules = (Collection<SoftwareModule>) actual;
 
-            if (modules.size() != expected.size()) {
-                return false;
-            }
-
-            for (final SoftwareModule repoSoftwareModule : modules) {
-                boolean containsElement = false;
-
-                for (final DmfSoftwareModule jsonSoftwareModule : expected) {
-                    if (!jsonSoftwareModule.getModuleId().equals(repoSoftwareModule.getId())) {
-                        continue;
-                    }
-                    containsElement = true;
-
-                    if (!jsonSoftwareModule.getModuleType().equals(repoSoftwareModule.getType().getKey())) {
-                        return false;
-                    }
-                    if (!jsonSoftwareModule.getModuleVersion().equals(repoSoftwareModule.getVersion())) {
-                        return false;
-                    }
-                    if (jsonSoftwareModule.getArtifacts().size() != repoSoftwareModule.getArtifacts().size()) {
-                        return false;
-                    }
-                }
-
-                if (!containsElement) {
-                    return false;
-                }
-
-            }
+            return expectedModules.stream().allMatch(e -> existsIn(e, modules));
+        }
 
-            return true;
+        private static boolean existsIn(final DmfSoftwareModule module, final Collection<SoftwareModule> actual) {
+            return actual.stream()
+                    .anyMatch(e -> module.getModuleType().equals(e.getType().getKey())
+                            && module.getModuleVersion().equals(e.getVersion())
+                            && module.getArtifacts().size() == e.getArtifacts().size());
         }
 
         @Override
         public boolean matches(final Object actualValue) {
-            return containsExactly(actualValue, expectedModules);
+            return containsExactly(actualValue);
         }
 
         @Override

hawkbit-dmf/hawkbit-dmf-api/src/main/java/org/eclipse/hawkbit/dmf/json/model/DmfMetadata.java

@@ -79,4 +79,8 @@ public boolean equals(final Object obj) {
         return true;
     }
 
+    @Override
+    public String toString() {
+        return String.format("DmfMetadata [key='%s', value='%s']", key, value);
+    }
 }

hawkbit-dmf/hawkbit-dmf-api/src/main/java/org/eclipse/hawkbit/dmf/json/model/DmfSoftwareModule.java

@@ -83,4 +83,11 @@ public void setMetadata(final List<DmfMetadata> metadata) {
         }
     }
 
+    @Override
+    public String toString() {
+        return String.format(
+                "DmfSoftwareModule [moduleId=%d, moduleType='%s', moduleVersion='%s', artifacts=%s, metadata=%s]",
+                moduleId, moduleType, moduleVersion, artifacts, metadata);
+    }
+
 }

hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/ActionRepository.java

@@ -127,6 +127,20 @@ public interface ActionRepository extends BaseEntityRepository<JpaAction, Long>,
     @Query("SELECT CASE WHEN COUNT(a)>0 THEN 'true' ELSE 'false' END FROM JpaAction a JOIN a.target t WHERE t.controllerId=:controllerId AND a.active=1")
     boolean activeActionExistsForControllerId(@Param("controllerId") String controllerId);
 
+    /**
+     * Check if any active actions with given action status and given controller
+     * ID exist.
+     *
+     * @param controllerId
+     *            of the target to check for actions
+     * @param currentStatus
+     *            of the active action to look for
+     *
+     * @return <code>true</code> if one or more active actions for the given
+     *         controllerId and action status are found
+     */
+    boolean existsByTargetControllerIdAndStatusAndActiveIsTrue(String controllerId, Action.Status currentStatus);
+
     /**
      * Retrieves latest {@link Action} for given target and
      * {@link SoftwareModule}.

hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/OnlineDsAssignmentStrategy.java

@@ -8,8 +8,6 @@
  */
 package org.eclipse.hawkbit.repository.jpa;
 
-import static org.eclipse.hawkbit.repository.RepositoryConstants.MAX_ACTION_COUNT;
-
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
@@ -39,8 +37,6 @@
 import org.eclipse.hawkbit.repository.model.TargetWithActionType;
 import org.springframework.cloud.bus.BusProperties;
 import org.springframework.context.ApplicationEventPublisher;
-import org.springframework.data.domain.PageRequest;
-import org.springframework.data.domain.Pageable;
 import org.springframework.util.CollectionUtils;
 
 import com.google.common.collect.Lists;
@@ -51,8 +47,6 @@
  */
 public class OnlineDsAssignmentStrategy extends AbstractDsAssignmentStrategy {
 
-    private static final Pageable ACTION_PAGE_REQUEST = PageRequest.of(0, MAX_ACTION_COUNT);
-
     private final Supplier<Boolean> multiAssignmentsConfig;
 
     OnlineDsAssignmentStrategy(final TargetRepository targetRepository,
@@ -196,8 +190,8 @@ private void sendTargetAssignDistributionSetEvent(final String tenant, final lon
     }
 
     private boolean hasPendingCancellations(final Target target) {
-        return actionRepository.findByActiveAndTarget(ACTION_PAGE_REQUEST, target.getControllerId(), true).getContent()
-                .stream().anyMatch(action -> action.getStatus() == Status.CANCELING);
+        return actionRepository.existsByTargetControllerIdAndStatusAndActiveIsTrue(target.getControllerId(),
+                Status.CANCELING);
     }
 
     /**

hawkbit-repository/hawkbit-repository-jpa/src/test/java/org/eclipse/hawkbit/repository/event/remote/AbstractRemoteEventTest.java

@@ -79,13 +79,15 @@ protected Message<?> createMessageWithImmutableHeader(final TenantAwareEvent eve
         return busProtoStuffMessageConverter.toMessage(event, new MessageHeaders(headers));
     }
 
-    protected TenantAwareEvent createJacksonEvent(final TenantAwareEvent event) {
-        final Message<?> message = createJsonMessage(event);
-        return (TenantAwareEvent) jacksonMessageConverter.fromMessage(message, event.getClass());
+    @SuppressWarnings("unchecked")
+    protected <T extends TenantAwareEvent> T createJacksonEvent(final T event) {
+        final Message<String> message = createJsonMessage(event);
+        return (T) jacksonMessageConverter.fromMessage(message, event.getClass());
     }
 
-    protected TenantAwareEvent createProtoStuffEvent(final TenantAwareEvent event) {
+    @SuppressWarnings("unchecked")
+    protected <T extends TenantAwareEvent> T createProtoStuffEvent(final T event) {
         final Message<?> message = createProtoStuffMessage(event);
-        return (TenantAwareEvent) busProtoStuffMessageConverter.fromMessage(message, event.getClass());
+        return (T) busProtoStuffMessageConverter.fromMessage(message, event.getClass());
     }
 }

hawkbit-repository/hawkbit-repository-jpa/src/test/java/org/eclipse/hawkbit/repository/event/remote/RemoteIdEventTest.java

@@ -94,10 +94,10 @@ protected void assertAndCreateRemoteEvent(final Class<? extends RemoteIdEvent> e
     protected void assertEntity(final RemoteIdEvent event) {
         assertThat(event.getEntityId()).isSameAs(ENTITY_ID);
 
-        final RemoteIdEvent protoStuffEvent = (RemoteIdEvent) createProtoStuffEvent(event);
+        final RemoteIdEvent protoStuffEvent = createProtoStuffEvent(event);
         assertDeserializeEvent(protoStuffEvent, event);
 
-        final RemoteIdEvent jacksonEvent = (RemoteIdEvent) createJacksonEvent(event);
+        final RemoteIdEvent jacksonEvent = createJacksonEvent(event);
         assertDeserializeEvent(jacksonEvent, event);
     }