Merge release 1.2.0

camunda-kafka-model/pom.xml

@@ -7,7 +7,7 @@
     <parent>
         <groupId>de.viadee.camunda</groupId>
         <artifactId>camunda-kafka-polling-client-parent</artifactId>
-        <version>1.1.2</version>
+        <version>1.2.0</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

camunda-kafka-model/src/main/java/de/viadee/camunda/kafka/event/IdentityLinkEvent.java

@@ -0,0 +1,73 @@
+package de.viadee.camunda.kafka.event;
+
+import java.util.Date;
+
+/**
+ * <p>
+ * IdentityLinkEvent class.
+ * </p>
+ *
+ *
+ * @version $Id: $Id
+ */
+public class IdentityLinkEvent extends DetailEvent {
+
+    public enum OperationType {
+        add, delete
+    }
+
+    private String type;
+    private String userId;
+    private String groupId;
+    private OperationType operationType;
+    private String assignerId;
+    private Date removalTime;
+
+    public String getType() {
+        return type;
+    }
+
+    public void setType(String type) {
+        this.type = type;
+    }
+
+    public String getUserId() {
+        return userId;
+    }
+
+    public void setUserId(String userId) {
+        this.userId = userId;
+    }
+
+    public String getGroupId() {
+        return groupId;
+    }
+
+    public void setGroupId(String groupId) {
+        this.groupId = groupId;
+    }
+
+    public OperationType getOperationType() {
+        return operationType;
+    }
+
+    public void setOperationType(OperationType operationType) {
+        this.operationType = operationType;
+    }
+
+    public String getAssignerId() {
+        return assignerId;
+    }
+
+    public void setAssignerId(String assignerId) {
+        this.assignerId = assignerId;
+    }
+
+    public Date getRemovalTime() {
+        return removalTime;
+    }
+
+    public void setRemovalTime(Date removalTime) {
+        this.removalTime = removalTime;
+    }
+}

camunda-kafka-polling-client/pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <groupId>de.viadee.camunda</groupId>
         <artifactId>camunda-kafka-polling-client-parent</artifactId>
-        <version>1.1.2</version>
+        <version>1.2.0</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

camunda-kafka-polling-client/src/main/java/de/viadee/camunda/kafka/pollingclient/config/properties/ApplicationProperties.java

@@ -69,7 +69,9 @@ public enum PollingEvents {
          */
         VARIABLE_CURRENT_UNFINISHED,
         PROCESS_DEFINITION,
-        TASK_COMMENTS
+        TASK_COMMENTS,
+        IDENTITY_LINKS_UNFINISHED_ACTIVITIES,
+        IDENTITY_LINKS_FINISHED_ACTIVITIES
     }
 
     /**

camunda-kafka-polling-client/src/main/java/de/viadee/camunda/kafka/pollingclient/job/runtime/RuntimeDataPollingService.java

@@ -2,6 +2,7 @@
 
 import de.viadee.camunda.kafka.event.ActivityInstanceEvent;
 import de.viadee.camunda.kafka.event.CommentEvent;
+import de.viadee.camunda.kafka.event.IdentityLinkEvent;
 import de.viadee.camunda.kafka.event.ProcessInstanceEvent;
 import de.viadee.camunda.kafka.event.VariableUpdateEvent;
 import de.viadee.camunda.kafka.pollingclient.config.properties.ApplicationProperties;
@@ -148,6 +149,12 @@ private void pollUnfinishedActivities(final String processInstanceId, final Poll
                         && activityInstanceEvent.getActivityType().equals("userTask")) {
                     pollComments(activityInstanceEvent);
                 }
+
+                if (properties.getPollingEvents()
+                              .contains(ApplicationProperties.PollingEvents.IDENTITY_LINKS_UNFINISHED_ACTIVITIES)
+                        && activityInstanceEvent.getActivityType().equals("userTask")) {
+                    pollIdentityLinks(activityInstanceEvent);
+                }
             }
         }
     }
@@ -175,6 +182,12 @@ private void pollFinishedActivities(final String processInstanceId, final Pollin
                         && activityInstanceEvent.getActivityType().equals("userTask")) {
                     pollComments(activityInstanceEvent);
                 }
+
+                if (properties.getPollingEvents()
+                              .contains(ApplicationProperties.PollingEvents.IDENTITY_LINKS_FINISHED_ACTIVITIES)
+                        && activityInstanceEvent.getActivityType().equals("userTask")) {
+                    pollIdentityLinks(activityInstanceEvent);
+                }
             }
         }
     }
@@ -200,6 +213,13 @@ private void pollComments(final ActivityInstanceEvent activityInstanceEvent) {
         }
     }
 
+    private void pollIdentityLinks(final ActivityInstanceEvent activityInstanceEvent) {
+        for (final IdentityLinkEvent identityLinkEvent : pollingService
+                                                                       .pollIdentityLinks(activityInstanceEvent)) {
+            eventService.sendEvent(identityLinkEvent);
+        }
+    }
+
     private boolean isProcessInstanceStartedBetween(final ProcessInstanceEvent processInstanceEvent,
                                                     final Date startTime, final Date endTime) {
         if (processInstanceEvent.getStartTime() == null) {

camunda-kafka-polling-client/src/main/java/de/viadee/camunda/kafka/pollingclient/service/polling/PollingService.java

@@ -107,4 +107,12 @@ Iterable<ActivityInstanceEvent> pollUnfinishedActivities(String processInstanceI
      * @return a {@link java.lang.Iterable} object.
      */
     Iterable<CommentEvent> pollComments(ActivityInstanceEvent activityInstanceEvent);
