Zenaton | How it Works

Zenaton Functions

The following functions are included in the Zenaton SDK. They can be combined within your programming language to manipulate data, and create an infinite amount of possibilities to build distributed workflows at scale.

Defining a distributed workflow

With Zenaton, building and running a distributed workflow is as simple as writing a few lines of code describing it. Behind the scenes, Zenaton manages state, checkpoints, automatic retry (in case of failure), and provides you all the tooling and infrastructure to monitor and debug your distributed executions.

module.exports.handle = function* (...input) {
  // describe your workflow steps here
  // using the functions below
}

<?php
use Zenaton\Interfaces\WorkflowInterface;
use Zenaton\Traits\Zenatonable;

class WelcomeFlow implements WorkflowInterface {
  use Zenatonable;
  public function handle() {
    // describe your workflow steps here
    // using the functions below
  }
}

require 'zenaton'
                
class WelcomeFlow < Zenaton::Interfaces::Workflow
  include Zenaton::Traits::Zenatonable
  def handle
    # describe your workflow steps here
    # using the functions below
  end
end

from zenaton.abstracts.workflow import Workflow
from zenaton.traits.zenatonable import Zenatonable

class WelcomeFlow(Workflow, Zenatonable):
  def handle(self):
    # describe your workflow steps here
    # using the functions below

Dispatching a task, and waiting for the output

Use this syntax if you need the output of a task before continuing your workflow. Keep in mind, that the task will be distributed to one of your workers to be processed. The inputs and the output will be serialized and deserialized to be used here.

const output = yield this.run.task("TaskB", ...input);

$output = (new TaskA(...input))->execute();

output = TaskA.new(input).execute

output = TaskA(input).execute()

Dispatching a task, without waiting for the output

Use this syntax if you do NOT need the output of a task before continuing your workflow. The task will be distributed to one of your workers to be processed, while the workflow continues execute.

this.run.task("TaskB", ...input);

(new TaskA(...input))->dispatch();

TaskA.new(input).dispatch

TaskA(input).dispatch()

Waiting for the outputs of multiple tasks

Use this syntax if you need the all the outputs of multiple tasks before continuing the workflow. The tasks will be distributed to your workers to be processed in parallel. The workflow will resume only when ALL tasks have been successfully completed.

[outputA, outputB] = yield this.run.task(
  ["TaskA", ...inputA],
  ["TaskB", ...inputB]
);

[outputA, outputB] = (new Zenaton\Parallel\Parallel(
  new TaskA(...inputA),
  new TaskB(...inputB)
))->execute();

outputA, outputB = Zenaton::Parallel.new(
  TaskA.new(inputA),
  TaskB.new(inputB),
).execute

outputA, outputB = Zenaton.parallel.Parallel(
  TaskA(inputA),
  TaskB(inputB)
).execute()

Waiting for a duration or a date

Use the wait function to suspend the execution of the workflow until a date or for specific duration. The wait is managed by Zenaton and there is no activity on your workers while waiting. Once the waiting time elapsed, your servers will be notified to resume the workflow.

yield this.wait.for(duration);
yield this.wait.until(timestamp);

(new Wait())->hours(2)->execute();
(new Wait())->monday()->at("8:00")->execute();

Zenaton::Tasks::Wait.new.hours(2).execute
Zenaton::Tasks::Wait.new.monday.at('8:00').execute

Zenaton.tasks.wait.Wait().hours(2).execute()
Zenaton.tasks.wait.Wait().monday().at('8:00').execute()

Waiting for an external event

Use this syntax to suspend the execution of the workflow until it receives a specific event. This can be useful in multiple situations, such as waiting for human validation, order cancelation, delivery updates, etc. If needed, you can also add a timeout to decide what to do if this event has not been received before a certain time. The Zenaton Library provides an easy way (see below) to send an event to a workflow - from anywhere.

yield this.wait.event(event).forever();
yield this.wait.event(event).for(5 * 24 * 3600);

