Generic Proxy Flow

The Generic Proxy flow enables flexible communication between two systems, offering both inbound and outbound capabilities to accommodate various use cases. In outbound mode, the flow can be utilized for webhook-like requirements, processing input data from an external system and returning the flow's output as the response. In inbound mode, the flow performs customizable tasks such as identifying special case orders and canceling them, or executing other system-specific operations based on the configured logic. With its adaptability, the Generic Proxy flow serves as a versatile tool for integrating and automating interactions between different systems.

Generic Proxy Flow Types​

• Inbound Flow: The Generic Proxy flow operates by executing custom logic in the Preprocessing Script step without any incoming data, enabling tasks like finding and processing special cases. During execution, the flow can make a single request to a configured "Remote URL" and process the response using custom logic in the Postprocessing Script step. This setup allows for dynamic and customizable operations, making it suitable for diverse inbound scenarios such as order cancellation or other system-specific tasks.

• Outbound Flow: The Generic Proxy flow operates by processing data sent by an external system as input to the Preprocessing Script step, where custom logic can be applied. The flow can also make a single request to a configured "Remote URL" during execution, enabling flexible interactions between systems. After processing, the Postprocessing Script step finalizes the logic and prepares the response, which is returned as the HTTP response to the external system. This design makes it ideal for webhook-like scenarios, allowing dynamic and customizable operations. More details can be found at the Outbound section.

Flow Steps​

• Login Step: Handles logging into the ERP system. If an error occurs during this step, the details are logged.

• Preprocessing Script Step: Executes custom logic to prepare or transform data for the flow. In inbound flows, this step starts with no incoming data, while in outbound flows, it uses data passed by the external system as input.

• Send Request Step: Sends a single HTTP request to the configured "Remote URL," enabling communication between systems. The request payload and headers can be customized based on the flow's requirements.

• Postprocessing Script Step: Processes the response from the "Remote URL" or the prepared data from the Preprocessing Script step, allowing further transformation, validation, or handling before finalizing the flow.

• HTTP Response Step (Only on Outbound): Returns the output of the flow as an HTTP response to the external system that initiated the request, providing the processed result or status.

Detailed Flow Designer Settings​

Configuration Card​

• Outbound Request URL (Outbound flows only): A read-only field that displays the URL used for the flow's outbound requests. This URL cannot be modified by the user and is defined by the flow's configuration.

• Remote URL: The target URL to which the flow will send requests. This field must be filled with the endpoint where the data exchange or interaction is intended to occur. This value can also be changed in run time using the flow attributes in preprocessing script step.

• HTTP Method: Specifies the HTTP method (e.g., GET, POST, PATCH, PUT) to be used for the request to the Remote URL. This allows the user to define the action to be performed, such as retrieving, creating, updating, or modifying data.

Preprocessing Script and Postprocessing Script​

This is a base script that can be updated to enable script functionality within the process step.

Scripts can read incoming data from the inputStream, the input stream will have the data sent by the external system that is triggering the flow on outbound flows, it will be empty for inbound GenericProxy flows. Any outgoing data is written using the outputStream.write() method.

Additionally, the script allows for the use of attributes, which provide supplementary information tied to the current execution process. These attributes can be freely accessed or modified throughout the script. For example, the get_attribute() function is used to read attribute values, while the session.putAttribute() method is used to write new string attribute values. Each attribute consists of a key-value pair, where the key uniquely identifies the attribute, and the value can be referenced in subsequent steps.

GenericProxy flow has some special attributes that are useful based on use cases:

• token.erp attribute: The token used to access ERP systems, provided only if a Login flow is specified in the Flow Settings of the current flow. Example: Bearer <token>. Even though the naming specifies “ERP” it doesn’t have to be a token belonging to ERP, the login setting specified in flow settings will be used for creating a token which can be accessed during script step.

• http.query.string attribute: This attribute can be used to access any query parameter sent by the triggering external system on outbound flows.

• remote.url attribute: This attribute can be used to overwrite the “Remote URL” setting during run time. It can be used to edit the request url to add or modify parameters, use a totally different request URL etc.

• remote.method attribute: This attribute can be used to overwrite the “HTTP Method” setting during run time. It must be a valid HTTP method that will be used while making a request to “remote url”.

• force_step attribute: The force_step attribute allows for conditional control over the execution of flow steps, enabling the flow to bypass or retry specific steps based on predefined conditions. This attribute can take different values, each influencing the flow behavior in a specific way:

  • “200response”: The attribute value "200response" can be set during the preprocessing script step to bypass the remote request and post processing script steps. This will result in a success (after the HTTP response step if outbound).

    For example, a generic proxy flow used for sending OMS package updates to an ERP system can use the attribute to skip packets without invoice numbers in the preprocessing script step.

  • “login”: If this value is set during any script step that results in a failure, the flow will redirect execution to the login step, ensuring a new authentication session is obtained. Once the login step completes, the execution resumes from the failed step. This mechanism helps recover from authentication failures, such as when a request fails due to an expired token in the token.erp attribute. However, this retry behavior is allowed only once—if login is forced a second time within the same execution, the flow will end with a failure.

Attributes whose keys begin with the prefix log. will be automatically logged at the end of the execution if a log file is generated, ensuring important information is recorded and can be reviewed later.

Example Script:

import json
import traceback
from java.nio.charset import StandardCharsets
from org.apache.commons.io import IOUtils
from org.apache.nifi.processor.io import StreamCallback

def get_attribute(flow_file, attr_name):
   all_var = flow_file.getAttribute("allVar")
   if all_var:
       all_attributes = json.loads(all_var)
       return all_attributes.get(attr_name, None)
   return flow_file.getAttribute(attr_name)

