Behaviour

Components receive incoming messages, process them and generate outgoing messages. The way messages are processed is called the component behaviour. It defines what components do internally and how they react on inputs.

Components are implemented as NodeJS modules that return an object with a set of methods (Component Virtual Methods) that the Appmixer engine understands. Let's start with a simple example, a SendSMS component that has one input port (message), no output ports and its purpose is to send an SMS using the Twilio API.

const twilio = require('twilio');

module.exports = {

    receive(context) {
        let { fromNumber } = context.properties;
        let { accountSID, authenticationToken } = context.auth;
        let message = context.messages.message.content;
        let client = new twilio(accountSID, authenticationToken);
        return client.message.create({
            body: message.body,
            to: message.to,
            from: fromNumber
        });
    }
};

Component Virtual Methods

As was mentioned in the previous paragraph, components are simple NodeJS modules that can implement a certain set of methods the Appmixer engine understands. The one most important method is the receive() method. This method is called by the engine every time messages are available on the input ports and the component is ready to fire (fire patterns match). The method must return a promise that when resolved, acknowledges the processing of the input messages. If the promise is rejected, the Appmixer engine automatically re-tries to send the messages again in the next turn (with some delay).

Messages that have been rejected 30-times are put in a special internal "dead-letter" queue and never returned to the flow for processing again. They can only be recovered manually by the Administrator.

For trigger-type of components, the most important virtual methods to remember is tick() and start().

Virtual Method	Description
receive(context)	Called whenever there are new messages on the input ports that the component is ready to consume. This method must return a promise that when resolved, tells the engine that the messages were successfully processed. When rejected, the engine re-tries to send the messages to the component again in the next turn (or with an arbitrary delay).
tick(context)	Called whenever the polling timer sends a tick. This method is usually used by trigger Components.
start(context)	Called when the engine signals the component to start (when the flow starts). This method is usually used by some trigger components that might schedule an internal timer to generate outgoing messages in regular intervals.
stop(context)	Called when the engine signals the component to stop (when the flow stops). This is the right place to do a graceful shutdown if necessary.

Context

All virtual methods have one argument, the context. The context contains all the information you need to process your messages and send new messages to the output ports.

Input/Output message(s)

context.messages

(applies to receive())

Incoming messages. An object with keys pointing to the input ports. Each message has a content property that contains the actual data of the message after all variables have been replaced. For example:

{
    receive(context) {
        const smsContent = context.messages.message.content;
    }
}

Remember, if before running the flow, the input port message was defined in the Inspector using variables:

Variables

where the flow descriptor would contain something like this:

Humidity: {{{$.ec8cd99f-0ad3-4bca-9efc-ebea5be6b596.weather.main.humidity}}}

the context.messages object contains the result of replacing variables with actual data that was sent through the output port of the connected component, i.e.

context.messages.message.content === 'Humidity: 75'

Each message also contains the correlation ID in the context.messages.myInputPort.correlationId property.

correlationId is a "session ID" that associates all the messages in one pass through the flow. Every time a trigger component sends a message to the flow (e.g. webhook, timer, ...) and the message does not have a correlation ID yet, the Appmixer engine assigns a new correlation ID to the message. This correlation ID is then copied to all the messages that were generated as a reaction to the original trigger message.

async context.sendJson(messageContent, outPort)

A method on the context object that you should call when you want to emit a message on one of the components output ports. The first argument can be any JSON object and the second argument is the name of an output port. The function returns a promise that has to be either returned from the receive() , tick() or start() methods or awaited.

Authentication

context.auth

The authentication object. It contains all the tokens you need to call your APIs. The authentication object contains properties that you defined in the auth object in your Authentication module (auth.js) for your service. For example, if our authentication module for our service (auth.js) looks like this:

