This package contains the Golden Hippo Service Messenger, which is responsible for managing the communication between services in the Golden Hippo ecosystem. It gives services the ability to connect to a Queue and issue messages that may be consumed by other services.
To install the Golden Hippo Service Messenger, run the following command:
npm install @goldenhippo/hippo-service-messengerTo use the Golden Hippo Service Messenger, you must have access to the Redis instance that the service group is using.
You will pass this information to the HippoMessenger constructor.
import { HippoMessenger } from '@goldenhippo/hippo-service-messenger';
const hippoMessenger = new HippoMessenger({
/** ioRedis instance */
redisConnection: redisConnection,
/** Set the service channel (queue) this service will operate on */
serviceName: HIPPO_SERVICE.ACCOUNT, // Account Service
/** Set which events that this service would like to be notified of */
listensTo: [HIPPO_EVENT.CONFIG_CHANGE, HIPPO_EVENT.ORDER_SUCCESS],
});
/**
* Now register your service with the message broker. This is the method that informs the message
* broker which events you would like to listen for.
*/
hippoMessenger.register().then(() => {
console.log('Service registered with the message broker.');
});The register method is an asynchronous method that adds a job to the message broker's queue. The job data provided
includes the service name and the events that the service would like to listen for. The message broker will then update
its local cache to know which services are listening for which events. You could programmatically change the events
that your service listens to by calling hippoMessenger.setListeners([...events]) and then calling hippoMessenger.register()
again. This will update the message broker with the new events that your service is listening for.
NOTE Registration is a job that is processed by the message broker, so it may take a few seconds to complete. You will
only receive the events once the registration job has been processed.
When a service is registered with the message broker, the following steps are taken to ensure that the registration is persistent:
- The service name and events are issued to the message broker via a job.
- The message broker processes this job with the following steps:
- The message broker saves this information to a local cache that contains the service name and the events that the service is listening for.
- The message broker creates a repeat job that will re-register the service every 5 minutes.
- This repeat job is updated (deleted and re-created) any time the service is registered.
To issue an event, you will use the send method on the HippoMessenger instance.
hippoMessenger.send({
eventType: HIPPO_EVENT.ORDER_SUCCESS,
data: {...orderData},
});By default, the message broker will not notify you if you're subscribed to the same event type that you are sending. You can override this using the notifySelf parameter.
hippoMessenger.send({
eventType: HIPPO_EVENT.ORDER_SUCCESS,
data: {...orderData},
notifySelf: true,
});The notifySelf directly sends the job to your listener while still sending to the message broker. This reduces the extra call back from the broker. It will still validate the data object, but the error job will show in both the Service's queue and the message broker queue.
When you create a HippoMessenger instance, you will pass in an array of events that you would like to listen for.
When any of these events are sent to the message broker, they will be issued to the Queue for your service to consume.
You will need to add a worker to consume these messages. This can be done by constructing a HippoWorker instance.
This worker will immediately begin listening for messages on the Queue and will call the processor function when a
message is received.
import { HippoWorker } from '@goldenhippo/hippo-service-messenger';
const hippoWorker = new HippoWorker({
/** Your messenger instance, this links the worker to the Queue. */
messenger: hippoMessenger,
/**
* This is the function that will be called when a message is received. Only one worker will
* process each job, so you can add as many workers as you like.
*/
processor: async (job: HippoJob) => {
const { data, jobType, origin } = job;
// Typically, you only need to define one worker and use the jobType to determine what to do with the data.
switch (jobType) {
case HIPPO_EVENT.ORDER_SUCCESS:
// Do something with the data
console.log(data.orderId, data.total);
break;
case HIPPO_EVENT.ACCOUNT_REGISTERED:
// Do something else with the data
console.log(data.accountId);
break;
default:
// You can log the jobType you're
break;
}
console.log(jobType, data);
/**
* The worker needs to return a value to the message broker to let it know if the job was
* successful or not. For failed jobs, you should throw an error. Return values must be
* serializable (typically a string, boolean, or simple JSON object is used) as they are
* stored in Redis.
*/
return true; //
},
});
// Now start the worker
hippoWorker.start();Note: All events are set to the same queue, you can have multiple workers (across multiple dynos) listening to the same queue. Only one worker will process each job, so you can add as many workers as you like. This allows you to scale your workers horizontally.
WARNING: Based on the above, you may realize that having multiple services on the same queue would mean that only one
service will process a given job. That is correct! This is why it important to use the correct serviceName when
creating the HippoMessenger instance. The goal is to ensure that each service is only listening to its own Queue,
therefore, assigning a unique serviceName to each service is important. See the Testing section for more
information on how to use Queues during testing and the Canary Deployment section for more
information on how to deploy two versions of the same service without causing worker conflicts.
Ideally, you will use worker dynos to process messages within your service, while web dynos will be used for your REST APIs. This approach will allow you to scale your workers independently of your web dynos and ensure that your workers are always available to process messages and your web dynos are always available to serve your API.
Event types are managed within this package. This allows for a consistent set of event types across all services. When a new event type is required, it will be added to this package. The job data within the event will be validated against a Zod schema to ensure that the data is correct before being issued to the other services.
View the EVENT_TYPES enumeration docs here.
New event types will be added on an as-needed basis. If a new event type is required, a new schema will be added to this package. The new event type will only be available to services who have updated to the latest version of this package. If your services do not need the new event type(s), you do not need to update. However, it is recommended to regularly update to the latest version to ensure that you can receive all event types (or access updated schemas -- see below).
If an event type needs to be modified, the schema for that event type will need to be updated. While additional fields can be added to the schema, existing fields should not be removed or changed. This will ensure that the data being sent is consistent across all services. Fields should be set as optional unless they are absolutely required, otherwise the data will be rejected by the schema.
The event job data should include as much information as possible (while maintaining a reasonable size) to allow multiple services to consume the data without needing to make additional requests.
In general, event types will NOT be removed. If an event type is no longer needed, it should be marked as deprecated and a new event type should be created to replace it. This will allow services to continue to consume the old event type while they update to the new event type. Once all services have updated, the deprecated event type can be removed.
When testing your service, or running your service in a Development / Staging / UAT environment, you have a few options:
You can run a local Redis instance on your machine and connect to it using the ioredis package. This will allow you to
test your service without needing to connect to the Golden Hippo Redis instance. This is useful for local development and
testing. You will not receive any messages from other services, but you can send messages to your own service. The package
includes all event types, so you can simply follow the schema of the event you'd like to test.
If you are running your service in a shared environment (e.g. Staging), you can connect to the shared Redis instance. This will allow you to send and receive messages from other services in the shared environment. This is the same process as connecting in production, just with a different Redis instance.
When deploying a new version of your service, you may want to deploy it in a canary deployment. This will allow you to
test the new version of your service in production without affecting all users. When deploying a canary version of your
service, you should ensure that you choose one of your deployed services to run the worker(s). Otherwise, you cannot be
sure which service will process the job. You can easily disable the worker on either the old or new version of your
service using various methods (e.g. environment variables, feature flags, etc.). However, to simplify things, the
HippoMessenger accepts an enableWorkers parameter that can be set to false to disable the worker(s) on that service.
import {HippoMessenger} from '@goldenhippo/hippo-service-messenger';
const hippoMessenger = new HippoMessenger({
/** ioRedis instance */
redisConnection: redisConnection,
/** Set the service channel (queue) this service will operate on */
serviceName: HIPPO_SERVICE.ACCOUNT, // Account Service
/** Set which events that this service would like to be notified of */
listensTo: [HIPPO_EVENT.CONFIG_CHANGE, HIPPO_EVENT.ORDER_SUCCESS],
/** Enable or disable any workers on this messenger instance */
enableWorkers: false, // e.g. process.env.ENABLE_WORKERS === 'true'
});Most canary deployments (and changes to your service) will be related to the REST API or other parts of your service that do not affect the worker(s). In these cases, the following process is suggested:
- Deploy the new version of your service with the worker(s) disabled.
- Test the new version of your service to ensure that it is working correctly.
- Disable the worker(s) on the old version of your service.
- Enable the worker(s) on the new version of your service.
If you are making changes to the worker(s) or the event types, you should ensure that the worker(s) are only running on one version of your service at a time. This effectively eliminates the option of a canary deployment for the worker(s) and you should follow the standard deployment process.
Under the hood, this package is effectively a wrapper around BullMQ. This package is used to simplify the process of setting up a worker and a messenger. It also provides a consistent set of event types and schemas to ensure that the data being sent is correct. This package is designed to be used in the Golden Hippo ecosystem and the event types are specific to Golden Hippo services. If you are looking for a more general solution, you may want to consider using BullMQ directly.
This package is written in Typescript and includes type definitions. This will allow you to use the package with Typescript and benefit from the type checking and intellisense that it provides. Additionally, this package makes use of enums to define the event types and services. This will allow you to use these enums in your code to ensure that you are using the correct event types and services. Often, the type definitions will not accept strings as input, but will require the use of the enum. This is intentional to ensure that the correct values are used.
This is an internal package and is not intended to be used outside the Golden Hippo ecosystem. If you are a Golden Hippo employee and would like to contribute to this package, please follow the standard contribution guidelines.
This package is proprietary and is not intended to be used outside the Golden Hippo ecosystem. Please refer to the Golden Hippo employee handbook for more information.