+
+    /**
+     * Poll Identity-Links for specified task
+     *
+     * @param activityInstanceEvent
+     * @return a {@link java.lang.Iterable} object.
+     */
+    Iterable<IdentityLinkEvent> pollIdentityLinks(ActivityInstanceEvent activityInstanceEvent);
 }

camunda-kafka-polling-client/src/main/java/de/viadee/camunda/kafka/pollingclient/service/polling/jdbc/CamundaJdbcPollingServiceImpl.java

@@ -11,6 +11,7 @@
 import org.camunda.bpm.engine.TaskService;
 import org.camunda.bpm.engine.history.HistoricActivityInstance;
 import org.camunda.bpm.engine.history.HistoricDetail;
+import org.camunda.bpm.engine.history.HistoricIdentityLinkLog;
 import org.camunda.bpm.engine.history.HistoricProcessInstance;
 import org.camunda.bpm.engine.history.HistoricVariableInstance;
 import org.camunda.bpm.engine.impl.persistence.entity.HistoricDetailVariableInstanceUpdateEntity;
@@ -228,6 +229,16 @@ public Iterable<CommentEvent> pollComments(ActivityInstanceEvent activityInstanc
                           .map(comment -> createCommentEventFromDetails(comment, activityInstanceEvent))::iterator;
     }
 
