Merge pull request #177 from camunda-community-hub/signal-event

feat: Add support for signal start events

src/main/kotlin/org/camunda/community/zeebe/play/rest/ProcessesResource.kt

@@ -2,17 +2,13 @@ package org.camunda.community.zeebe.play.rest
 
 import io.camunda.zeebe.client.ZeebeClient
 import io.camunda.zeebe.model.bpmn.Bpmn
-import io.camunda.zeebe.model.bpmn.instance.MessageEventDefinition
-import io.camunda.zeebe.model.bpmn.instance.Process
-import io.camunda.zeebe.model.bpmn.instance.StartEvent
-import io.camunda.zeebe.model.bpmn.instance.TimerEventDefinition
+import io.camunda.zeebe.model.bpmn.instance.*
+import io.zeebe.zeeqs.data.entity.ProcessInstanceState
 import io.zeebe.zeeqs.data.entity.TimerState
-import io.zeebe.zeeqs.data.repository.MessageCorrelationRepository
-import io.zeebe.zeeqs.data.repository.MessageSubscriptionRepository
-import io.zeebe.zeeqs.data.repository.ProcessRepository
-import io.zeebe.zeeqs.data.repository.TimerRepository
+import io.zeebe.zeeqs.data.repository.*
 import org.camunda.community.zeebe.play.connectors.ConnectorService
 import org.camunda.community.zeebe.play.services.ZeebeClockService
+import org.springframework.data.domain.PageRequest
 import org.springframework.data.repository.findByIdOrNull
 import org.springframework.web.bind.annotation.*
 import java.io.ByteArrayInputStream