const twilio = require('twilio');
module.exports = {
    type: 'apiKey',
    definition() {
        return {
            tokenType: 'authentication-token',
            accountNameFromProfileInfo: 'accountSID',
            auth: {
                accountSID: {
                    type: 'text',
                    name: 'Account SID',
                    tooltip: 'Log into your Twilio account and find <i>API Credentials</i> on your settings page.'
                },
                authenticationToken: {
                    type: 'text',
                    name: 'Authentication Token',
                    tooltip: 'Found directly next to your Account SID.'
                }
            },
            validate: context => {
                let client = new twilio(context.accountSID, context.authenticationToken);
                return client.api.accounts.list();
            }
        };
    }
};

we can use the context.auth.accountSID and context.auth.authenticationToken in the component virtual methods:

{
    receive(context) {
        let { accountSID, authenticationToken } = context.auth;
    }
}

Properties

context.properties

The configuration properties of the component. This corresponds to the properties object from the component manifest file. For example, if our component defines the following properties in the manifest file:

{
    "properties": {
        "schema": {
            "properties": {
                "fromNumber": { "type": "string" }
            }
        },
        "inspector": {
            ...
        }
    }
}

context.properties.fromNumber will contain the value the user entered in the Designer UI Inspector:

{
    receive(context) {
        const fromNumber = context.properties.fromNumber;
    }
}

State

context.state

A persistent state of the component. Sometimes you need to store data for the component that must be available across multiple receive() calls for the same component instance. If you also need the data to be persistent when the flow is stopped and restarted again, set the state: { peristent: true } property in your component manifest. context.state is a simple object with keys mapped to values that is persistently stored in DB. This object is loaded on demand in each receive() call. It is not recommended to store large amounts of data here. Example:

{
    async receive(context) {
        // Emit a message only once per day for this component instance.
        const day = (new Date).getDay();
        const state = context.state;
        if (!state[day]) {
            state[day] = true;
            await context.saveState(state);
            return context.sendJson({ tick: true }, 'out');
        }
    }
}

The context.state is especially useful for trigger-type of components when polling an API for changes to store the ID of the latest processed item from the API.

The context.state object should not be used to store large amounts of data. The state is loaded with each received message on a component input port. The maximum limit is 16MB but storing such large objects will slow down processing of the component input messages.

async context.saveState(object)

Save an updated state object. See context.state for details. The function returns a promise that resolves if storing of the state was successful and rejects otherwise.

async context.stateSet(key, value)

Set a state key to hold the value. key must be a string. value can be any JSON object.

async context.stateGet(key)

Get a state value stored under key.

async context.stateUnset(key)

Remove a value under key.

async context.stateClear()

Clears the entire state.

Files

async context.saveFile(fileName, mimeType, buffer)

Save a file to the Appmixer file storage. This function returns a promise that when resolved gives you an UUID that identifies the stored file (this is different from the file Mongo ID). You can pass this ID through your flow (send it to an output port of your component) so that later components can load the file from the Appmixer storage using the file ID.

{
    receive(context) {
        // getAttachment() is some function that retrieves a file from an API
        return getAttachment(context.auth, context.messages.attachment.content.id)
            .then((file) => {
                return context.saveFile(file.name, file.mimeType, Buffer.from(file.data, 'base64'));
            })
            .then((result) => {
                return context.sendJson({ fileId: result.fileId }, 'file');
            });
    }
}

async context.saveFileStream(fileName, stream)

Save a file to the Appmixer file storage. The function returns a promise that resolves with ID of the stored file. This is a more efficient and recommended version of context.saveFile(name, mimeType, buffer).

async context.getFileInfo(fileId)

Returns a promise, which when resolved returns the file information (name, length, content type...). For backward compatibility, fileId can be either Mongo ID or UUID.

Example:

{
  "filename": "testFile",
  "contentType": "text",
  "length": 7,
  "chunkSize": 261120,
  "uploadDate": "2021-01-22T12:20:29.227Z",
  "metadata": {
    "userId": "5f804b96ea48ec47a8c444a7",
    "fileId": "fd0e9149-3249-4d42-b519-bfd9ab6773c5"
  },
  "md5": "9a0364b9e99bb480dd25e1f0284c8555",
  "fileId": "fd0e9149-3249-4d42-b519-bfd9ab6773c5"
}

async context.loadFile(fileId)

Load a file from the Appmixer file storage. For backward compatibility, fileId can be either Mongo ID or UUID. The function returns a promise that when resolved, gives you the file data as a Buffer.

{
    receive(context) {
        return context.loadFile(context.messages.file.content.fileId)
            .then((fileContent) => {
                // uploadFileToAPI() is some function that uploads a file to an API
                return uploadFileToAPI(context.auth, content.messages.file.content.fileName, fileContent);
            });
    }
}

context.readFileStream(fileId)

This method has been deprecated. Use context.getFileReadStream instead. Read a file stream from the Appmixer file storage. The function returns a NodeJS read stream that you can e.g. pipe to other, write streams (usually to a request object when uploading a file to a 3rd party API). This is a more efficient and recommended version of context.loadFile(fileId). fileId must be Mongo ID.

async context.getFileReadStream(fileId)

Read a file stream from the Appmixer file storage. The function returns promise, which when resolved, returns a NodeJS read stream that you can e.g. pipe to other, write streams (usually to a request object when uploading a file to a 3rd party API). This is a more efficient and recommended version of context.loadFile(fileId). This method is backward compatible, sofileId can be either be Mongo ID or UUID.

async context.removeFile(fileId)

Remove a file from the Appmixer file storage. The function returns a promise. fileId can be either Mongo ID or UUID.

Webhook

context.getWebhookUrl()

Get a URL that you can send data to with HTTP POST or GET. When the webhook URL is called, the receive() method of your component is called by the engine with context.messages.webhook object set and context.messages.webhook.content.data contains the actual data sent to the webhook URL:

module.exports = {
    async receive(context) {
        if (context.messages.webhook) {
            // Webhook URL received data.
            await context.sendJson(context.messages.webhook.content.data, 'myOutPort');
            // Send response to the webhook HTTP call.
            // Note: you can also skip sending response immediately and send it
            // in other connected components in the flow.
            // If context.response() is not called, the engine waits for the first component
            // that sends the response (in the same "session", i.e. the same "message flow").
            return context.response('<myresponse></myresponse>', 200, { 'Content-Type': 'text/xml' });
        }
        // Otherwise, normal input port received data.
        const input = context.messages.myInPort.content;

        // The webhook URL. Do something with it (send to your API, send to other connected,
        // components, send to your backend, ...)
        const url = context.getWebhookUrl();
    }
};

Note: The context.getWebhookUrl() is only available if you set webhook: true in your component manifest file (component.json). This tells the engine that this is a "webhook"-type of component.

The full context.messages.webhook object contains the following properties:

Property	Description
content.method	HTTP method of the request.
content.hostname	Hostname of the Appmixer API.
content.headers	HTTP headers of the request.
content.query	Object with query parameters, i.e. query string parsed into a JSON object.
content.data	Object with the body parameters of the request.
correlationId	A special ID generated by the Appmixer engine that uniquely identifies the input message which resulted in generating the webhook URL. In other words, if you call context.getWebhookUrl() in the receive() method in a reaction on an input message that arrived on an input port of the webhook component, the correlationId will be part of the returned webhook URL. This allows you to later associate the input message with the HTTP call to the webhook. A common pattern is to store the input message in the context.state object and later use the context.messages.webhook.correlationId to retrieve it back. For example, if you have an input port named myInPort, you can get the correlationId of the input message that just arrived by accessing the context.messages.myInPort.correlationId.

async context.response(body, statusCode, headers)