+    @Override
+    public Iterable<IdentityLinkEvent> pollIdentityLinks(ActivityInstanceEvent activityInstanceEvent) {
+
+        return historyService.createHistoricIdentityLinkLogQuery()
+                             .taskId(activityInstanceEvent.getTaskId())
+                             .list()
+                             .stream()
+                             .map(historicIdentityLinkLog -> createIdentityLinkEventFromDetails(historicIdentityLinkLog))::iterator;
+    }
+
     private ProcessDefinitionEvent createProcessDefinitionEvent(Deployment d, ProcessDefinition pd) {
 
         ProcessDefinitionEvent e = new ProcessDefinitionEvent();
@@ -317,6 +328,27 @@ private CommentEvent createCommentEventFromDetails(
         return event;
     }
 
+    private IdentityLinkEvent createIdentityLinkEventFromDetails(HistoricIdentityLinkLog historicIdentityLinkLog) {
+
+        final IdentityLinkEvent event = new IdentityLinkEvent();
+
+        event.setId(historicIdentityLinkLog.getId());
+        event.setTimestamp(historicIdentityLinkLog.getTime());
+        event.setType(historicIdentityLinkLog.getType());
+        event.setUserId(historicIdentityLinkLog.getUserId());
+        event.setGroupId(historicIdentityLinkLog.getGroupId());
+        event.setTaskId(historicIdentityLinkLog.getTaskId());
+        event.setProcessDefinitionId(historicIdentityLinkLog.getProcessDefinitionId());
+        event.setProcessDefinitionKey(historicIdentityLinkLog.getProcessDefinitionKey());
+        event.setOperationType(IdentityLinkEvent.OperationType.valueOf(historicIdentityLinkLog.getOperationType()));
+        event.setAssignerId(historicIdentityLinkLog.getAssignerId());
+        event.setTenantId(historicIdentityLinkLog.getTenantId());
+        event.setRemovalTime(historicIdentityLinkLog.getRemovalTime());
+        event.setProcessInstanceId(historicIdentityLinkLog.getRootProcessInstanceId());
+
+        return event;
+    }
+
     private void setVariableComplexValue(VariableUpdateEvent event, String serializerName, byte[] value) {
         if (StringUtils.contains(serializerName, "spin:") && value != null) {
             try {

camunda-kafka-polling-client/src/main/java/de/viadee/camunda/kafka/pollingclient/service/polling/rest/CamundaRestPollingServiceImpl.java

@@ -39,6 +39,7 @@ public class CamundaRestPollingServiceImpl implements PollingService {
     private static final String ACTIVITY_INSTANCE_ID = "activityInstanceId";
     private static final String PROCESS_DEFINITION_ID = "processDefinitionId";
     private static final String DEPLOYMENT_ID = "deploymentId";
+    private static final String TASK_ID = "taskId";
 
     private final ObjectMapper objectMapper;
 
@@ -336,10 +337,11 @@ public Iterable<ProcessDefinitionEvent> pollProcessDefinitions(final Date startT
     public Iterable<CommentEvent> pollComments(final ActivityInstanceEvent activityInstanceEvent) {
 
         final String url = camundaProperties.getUrl()
-                + "task/" + activityInstanceEvent.getTaskId() + "/comment";
+                + "task/{taskId}/comment";
         try {
             final Map<String, Object> variables = new HashMap<>();
-            LOGGER.debug("Polling comments for taskId: {}", activityInstanceEvent.getTaskId());
+            variables.put(TASK_ID, activityInstanceEvent.getTaskId());
+            LOGGER.debug("Polling comments from {} ({})", url, variables);
 
             List<GetCommentResponse> result = this.restTemplate
                                                                .exchange(url,
@@ -365,6 +367,44 @@ public Iterable<CommentEvent> pollComments(final ActivityInstanceEvent activityI
         }
     }
 
+    /**
+     * {@inheritDoc}
+     *
+     * @param activityInstanceEvent
+     */
+    @Override
+    public Iterable<IdentityLinkEvent> pollIdentityLinks(final ActivityInstanceEvent activityInstanceEvent) {
+
+        final String url = camundaProperties.getUrl()
+                + "history/identity-link-log/?taskId={taskId}";
+        try {
+            final Map<String, Object> variables = new HashMap<>();
+            variables.put(TASK_ID, activityInstanceEvent.getTaskId());
+            LOGGER.debug("Polling identity-links from {} ({})", url, variables);
+
+            List<GetIdentityLinkResponse> result = this.restTemplate
+                                                                    .exchange(url,
+                                                                              HttpMethod.GET,
+                                                                              null,
+                                                                              new ParameterizedTypeReference<List<GetIdentityLinkResponse>>() {
+
+                                                                              }, variables)
+                                                                    .getBody();
+
+            if (result == null) {
+                return new ArrayList<>();
+            }
+
+            LOGGER.debug("Found {} identity-links for taskId: {} ", result.size(), activityInstanceEvent.getTaskId());
+
+            return result
+                         .stream()
+                         .map(getIdentityLinkResponse -> createIdentityLinkEventFromDetails(getIdentityLinkResponse))::iterator;
+        } catch (RestClientException e) {
+            throw new RuntimeException("Error requesting Camunda REST API (" + url + ") for identity-link-log", e);
+        }
+    }
+
     private GetProcessDefinitionXmlResponse getProcessDefinitionXML(String processDefinitionId) {
         final String url = camundaProperties.getUrl()
                 + "process-definition/{processDefinitionId}/xml";
@@ -568,6 +608,26 @@ private CommentEvent createCommentEventFromDetails(
         return event;
     }
 
+    private IdentityLinkEvent createIdentityLinkEventFromDetails(GetIdentityLinkResponse identityLinkResponse) {
+        final IdentityLinkEvent event = new IdentityLinkEvent();
+
+        event.setId(identityLinkResponse.getId());
+        event.setTimestamp(identityLinkResponse.getTime());
+        event.setType(identityLinkResponse.getType());
+        event.setUserId(identityLinkResponse.getUserId());
+        event.setGroupId(identityLinkResponse.getGroupId());
+        event.setTaskId(identityLinkResponse.getTaskId());
+        event.setProcessDefinitionId(identityLinkResponse.getProcessDefinitionId());
+        event.setProcessDefinitionKey(identityLinkResponse.getProcessDefinitionKey());
+        event.setOperationType(IdentityLinkEvent.OperationType.valueOf(identityLinkResponse.getOperationType()));
+        event.setAssignerId(identityLinkResponse.getAssignerId());
+        event.setTenantId(identityLinkResponse.getTenantId());
+        event.setRemovalTime(identityLinkResponse.getRemovalTime());
+        event.setProcessInstanceId(identityLinkResponse.getRootProcessInstanceId());
+
+        return event;
+    }
+
     private void setVariableValue(VariableUpdateEvent event, Object value, String type,
                                   String serializationDataFormat) {