class TransformCallback(StreamCallback):
   def __init__(self, flowFile):
       self.flowFile = flowFile
       self.erpToken = get_attribute(flowFile, "token.erp")

   def process(self, inputStream, outputStream):
       input_text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
       input_obj = json.loads(input_text)

       # Transform content
       output_obj = self.transform(input_obj)

       # Write output content
       outputStream.write(bytearray(json.dumps(output_obj, indent=4).encode('utf-8')))
      
   def transform(self, output_obj):
       # Transform content
       return output_obj

flowFile = session.get()
if flowFile != None:
   try:
       flowFile = session.write(flowFile, TransformCallback(flowFile))
       # Finish by transferring the FlowFile to an output relationship
       session.transfer(flowFile, REL_SUCCESS)
   except:
       var = traceback.format_exc()
       session.putAttribute(flowFile, 'log.script.error', str(var))
       session.transfer(flowFile, REL_FAILURE)
   session.commit()

Script Testing: The ERP response is placed in the input field, and the script's result along with the written attributes can be viewed in the result field. When the test script is run, attributes like log.script.error will appear under the "attributes" section in the result panel and should be referenced if any exceptions occur during execution. The script's output content will be found under the result_data value.

The “Add Attributes” button can be used to simulate incoming, readable, attributes for the script. This feature is useful when an attribute value is read and used in the script. Note that attributes added through the button are only for testing purposes and will not be saved in the flow information—they will be removed when the page is refreshed.

For more information about script card step, please refer to the Script Document.

Outbound Flow​

In outbound flows, triggers will be configured to send a POST request to the URL specified in the "Outbound Request Url" setting under the Configuration card. The content of the POST request will be passed to the preprocessing script step which can be used to modify and change the data.

The authorization token will be obtained by logging in with the Integrator user and must be included in the request headers for authentication purposes.

Usage Examples​

Omnitron Order Cancel Release Stock Webhook​

This is a webhook event originating from Omnitron, it triggers when a stock reservation is released because of order cancellation before the order is delivered to the ERP system. As this is an event based on Omnitron, the event details will not be included in this example but instead a generic proxy flow dealing with the event information and trigger will be prepared.

The event sends the product and stock information with each trigger, in this example the related stock flow will be triggered for the event triggering product and stock list. As this flow will be triggered by an external system an outbound generic proxy flow will be used.

1. A place holder Remote URL is written, since this url will be changed for the related stock flow that we will trigger based on the event information during run time in the preprocessing script. HTTP Method is selected as “GET” since inbound Integrator stock flows are triggered with a GET request.

   
2. In the prescript step, the event information is received, which includes the product SKU and stock list information. Based on these values, the stock list is matched with the desired stock flow to trigger, and the “remote.url” is updated with the corresponding flow's trigger URL.

   A flow’s trigger URL closely resembles the generic proxy’s outbound request URL, such as:

   https://flow-dev.entegrator.akinon.net:3002/api/p1/akinon/stabletest/loggedgenericproxy.

   In this URL, “loggedgenericproxy” corresponds to the flow’s slug name. This slug name can be found on the log page of the related stock flow.

Example Event Request Data:

{
  "event": {
    "type": "product_stock.cancel.release_reservation",
    "uuid": "dee9c689671d4125b871bbd1b96cedb6"
  },
  "payload": {
    "product_sku": "sku123",
    "stock_list_name": "shop_stock_list",
    ...
  }
}

import json
import traceback
from java.nio.charset import StandardCharsets
from org.apache.commons.io import IOUtils
from org.apache.nifi.processor.io import StreamCallback


def get_attribute(flow_file, attr_name):
    all_var = flow_file.getAttribute("allVar")
    if all_var:
        all_attributes = json.loads(all_var)
        return all_attributes.get(attr_name, None)
    return flow_file.getAttribute(attr_name)


stock_list_flow_map = {
    "shop_stock_list": "https://flow-dev.entegrator.akinon.net:3002/api/p1/akinon/stabletest/stock", # flow slug: stock
}

class TransformCallback(StreamCallback):
    remote_url = None
    
    def __init__(self, flowFile):
        self.flowFile = flowFile

    def process(self, inputStream, outputStream):
        input_text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
        input_obj = json.loads(input_text)

    # Get stock list name and product sku from event payload.
    payload = input_obj.get("payload", {})
        product_sku = payload.get("product_sku")
        stock_list = payload.get("stock_list_name")
    
    # Try and find the related stock flow using stock list information and raise an error if no stock flow is found for given stock list
        self.remote_url = stock_list_flow_map.get(stock_list, None)
    if not self.remote_url:
        raise Exception("No matching stock flow found for stock list.")
        self.remote_url = self.remote_url + "?sku=" + product_sku

        # Write output content
        outputStream.write(bytearray("".encode("utf-8")))



flowFile = session.get()
if flowFile != None:
    try:
        transform = TransformCallback(flowFile)
        flowFile = session.write(flowFile, transform)
        # Set the new remote flow using flow attributes.
    session.putAttribute(flowFile, "remote.url", transform.remote_url)
        session.transfer(flowFile, REL_SUCCESS)
    except:
        var = traceback.format_exc()
        session.putAttribute(flowFile, "log.script.error", str(var))
        session.transfer(flowFile, REL_FAILURE)
    session.commit()

1. When the webhook runs or a request is made in any other way, the process will be visible on the flow logs. (A webhook like request is simulated in Postman for testing purposes)

These are the "Send Request" step logs of the GenericProxy. The logs show that a trigger was sent to the flow with the slug "stock" for the SKU 123123, which, as a result, returned the "execution_id" value for this trigger. By searching for this ID in the stock flow logs, you can view the stock flow logs related to that specific trigger.