Send response to the webhook HTTP call. When you set your component to be a webhook-type of component (webhook: true in your component.json file), context.getWebhookURL() becomes available to you inside your component virtual methods. You can use this URL to send HTTP POST requests to. When a request is received by the component, the context.messages.webhook.content.data contains the body of your HTTP POST call. In order to send a response to this HTTP call, you can use the context.response() method. See context.getWebhookUrl() for details and example.

Miscellaneous

async context.setTimeout(messageContent, delay)

Set a timer which causes the component to receive messageContent in the receive() method in the special context.messages.timeout.content object. delay is the time, in milliseconds, the timer should wait before sending the messageContent to the component itself. Note that this is especially useful for any kind of scheduling components. Note that the context.setTimeout() function works in a cluster environment as opposed to using the global setTimeout() JavaScript function. For example, a component that just "slows down" incoming messages before sending them to its output port, waiting e.g. 5 minutes, can look like this:

module.exports = {
    receive(context) {
        if (context.messages.timeout) {
            // Timeout message.
            return context.sendJson(context.messages.timeout.content, 'out');
        } else {
            // Normal input message.
            return context.setTimeout(context.messages.in.content, 5 * 60 * 1000);
        }
    }
};

Note that you can also access the correlation ID of the timeout message which can be useful in some scenarios. The correlation ID is available in the context.messages.timeout.correlationId property.

async context.callAppmixer(request)

Call an Appmixer REST API endpoint. You can call any of the Appmixer endpoints defined in the API section. The main advantage of this method (as opposed to calling the API endpoint manually) is that the method automatically populates the "Authorization" header of the request to the access token of the user who owns the flow this component lives in. For example:

const task = await context.callAppmixer({
    endPoint: '/people-task/tasks',
    method: 'POST',
    body: {
        title: 'My Task',
        description: 'My Example Task',
        requester: '[e-mail]',
        approver: '[e-mail]',
        decisionBy: (function() {
            const tomorrow = new Date;
            tomorrow.setDate(tomorrow.getDate() + 1);
            return tomorrow.toISOString();
        })()
    }
});

async context.stopFlow()

Stop the running flow. Example:

module.exports = {
    async receive(context) {
        return context.stopFlow();
    }
};

context.componentId

The ID of the component.

context.flowId

The ID of the flow the component lives in.

context.flowDescriptor

The flow descriptor of the flow in which the component instance lives. This allows you to access configuration of the entire flow within your component virtual methods. To get the configuration of the component itself, you can use context.flowDescriptor[context.componentId]. Note that this is normally not necessary since you can access the properties of the components by context.properties and the current input message by context.messages.myInPort.content but can be useful in some advanced scenarios.

async context.loadVariables()

Allows you to load variables in your component definition. Variables are data available from components connected back in the chain. loadVariables() returns a promise that resolves to an array that looks like this:

[
  {
    "variables": {
      "dynamic": [
        {
          "label": "Column A",
          "value": "columnA",
          "componentId": "2c724dff-a04b-43ed-9787-c9a891a721cc",
          "port": "out"
        },
        {
          "label": "Column B",
          "value": "columnB",
          "componentId": "2c724dff-a04b-43ed-9787-c9a891a721cc",
          "port": "out"
        }
      ],
      "static": {}
    },
    "sourceComponentId": "2c724dff-a04b-43ed-9787-c9a891a721cc",
    "outPort": "out",
    "inPort": "in"
  },
  {
    "variables": {
      "dynamic": [
        {
          "label": "Column C",
          "value": "columnC",
          "componentId": "71eb1f40-ce29-4963-8922-102739bafee4",
          "port": "out"
        },
        {
          "label": "Column D",
          "value": "columnD",
          "componentId": "71eb1f40-ce29-4963-8922-102739bafee4",
          "port": "out"
        }
      ],
      "static": {}
    },
    "sourceComponentId": "71eb1f40-ce29-4963-8922-102739bafee4",
    "outPort": "out",
    "inPort": "in"
  }
]

