raven-worker

Raven-worker package

Description

Raven-worker is used as the base of every worker in the Raven framework.
A worker is a building block that can perform a specific task in a streaming data-flow.
A data-stream that is configured from one or multiple workers, is called a flow.
The goal of a flow is to filter, extract, enrich and store streaming data to easily manage all your (streaming) data.

Worker types

Workers come in three different types: extract, transform and load workers. These types are based on the graph theory where each worker represents a node.
Consider any Raven flow a directed acyclic graph.

About Workers

• Workers need to be as simple as possible, only do their job. Everything else will be handled by Raven.
• If an error occurs, just panic. Raven will handle monitoring and restarting workers. The Raven Worker base package will use a sequential backoff algorithm for handling incidental errors.
• If there isn't enough work, for a longer period, just stop. Raven will monitor backlogs and start new workers when necessary.

How to use

The following functions are exposed by the package:

New

This initializes the worker.

For the Raven framework, a worker needs three variables to work with:

• ravenworker.WithRavenURL to connect to the Raven framework. Multiple values can be used here, as it will use them in a round robin configuration.
• ravenworker.WithWorkerID to identify itself and get new jobs.
• ravenworker.WithFlowID to get the right jobs for the flow it belongs to and therefore receive the right events (messages) to process.

The ravenworker.DefaultEnvironment() method can be used for convenience, to load the configuration from environment variables.

Example:

	c, err := ravenworker.New(
        ravenworker.DefaultEnvironment(),
    )

    if err != nil {
        // handle error
    }

The ravenworkworker.CustomEnvironment can be used when environment variables are not available. CustomEnvironment takes three string arguments to set the raven workflow url, flow ID and worker ID.

Example:

	c, err := ravenworker.New(
        ravenworker.Customenvironment("localhost:8023", "568e8bee-aca8-40c2-bfed-504ce103d4b6", "b557f6b3-b436-4635-9ae0-010fed184168"),
    )

    if err != nil {
        // handle error
    }

Note: when specifying the raven workflow url, leave out any scheme (http/https) in the string. Raven-worker uses the capnproto protocol and the function will add the proper scheme for you.

Now you can use the methods:

Consume

When a worker is of type transform or load, use Consume to retrieve the message from the stream.

Example:

    ref, err := c.Consume(context.Background())
    if err != nil {
        // handle error
    }

Get

Get will retrieve the actual message.

Example:

    msg, err := c.Get(ref)
    if err != nil {
        // handle error
    }

Ack

Ack will acknowledge the message and proceeds the flow.

Example:

    msg, err := c.Ack(ref, WithMessage(msg), WithFilter())
    if err != nil {
        // handle error
    }

Produce

When a worker is of type transform or load, use Produce to put the new message or ack the message.
The actual content (payload) is stored in message.Content which takes a byte array. Convenience functions like JsonContent (which encodes the object to json byte array) exists.

Example:

    message := NewMessage()
    message.Content = JsonContent(obj)

    if err := c.Produce(message); err != nil {
        // handle error
    }