$event = (new Wait($eventName))->days(5)->execute();

event = Wait.new(eventName).days(5).execute

event = Wait.new(eventName).days(5).execute

Reacting to external events

This syntax allows you to react to an external event by describing what should happen each time you receive an event. Note that that as both handle and onEvent functions are sharing a common state, receiving events can influence main handle execution.

module.exports = { 
  *handle (...input) {
    // describe your workflow steps here
    // using the functions described here
  },
  *onEvent(name, ...data) {
    // describe what to do
    // when receiving an external event
  }
});

public function onEvent(EventInterface $event)
{
  // describe what to do here
}

def on_event(event)
  # describe what to do here
end

def on_event(self, event):
  # describe what to do here

HTTP calls as tasks

Use the "http" connector to request a server as if it were a task within your workflow. For example, this can be very useful to set up a workflow orchestrating microservices or external APIs.

const http = this.connector("http");

const output = yield http.get("https://google.com");

Easily use 3rd Party APIs

Use the connector function when you want to call an API using a Zenaton API Connector. You first setup your credentials in your dashboard. Continue to reuse the connection to call this API without having to manage the authentication anymore.

const gmail = this.connector("gmail", "CONNECTOR_ID");
yield gmail.get("users/me/messages");

How To Interact With Workflows

The following functions are included in the Zenaton SDK. Most of the time they are used in the code base of your application to dispatch workflows or interact with them.

Set Up a Client

By initializing a Zenaton client, you ensure that you have all the authorization needed to run within a specific domain defined by an Application Id and an Application Environment. An environment - defined by any string (eg. "dev", "staging", "production") - isolates your workflows.

const { Client } = require ("zenaton");
                  
const client = new Client(appId, apiToken, AppEnv);

Zenaton\Client::init($appId, $apiToken, $appEnv);

require 'zenaton'

Zenaton::Client.init(app_id, api_token, app_env)

from zenaton.client import Client

Client(app_id, api_token, app_env)

Dispatching workflows

Use this syntax to launch a workflow on your workers.

client.run.workflow(name, ...input);

(new MyWorkflow(...input))->execute();

MyWorkflow.new(input).execute

MyWorkflow(input).execute()

Dispatching tasks

Use this syntax to process an individual task on one of your workers.

client.run.task(name, ...input);

(new MyTask(...input))->execute();

MyTask.new(input).execute

MyTask(input).execute()

Sending Event to a workflow

The send syntax is used to send an event to a running workflow.
First we select a specific workflow, by name and by tag, then we provide the event. An event is simply defined by a name and some data.

client.select.workflow(name).withTag(tag).send(eventName, ...eventData);

WelcomeFlow::whereId($email)->send(new AddressUpdatedEvent('One Infinite Loop Cupertino, CA 95014'));

WelcomeFlow.where_id(email).send_event(AddressUpdatedEvent.new('One Infinite Loop Cupertino, CA 95014'))

WelcomeFlow().where_id(email).send_event(AddressUpdatedEvent('One Infinite Loop Cupertino, CA 95014'))

Scheduling tasks or workflows

Use the schedule function to set up recurrent execution of tasks and workflows using cron expressions.

client.schedule('* * * * * *').workflow(name, ...input);
client.schedule('* * * * * *').task(name, ...input);

(new RecurrentWorkflow(...input))->schedule('* * * * * *');
(new RecurrentTask(...input))->schedule('* * * * * *');

RecurrentWorkflow.new(input).schedule('* * * * * *')
RecurrentTask.new(input).schedule('* * * * * *')

RecurrentWorkflow(input).schedule('* * * * * *')
RecurrentTask(input).schedule('* * * * * *')

Pause, resume or terminate workflows

Use these functions to pause and resume or even terminate a workflow. For example, for circumstances when a workflow is no longer relevant - such as a user cancellation.