@@ -31,6 +27,8 @@ class ProcessesResource(
     private val processRepository: ProcessRepository,
     private val messageSubscriptionRepository: MessageSubscriptionRepository,
     private val messageCorrelationRepository: MessageCorrelationRepository,
+    private val signalSubscriptionRepository: SignalSubscriptionRepository,
+    private val processInstanceRepository: ProcessInstanceRepository,
     private val timerRepository: TimerRepository,
     private val clockService: ZeebeClockService
 ) {
@@ -88,6 +86,8 @@ class ProcessesResource(
             return createProcessInstanceWithMessageStartEvent(processKey, variables)
         } else if (startEvent.eventDefinitions.any { it is TimerEventDefinition }) {
             return createProcessInstanceWithTimerStartEvent(processKey)
+        } else if (startEvent.eventDefinitions.any { it is SignalEventDefinition }) {
+            return createProcessInstanceWithSignalStartEvent(processKey, variables)
         } else {
             val type = startEvent.eventDefinitions.first().elementType.typeName
             throw RuntimeException("Can't start process instance with start event of type '$type'")
@@ -169,6 +169,55 @@ class ProcessesResource(
         return processInstanceKey
     }
 
+    private fun createProcessInstanceWithSignalStartEvent(
+        processKey: Long,
+        variables: String
+    ): Long {
+        val signalSubscription = signalSubscriptionRepository
+            .findByProcessDefinitionKey(processKey)
+            .firstOrNull()
+            ?: throw RuntimeException("No signal subscription found for process '$processKey'")
+
+        val signalKey = zeebeClient.newBroadcastSignalCommand()
+            .signalName(signalSubscription.signalName)
+            .variables(variables)
+            .send()
+            .join()
+            .key
+
+        return executor.submit(Callable {
+            getProcessInstanceKeyForSignal(
+                processKey = processKey,
+                signalKey = signalKey
+            )
+        }).get()
+    }
+
+    private fun getProcessInstanceKeyForSignal(processKey: Long, signalKey: Long): Long {
+        var processInstanceKey = -1L
+        while (processInstanceKey < 0) {
+            processInstanceKey =
+                processInstanceRepository.findByProcessDefinitionKeyAndStateIn(
+                    processDefinitionKey = processKey,
+                    stateIn = listOf(
+                        ProcessInstanceState.ACTIVATED,
+                        ProcessInstanceState.COMPLETED,
+                        ProcessInstanceState.TERMINATED
+                    ),
+                    pageable = PageRequest.of(0, 1000)
+                )
+                    // since the signal was broadcast first, the signal key should be higher
+                    .firstOrNull { it.key > signalKey }
+                    ?.key
+                    ?: run {
+                        // wait and retry
+                        Thread.sleep(RETRY_INTERVAL.toMillis())
+                        -1L
+                    }
+        }
+        return processInstanceKey
+    }
+
     @RequestMapping(
         path = ["/{processKey}/missing-connector-secrets"],
         method = [RequestMethod.GET]

src/main/kotlin/org/camunda/community/zeebe/play/rest/SignalResource.kt

@@ -0,0 +1,28 @@
+package org.camunda.community.zeebe.play.rest
+
+import io.camunda.zeebe.client.ZeebeClient
+import org.springframework.web.bind.annotation.RequestBody
+import org.springframework.web.bind.annotation.RequestMapping
+import org.springframework.web.bind.annotation.RequestMethod
+import org.springframework.web.bind.annotation.RestController
+
+@RestController
+@RequestMapping("/rest/signals")
+class SignalResource(private val zeebeClient: ZeebeClient) {
+
+    @RequestMapping(method = [RequestMethod.POST])
+    fun broadcastSignal(@RequestBody command: BroadcastSignalCommand): Long {
+        return zeebeClient.newBroadcastSignalCommand()
+            .signalName(command.signalName)
+            .variables(command.variables)
+            .send()
+            .join()
+            .key
+    }
+
+    data class BroadcastSignalCommand(
+        val signalName: String,
+        val variables: String?
+    )
+
+}

src/main/resources/public/js/rest-client.js

@@ -150,3 +150,10 @@ function sendEvaluateDecisionRequest(decisionKey, variables) {
 function sendGetDecisionInputsRequest(decisionKey) {
   return sendGetRequest(`decisions/${decisionKey}/inputs`);
 }
+
+function sendBroadcastSignalRequest(signalName, variables) {
+  return sendPostRequest("signals", {
+    signalName: signalName,
+    variables: variables,
+  });
+}

src/main/resources/public/js/view-bpmn.js

@@ -116,6 +116,28 @@ function removePublishMessageButton(elementId) {
   overlays.remove({ element: elementId, type: "publish-message" });
 }
 
+function addBroadcastSignalButton(elementId, clickAction) {
+  const buttonId = "broadcast-signal-diagram-action-" + elementId;
+  const content = `
+    <button id="${buttonId}" type="button" class="btn btn-sm btn-primary" title="Broadcast signal">   
+      <svg class="bi" width="18" height="18" fill="white"><use xlink:href="/img/bootstrap-icons.svg#triangle"/></svg>  
+    </button>`;
+
+  overlays.add(elementId, "broadcast-signal", {
+    position: {
+      top: -20,
+      left: -20,
+    },
+    html: content,
+  });
+
+  $("#" + buttonId).click(clickAction);
+}
+
+function removeAllBroadcastSignalButtons() {
+  overlays.remove({ type: "broadcast-signal" });
+}
+
 function highlightElement(elementId) {
   if (highlightedElementId && highlightedElementId !== elementId) {
     canvas.removeMarker(highlightedElementId, "bpmn-element-selected");
@@ -579,9 +601,11 @@ function toggleDetailsCollapse() {
 function zoomIn() {
   bpmnViewer.get("zoomScroll").stepZoom(0.1);
 }
+
 function zoomOut() {
   bpmnViewer.get("zoomScroll").stepZoom(-0.1);
 }
+
 function resetViewport() {
   const outerViewbox = canvas.viewbox().outer;
   canvas.viewbox({
@@ -591,6 +615,7 @@ function resetViewport() {
     height: outerViewbox.height,
   });
 }
+
 function enterFullscreen() {
   const button = document.querySelector("#toggleFullscreenButton");
 
@@ -607,6 +632,7 @@ function enterFullscreen() {
 
   document.documentElement.requestFullscreen();
 }
+
 function exitFullscreen() {
   const button = document.querySelector("#toggleFullscreenButton");
 

src/main/resources/public/js/view-common.js

@@ -949,3 +949,33 @@ if (resizeHandle) {
     });
   });
 }
+
+function showBroadcastSignalModal(signalName) {
+  $("#broadcast-signal-name").val(signalName);
+  $("#broadcast-signal-variables").val("");
+
+  $("#broadcast-signal-modal").modal("show");
+}
+
+function broadcastSignalFromModal() {
+  const signalName = $("#broadcast-signal-name").val();
+  const variables = $("#broadcast-signal-variables").val();
+
+  history.push({
+    action: "broadcastSignal",
+    signalName,
+    variables,
+  });
+  refreshHistory();
+
+  sendBroadcastSignalRequest(signalName, variables)
+    .done((signalKey) => {
+      const toastId = "signal-broadcasted-" + signalKey;
+      showNotificationSuccess(
+        toastId,
+        "New signal broadcasted",
+        "<code>" + signalName + "</code>"
+      );
+    })
+    .fail(showFailure("broadcast-signal-failed", "Failed to broadcast signal"));
+}

src/main/resources/public/js/view-process-instance.js

@@ -200,6 +200,11 @@ async function rewind(task) {
     ) {
       // timer start event
       newId = await createNewInstanceFromTimerStartEvent(startEvent);
+    } else if (
+      startEvent.eventDefinitions[0].$type === "bpmn:SignalEventDefinition"
+    ) {
+      // signal start event
+      newId = await createNewInstanceFromSignalStartEvent(startEvent);
     }
 
     track?.("zeebePlay:bpmnelement:completed", {
@@ -415,6 +420,20 @@ function waitForMessageSubscription(id, messageName, correlationKey) {
   });
 }
 
+async function createNewInstanceFromSignalStartEvent(startEvent) {
+  const signalName = startEvent.eventDefinitions[0].signalRef.name;
+
+  const numberOfCurrentInstances = await getNumberOfCurrentInstancesFor(
+    currentProcessKey
+  );
+  await sendBroadcastSignalRequest(signalName);
+
+  return await waitForNewInstanceFor(
+    currentProcessKey,
+    numberOfCurrentInstances
+  );
+}
+
 function fetchTimerForElement(id, elementId) {
   return new Promise((resolve, reject) => {
     let remainingTries = 6;

src/main/resources/public/js/view-process.js

@@ -47,6 +47,7 @@ function loadProcessView() {
 
   loadInstancesOfProcess(instancesOfProcessCurrentPage);
   loadMessageSubscriptionsOfProcess();
+  loadSignalSubscriptionsOfProcess();
   loadTimersOfProcess();
 }
 
@@ -337,6 +338,68 @@ function loadTimersOfProcess() {
   });
 }
 
+function loadSignalSubscriptionsOfProcess() {
+  const processKey = getProcessKey();
+
+  querySignalSubscriptionsByProcess(processKey).done(function (response) {
+    let process = response.data.process;
+
+    let signalSubscriptions = process.signalSubscriptions;
+    let totalCount = signalSubscriptions.length;
+
+    $("#signal-subscriptions-total-count").text(totalCount);
+
+    $("#signal-subscriptions-of-process-table tbody").empty();
+
+    removeAllBroadcastSignalButtons();
+
+    const indexOffset = 1;
+
+    signalSubscriptions.forEach((signalSubscription, index) => {
+      const isActive = signalSubscription.state === "CREATED";
+
+      const buttonId = "broadcast-signal-action-" + signalSubscription.key;
+
+      let action = "";
+      if (isActive) {
+        action = `
+          <button id="${buttonId}" type="button" class="btn btn-sm btn-primary" title="Broadcast signal">   
+            <svg class="bi" width="18" height="18" fill="white"><use xlink:href="/img/bootstrap-icons.svg#triangle"/></svg>  
+             Broadcast signal
+          </button>`;
+      }
+
+      $("#signal-subscriptions-of-process-table > tbody:last-child").append(`
+        <tr>
+          <td>${indexOffset + index}</td>
+          <td>${signalSubscription.key}</td>
+          <td>${signalSubscription.signalName}</td>
+          <td>${formatBpmnElementInstance(signalSubscription.element)}</td>
+          <td>${action}</td>
+        </tr>`);
+
+      if (isActive) {
+        $("#" + buttonId).click(function () {
+          showBroadcastSignalModal(signalSubscription.signalName);
+        });
+
+        addBroadcastSignalButton(
+          signalSubscription.element.elementId,
+          function () {
+            track?.("zeebePlay:bpmnelement:completed", {
+              element_type: "START_EVENT",
+              From: "processPage",
+              process_id: getProcessId(),
+            });
+
+            showBroadcastSignalModal(signalSubscription.signalName);
+          }
+        );
+      }
+    });
+  });
+}
+
 function loadProcessElementOverview() {
   const processKey = getProcessKey();
 

src/main/resources/public/js/zeeqs-client.js

@@ -72,6 +72,23 @@ const messageSubscriptionsByProcessQuery = `query MessageSubscriptionsOfProcess(
     }
   }`;
 
+const signalSubscriptionsByProcessQuery = `query SignalSubscriptionsOfProcess($key: ID!, $zoneId: String!) {  
+    process(key: $key) {
+    
+      signalSubscriptions {
+        key
+        signalName       
+        state
+        timestamp(zoneId: $zoneId)
+        element {
+          elementId
+          elementName
+          bpmnElementType
+        }        
+      }
+    }
+  }`;
+
 const timersByProcessQuery = `query TimersOfProcess($key: ID!, $zoneId: String!) {  
     process(key: $key) {
     
@@ -760,6 +777,13 @@ function queryMessageSubscriptionsByProcess(processKey) {
   });
 }
 
+function querySignalSubscriptionsByProcess(processKey) {
+  return fetchData(signalSubscriptionsByProcessQuery, {
+    key: processKey,
+    zoneId: getTimeZone(),
+  });
+}
+
 function queryTimersByProcess(processKey) {
   return fetchData(timersByProcessQuery, {
     key: processKey,