The return array has as many items as there are other components connected to this component.

Example:

{
    receive(context) {
        return context.loadVariables()
            .then(data => {
                const newSchema = data.reduce((acc, item) => {
                    return acc.concat(item.variables.dynamic.map(o => ({ label: o.label, value: o. value })));
                }, []);
                context.sendJson(newSchema, 'leftJoin');
                context.sendJson(newSchema, 'innerJoin');
                context.sendJson(newSchema, 'rightJoin');
            });
    }
}

async context.log(object)

Log message into InsightsLogs. The argument has to be an object.

Example:

{
    async start(context) {
        
        await context.log({ test: 'my test log' });
        return context.sendJson({ started: (new Date()).toISOString() }, 'out');
    }
}

And the object can be seen in logs (and InsightsLogs as well):

async context.lock(lockName, options)

This method allows component to create lock. This is useful when creating mutually exclusive section inside component's code. Such thing can be achieved in Appmixer using either quotas (you can define quota the way that only one receive call can be executed at a time) or using this lock. This method returns lock instance. Don't forget to call lock.unlock() when you're done. Otherwise the lock will be released after TTL.

lockName string will be prefixed with vendor.service:. If component type is appmixer.google.gmail.NewEmail, then lockName will be prefixed with appmixer.google:. This allows you to create lock that is shared among all component within a service and prevents possible collisions between components from different vendors or services.

The first parameter is required, the second (options) is optional with the following optional properties:

• ttl, number, 20000 by default (ms)

• retryDelay, number, 200 by default (ms)

• maxRetryCount, number, 30 by default

Example:

    async receive(context) {

        let lock = null;
        try {
            lock = await context.lock(context.flowId, {
                ttl: 30000,
                maxRetryCount: 1
            });
            let { callCount = 0, messages = [] } = await context.loadState();
            messages.push(context.messages.in.originalContent);

            if (++callCount === context.messages.in.content.callCount) {
                await context.sendJson(messages || [], 'out');
                await context.saveState({ callCount });
                return;
            }

            await context.saveState({ callCount, messages });
        } finally {
            if (lock) {
                await lock.unlock();
            }
        }
    }

Error Handling

Every function a component implements may thrown an exception (or return rejected promise).

receive(context)

If this function throws an exception, then the Appmixer engine will try to process the message that triggered this receive call again in a minute. There is an exponential backoff strategy, so the next attempt will happen in a minute and a half and so on. In total, Appmixer will try to process that message 30 times before it is saved into unprocessedMessages collection. Every unsuccessful attempt will be logged and visible in Insights.

Sometimes you, as a developer of a component, know that there is no point in retrying a message. It would fail again and again, 30 times. In such case you can tell Appmixer to cancel the message.

/**
 * Example of context.CancelError
 */
module.exports = {

    async receive(context) {

        let data = context.messages.in.content;

        try {
            // In this component is trying to create a record in a 3rd party
            // system.
            const resp = await someAPI.createSomething(data);
            await context.sendJson(resp, 'out');
        } catch (err) {
            // And there might be a unique constraint, let's say an email
            // address. And the 3rd party API will return an error with a
            // message saying that this record cannot be created.
            if (err.message === 'duplicate record') {
                // In this case, we can tell Appmixer to cancel the message.
                // Because next attempt would fail again with the same result.
                throw new context.CancelError(err.message);
            }
            // In case of any other error, rethrow the exception. Appmixer will
            // then try to process it again.
            throw err;
        }
    }
};

tick(context)

If a tick function throws an exception, such exception is logged (and visible in Insights) and that is it. Appmixer will trigger this function again in the future.

start(context)

Appmixer won't start a flow if any component in the flow throws an exception in the start function. Such error will be logged and visible in Insights.

stop(context)

Appmixer will stop the flow even when a component in the flow throws an exception in the stop function. Such error will be logged and visible in Insights.