client.select.workflow(name).withTag(tag).pause();
client.select.workflow(name).withTag(tag).resume();
client.select.workflow(name).withTag(tag).terminate();

WelcomeFlow::whereId($email)->pause();
WelcomeFlow::whereId($email)->resume();
WelcomeFlow::whereId($email)->kill();

WelcomeFlow.where_id(email).pause
WelcomeFlow.where_id(email).resume
WelcomeFlow.where_id(email).kill

WelcomeFlow().where_id(email).pause()
WelcomeFlow().where_id(email).resume()
WelcomeFlow().where_id(email).kill()

Using code versatility

Building workflows in code lets you use the native versatility of yourpreferred programming language. Use and combine controlstructure (if, else, for each, while, ...) as well as whatever data structureand functions (map, filter, array...).

// SequentialWorkflow
module.exports.handle = function*() {
  // Run the TaskA
  const a = yield this.run.task("TaskA");

  // then, depending on the result of TaskA, run TaskB or TaskC
  if (a === 0) {
    yield this.run.task("TaskB");
  } else {
    yield this.run.task("TaskC");
  }

  // Finally, run the TaskD
  yield this.run.task("TaskD");
};

//ParallelWorkflow
module.exports.handle = function*() {
  // Run both TaskA and TaskB in parallel and wait for both of them to be completed.
  const [a, b] = yield this.run.task(["TaskA", "foo"], ["TaskB", "bar"]);

  if (a > b) {
    yield this.run.task("TaskC");
  } else {
    yield this.run.task("TaskD");
  }
};

// AsynchronousWorkflow
module.exports.handle = function*() {
  // Start TaskA and TaskB without waiting them to complete
  this.run.task("TaskA");
  this.run.task("TaskB");

  yield this.run.task("TaskC");
  yield this.run.task("TaskD");
};

// EventWorkflow
module.exports = {
  *handle() {
    this.state = true;

    yield this.run.task("TaskA");

    if (this.state) {
      yield this.run.task("TaskB");
    } else {
      yield this.run.task("TaskC");
    }
  },
  *onEvent(eventName, eventData) {
    if (eventName === "MyEvent") {
      this.state = false;
    }
  }
};

// WaitWorkflow
module.exports.handle = function*() {
  yield this.run.task("TaskA");

  // wait for 5 seconds
  yield this.wait.for(5);

  yield this.run.task("TaskB");
});

// WaitEventWorkflow
module.exports.handle = function*() {
  // Wait until the event at most 4 seconds
  const event = yield this.wait.event("MyEvent").for(4);

  if (event) {
    // If event has been triggered within 4 seconds
    yield this.run.task("TaskA");
  } else {
    // else
    yield this.run.task("TaskB");
  }
};

How it works

Install our Agent on your servers, download the Zenaton libraries and start processing jobs.

Read Documentation

Libraries

Install our open source libraries in your code. They provide an easy syntax to define, dispatch and orchestrate background jobs on your workers.

Agent

Install our Agent and use your server as a worker to process jobs concurrently. You can also designate specific jobs on dedicated servers depending such as micro-services or cpu intensive jobs.

Hosted Queues

Zenaton automatically deploys clustered queues infrastructure according to your usage. Leave server management to queuing system specialists.

Zenaton Engine

Manage the state of your workflows and orchestrate logic

Monitoring and Error Management

Our control panels provide real-time and historic logs of your asynchronous processing, so that you understand what is actually happening on your system. Errors trigger alarms, and you can retry faulty jobs manually or automatically.

Simplified Infrastructure Sophisticated Logic

The Business logic is written into your code and executed through Zenaton workflow engine that manages the queuing, storing of database states and then executes tasks on your servers.

Workflows as code

Write domain logic directly in your application code. Include loops, wait function, parallel and asynchronous executions.

Real-Time Monitoring

Insights into overall processes and execution history. View error data and logs in a central location. Resume, retry or kill processes from the dashboard. Learn More