Hi I have a BPMN flow and it has to be full automated, meaning no User Task at all. after a start, i am having a service task, and inside that I am creating process variables, post the service task, I am having a gateway, where I am trying to access a process variable, wherein i am checking if the process variable is null or has some value. When i start the process, I am getting an error where it says, “timeStamp is undefined”
Hi @satyaram413,
Please set asyncAfter attribute of "MQTT Task” service task to true. as below
camunda:asyncAfter="true"
For more information, please have a look at below post
Hi @hassang,
Well that worked, But now I am facing the following issue:
[MQTT Call: 83f3d665210b4c788227a9a8f3bb2d05] org.camunda.commons.logging.BaseLogger.logError ENGINE-16004 Exception while closing command context: execution a47666b0-8232-11ea-a903-025041000001 doesn't exist: execution is null
My service task is a mqtt task, which means, it actually subscribes to data from mqtt simulator, every 10 seconds, after it gets the data, i am saving the variables as process variable. The process variables were being used inside Exclusive Gateway, as i have mentioned before. Can you please let me know why I am facing this issue.
HI @hassang,
I tried async:before the control is waiting at my service task, but the process variable which I was creating inside MQTT service task, ar enot getting created now.
Hi @hassang,
I set asyncAfter=true, but the task isnt moving forward to script task. I have kept console.log in both of my script task, but nothing gets printed.
These are the primary methods that are important and are getting called, from service class of my MQTT task.
MqttSubscriberServiceTask.class
public void execute(DelegateExecution execution) throws Exception {
try {
RuntimeService runtimeService=execution.getProcessEngine().getRuntimeService();
MqttSubscriberServiceTask mqtt_camunda_service_task = new MqttSubscriberServiceTask();
// BPMN Logic
// String bpmnurl = mqtt_camunda_service_task.getURL(execution.getProcessDefinitionId(),
// execution.getCurrentActivityId());
org.camunda.bpm.model.bpmn.impl.instance.ServiceTaskImpl startEvent = (org.camunda.bpm.model.bpmn.impl.instance.ServiceTaskImpl) execution.getBpmnModelElementInstance();
ExtensionElements extensionElements = startEvent.getExtensionElements();
Collection<CamundaProperty> properties = extensionElements.getElementsQuery()
.filterByType(CamundaProperties.class)
.singleResult()
.getCamundaProperties();
String dmnName = null;
String mqttVariableName = null;
String mqttBrokerEndPoint = null;
for (CamundaProperty property : properties) {
String name = property.getCamundaName();
String value = property.getCamundaValue();
switch (name) {
case DMNNAME:
dmnName = value;
break;
case MQTTVARIABLENAME:
mqttVariableName = value;
break;
case MQTTBROKERENDPOINT:
mqttBrokerEndPoint = value;
break;
}
}
// MQTT Logic
// setting the process variables with the received Message
String mqttMessage = mqtt_camunda_service_task.subscribeWithBrokers(mqttBrokerEndPoint, execution, mqttVariableName, runtimeService);
if (mqttMessage == "-1") {
throw new RuntimeException(
"Error: MQTT Broker Connection failed - Invalid Broker IP Address :: Contact Administrator.");
} else if (mqttMessage == "-2") {
throw new RuntimeException("Error: Exception occured in Thread Processing :: Contact Administrator.");
}
} catch (Exception e) {
e.printStackTrace();
}
}
public String subscribeWithBrokers(String mqttBrokerEndPoint, DelegateExecution execution, String mqttVariableName, RuntimeService runtimeService) {
try {
// 1.Subscribe to the broker upon activation in a new Thread
// logger.info("Inside MqttSubscriberServiceTask :: Subscribing to the broker
// upon activation in a new Thread");
MqttSubscriber subscriber = new MqttSubscriber(syncObj, receivedMsg, mqttVariableName, mqttBrokerEndPoint, execution, runtimeService);
subscriber.subscribe();
logger.info(
"Inside MqttSubscriberServiceTask :: Block Service Task from completion by blocking the subscribed Thread");
// 2.Block Service Task from completion by blocking the Thread
logger.info("Inside MqttSubscriberServiceTask :: Parent thread waiting...");
logger.info("Inside MqttSubscriberServiceTask :: Parent thread exiting...");
logger.info("Inside MqttSubscriberServiceTask :: Parent unsubscribed");
// logger.info("Final mqttMessage received in subscribeWithBrokers:: " +
// receivedMsg.getMessage());
logger.info("Final Execution Variable :: " + execution.getVariables());
logger.info("Final Execution VariableNames :: " + execution.getVariableNames());
} catch (Exception e) {
logger.error("Exception occured in subscribeWithBrokers:: " + e.getMessage());
return "-2";
}
return receivedMsg.toString();
}
MqttSubscriber.class
public void subscribe() {
logger.info("Background Thread doing something big... in run ::");
logger.info("Runtime Service Accessing " +this.runtimeService);
try {
MqttConnectOptions connOpt;
// Subscription with mqttBrokerEndPoint
connOpt = new MqttConnectOptions();
connOpt.setAutomaticReconnect(true);
connOpt.setCleanSession(false);//if your not setting cleanSession to false then subscriptions shouldn't be persisted.
connOpt.setUserName(USERNAME);
connOpt.setPassword(PASSWORD.toCharArray());
String clientID = UUID.randomUUID().toString().replace("-", "");
logger.info("Generated clientID ::" + clientID);//Always subscriber's clientID should be different from the publishers clientID. If not we will get socket errors
myClient = new MqttClient(mqttBrokerEndPoint, clientID);
myClient.connect(connOpt);
myClient.setCallback(this);
myClient.subscribe(TOPIC);
} catch (MqttException e) {
logger.error(e.getMessage());
throw new RuntimeException("Error: MQTT Broker Connection failed - Invalid Broker IP Address :: Contact Administrator.");
}
}
@Override
public void messageArrived(String topic, org.eclipse.paho.client.mqttv3.MqttMessage message) throws Exception {
OffsetDateTime now = OffsetDateTime.now();
JSONObject js = new JSONObject();
js.put(MaintenanceVariable.VAR_VALUE.varName(), message.toString());
js.put(MaintenanceVariable.VAR_TYPE.varName(), MaintenanceVariable.VAR_STRING.varName());
JSONObject js1 = new JSONObject();
js1.put(MaintenanceVariable.VAR_VALUE.varName(), now.toString());
js1.put(MaintenanceVariable.VAR_TYPE.varName(), MaintenanceVariable.VAR_STRING.varName());
this.runtimeService.setVariable(this.delegateExecution.getId(), "MQTT", js.toJSONString());
System.out.println(js1.toJSONString());
System.out.println("Get Process Engine Services 123" +this.runtimeService );
this.runtimeService.setVariable(this.delegateExecution.getId(), "timeStamp", js1.toJSONString());
myClient.unsubscribe(TOPIC);
}Hi @satyaram413,
I think in your case, you should have a process model without MQTT task being part of it.
And MQTT call should be in a standalone Java class from where a new process instance of that model is started everytime output is returned. “every 10 seconds”