Class HotMesh

This example shows the full lifecycle of a HotMesh engine instance, including: initialization, deployment, activation and execution.

The system is self-cleaning and self-healing, with a built-in quorum for consensus and a worker pool for distributed processing. Workflows are automatically removed from the system once completed.

Example

import Redis from 'ioredis';
import { HotMesh } from '@hotmeshio/hotmesh';

const hotMesh = await HotMesh.init({
  appId: 'abc',
  engine: {
    redis: {
      class: Redis,
      options: { host, port, password, db }
    }
  }
});

await hotMesh.deploy(`
app:
  id: abc
  version: '1'
  graphs:
    - subscribes: abc.test
      activities:
        t1:
          type: trigger
`);

await hotMesh.activate('1');

await hotMesh.pubsub('abc.test');

await HotMesh.stop();

Properties

appId guid logger namespace disconnecting

Methods

activate add deploy export getQueryState getRaw getState getStatus hook interrupt psub pub pubQuorum pubsub punsub rollCall scrub stop sub subQuorum throttle unsub unsubQuorum guid init stop

Properties

appId

appId: string

guid

guid: string

logger

logger: ILogger

namespace

namespace: string

Staticdisconnecting

disconnecting: boolean = false

Methods

activate

  • activate(version, delay?): Promise<boolean>

  • Once the app YAML file is deployed to Redis, the activate function can be called to enable it for the entire quorum at the same moment.

    The approach is to establish the coordinated health of the system through series of call/response exchanges. Once it is established that the quorum is healthy, the quorum is instructed to run their engine in no-cache mode, ensuring that the Redis backend is consulted for the active app version each time a call is processed. This ensures that all engines are running the same version of the app, switching over at the same moment and then enabling cache mode to improve performance.

    Add a delay for the quorum to reach consensus if traffic is busy, but also consider throttling traffic flow to an acceptable level.

    Parameters

    • version: string
    • Optionaldelay: number

    Returns Promise<boolean>

add

  • add(streamData): Promise<string>

  • Add a transition message to the workstream, resuming leg 2 of a paused reentrant activity (e.g., await, worker, hook)

    Parameters

    Returns Promise<string>

deploy

  • deploy(pathOrYAML): Promise<HotMeshManifest>

  • When the app YAML file is ready, the deploy function can be called. This function is responsible for merging all referenced YAML source files and writing the JSON output to the file system and to Redis. It is also possible to embed the YAML in-line as a string.

    The version will not be active until activation is explicitly called.

    Parameters

    • pathOrYAML: string

    Returns Promise<HotMeshManifest>

export

getQueryState

  • getQueryState(jobId, fields): Promise<StringAnyType>

  • Returns searchable/queryable data for a job. In this example a literal field is also searched (the colon is used to track job status and is a reserved field; it can be read but not written).

    Parameters

    • jobId: string
    • fields: string[]

    Returns Promise<StringAnyType>

    Example

    const fields = ['fred', 'barney', '":"'];
    const queryState = await hotMesh.getQueryState('123', fields);
    //returns { fred: 'flintstone', barney: 'rubble', ':': '1' }

getRaw

getState

getStatus

  • getStatus(jobId): Promise<number>

  • Returns the status of a job. This is a numeric semaphore value that indicates the job's state. Any non-positive value indicates a completed job. Jobs with a value of -1 are pending and will automatically be scrubbed after a set period. Jobs a value around -1billion have been interrupted and will be scrubbed after a set period. Jobs with a value of 0 completed normally. Jobs with a positive value are still running.

    Parameters

    • jobId: string

    Returns Promise<number>

hook

  • hook(topic, data, status?, code?): Promise<string>

  • Re/entry point for an active job. This is used to resume a paused job and close the reentry point or leave it open for subsequent reentry. Because hooks are public entry points, they include a topic which is established in the app YAML file.

    When this method is called, a hook rule will be located to establish the exact activity and activity dimension for reentry.

    Parameters

    Returns Promise<string>

interrupt

psub

  • psub(wild, callback): Promise<void>

  • Listen to all output and interim emissions of a workflow topic matching a wildcard pattern.

    Parameters

    Returns Promise<void>

    Example

    await hotMesh.psub('a.b.c*', (topic, message) => {
     console.log(message);
    });

pub

  • pub(topic, data?, context?, extended?): Promise<string>

  • Starts a workflow

    Parameters

    Returns Promise<string>

    Example

    await hotMesh.pub('a.b.c', { key: 'value' });

pubQuorum

pubsub

  • pubsub(topic, data?, context?, timeout?): Promise<JobOutput>

  • Starts a workflow and awaits the response

    Parameters

    • topic: string
    • data: JobData = {}
    • Optionalcontext: JobState
    • Optionaltimeout: number

    Returns Promise<JobOutput>

    Example

    await hotMesh.pubsub('a.b.c', { key: 'value' });

punsub

rollCall

scrub

  • scrub(jobId): Promise<void>

  • Immediately deletes (DEL) a completed job from the system.

    Scrubbed jobs must be complete with a non-positive status value

    Parameters

    • jobId: string

    Returns Promise<void>

stop

sub

  • sub(topic, callback): Promise<void>

  • Subscribe (listen) to all output and interim emissions of a single workflow topic

    Parameters

    Returns Promise<void>

    Example

    await hotMesh.psub('a.b.c', (topic, message) => {
     console.log(message);
    });

subQuorum

throttle

  • throttle(options): Promise<boolean>

  • Sends a throttle message to the quorum (engine and/or workers) to limit the rate of processing. Pass -1 to throttle indefinitely. The value must be a non-negative integer and not exceed MAX_DELAY ms.

    When throttling is set, the quorum will pause for the specified time before processing the next message. Target specific engines and workers by passing a guid and/or topic. Pass no arguments to throttle the entire quorum.

    In this example, all processing has been paused indefinitely for the entire quorum. This is equivalent to an emergency stop.

    HotMesh is a stateless sequence engine, so the throttle can be adjusted up and down with no loss of data.

    Parameters

    Returns Promise<boolean>

    Example

    await hotMesh.throttle({ throttle: -1 });

unsub

  • unsub(topic): Promise<void>

  • Stop listening in on a single workflow topic

    Parameters

    • topic: string

    Returns Promise<void>

unsubQuorum

Staticguid

  • guid(): string

  • returns a guid using the same core guid generator used by the HotMesh (nanoid)

    Returns string

Staticinit

  • init(config): Promise<HotMesh>

  • Instance initializer

    Parameters

    Returns Promise<HotMesh>

    Example

    const config: HotMeshConfig = {
      appId: 'myapp',
      engine: {
        redis: {
          class: Redis,
          options: { host: 'localhost', port: 6379 }
        },
      },
      workers [...]
    };
    const hotMesh = await HotMesh.init(config);

Staticstop

Settings

Member Visibility

On This Page

Properties
Methods