Start and Stop Flow At Runtime

Sometimes the requirement is to manage the flow state at runtime, say the service being called is down, and stop the flow to avoid the unnecessary processing. This can be achieved in Mule via two methods mainly:

1. Using Groovy Script
2. Using Anypoint Runtime Manager (ARM) Service (Only for On-Prem or Hybrid APIs)

Using Groovy Script

Using the Execute operation, we can write a Groovy script to manage the state of the flow. Let’s create a simple application where we have two flows, one with HTTP listener and another a scheduler based flow. The first flow will be used to start/stop the flow with the scheduler as a source. (Yes, schedulers can be stopped in Cloudhub apps separately instead of stopping the whole flow, it’s just an example to show how to stop the flow)

1. Configure the HTTP listener to listen and trigger the flow which will have the Groovy script.
2. Add the scripting module and put an Execute operation after the listener and add the below script.

flow = registry.lookupByName(‘schedulerFlow’).get();
if (flow.isStarted())
flow.stop()
else flow.start()

We are looking in the registry for a flow named ‘schedulerFlow’ and depending on the current state, changing the state to start or stop.

3. Create a new flow and add scheduler there, keep default configurations. Change the flow name to ‘schedulerFlow’ and add a logger.
4. Run the app; scheduler flow will start printing logs. Hit the HTTP listener endpoint configured in the first flow and observe the logs. Scheduler flow will be stopped.

Using Anypoint Runtime Manager Services

If the application is deployed in an On-Prem or Hybrid deployment model, the flow’s starting/stopping can be managed using ARM Rest Services. We can break down the flow in the below steps-

1. Get Cloudhub Access Token
2. Get Organization Info
3. Get all applications which are on the server and filter out the one we need.
4. Get the flows of our application.
5. Start/Stop the desired flow.

Deploy our demo application to the server and follow the steps given below

1. Configure the HTTP listener in the source section of the flow.
2. Access Management API will be used to get the access token to hit subsequent requests. To get the token, we need to send the Anypoint platform’s username and password. Set the payload to Anypoint credentials.

output application/JSON

{
    "username": "anypoint-username",
    "password": "anypoint-password"
}

3. Configure the HTTP Requester to hit https://anypoint.mulesoft.com/accounts/login and get the access token.

Save the response in a variable.

Access token will be passed in authorization header for all mule API calls.

Add HTTP requester to hit https://anypoint.mulesoft.com/accounts/api/profile to get the user profile which will contain organization and environment Id.

Save the response to the target variable; here it is  userProfile

5. Save environment Id in a variable, userProfile.organization.environments[?($.name == “Sandbox”)].id[0]. Application is deployed in the Sandbox environment hence using the same to filter out the environments from profile received.
6. Send a GET request to https://anypoint.mulesoft.com/hybrid/api/v1/applications to get a list of all applications on the server. Pass the Access token, organization Id and environment Id in the header.

Since the application name is known and we just need the application Id, save it in the target variable

7. To get the list of flows in the application, send a GET request to https://anypoint.mulesoft.com/hybrid/api/v1/applications/{app-id}/flows

8. Add a Transform Message to fetch the desired flow Id and set the new state for the flow depending on the previous state.

%dw 2.0
output application/json
fun getFlow() = payload.data[?($.name == "schedulerFlow")]
---
{
    id: getFlow()[0].id,
    newState: if(getFlow()[0].targets[0].lastReportedStatus == "STARTED") "STOPPED" else "STARTED"
}

9. Send a PATCH request to https://anypoint.mulesoft.com/hybrid/api/v1/applications/{app-id}/flows/{flow-id}

Add the required headers (Access token, org and env Id) before sending the request