@sourceloop /任务服务
概述
一个可重用的、可定制的、基于工作流的任务服务，它根据分布式系统中通过不同微服务发生的各种事件创建一些可操作的任务。
要开始使用此服务的基本实现，请参见 沙盒/ task-ms-example.
安装
npm i @sourceloop/task-service
开始使用
你可以开始使用 @sourceloop/task-service 只需4步：
绑定组件
- 选择并绑定一个传入连接器和一个传出连接器。参考 输入和输出连接器部分了解更多详细信息
- 绑定任务服务的配置（可选）-
this.bind(TaskServingBindings.Config).to({
useCustomSequence: true, // enable this if you want to use your own sequence instead of one provided by the task service
// though note that using a custom sequence may break or completely disable the authentication and authorization implemenation of task service
});
配置数据源
- 要使用任务服务，您需要绑定至少两个数据源（尽管它们都可以连接到同一个数据库）。
- 这个
dataSourceName这两个的性质应该是WorkflowServiceSourceName和TaskDbSourceName任务服务导出的变量。例如，其中一个数据源看起来像这样-
import {inject, lifeCycleObserver, LifeCycleObserver} from '@loopback/core';
import {juggler} from '@loopback/repository';
import {TaskDbSourceName} from '@sourceloop/task-service';
import {config} from './config';
// Observe application's life cycle to disconnect the datasource when
// application is stopped. This allows the application to be shut down
// gracefully. The `stop()` method is inherited from `juggler.DataSource`.
// Learn more at https://loopback.io/doc/en/lb4/Life-cycle.html
@lifeCycleObserver('datasource')
export class PgDbDataSource
extends juggler.DataSource
implements LifeCycleObserver
{
static dataSourceName = TaskDbSourceName; // this is the line that should variable from task service
static readonly defaultConfig = config;
constructor(
@inject(`datasources.config.${TaskDbSourceName}`, {optional: true})
dsConfig: object = config,
) {
super(dsConfig);
}
}
运行迁移
- 要运行任务服务提供的迁移(可在源代码的迁移文件夹和您的
node_modules)，使用 db-migrate 包中。 - 只有在计划使用HTTP传出连接器时才运行HTTP迁移。
- 此外，现在有一个选项可以选择SQL迁移或PostgreSQL迁移。 注意：对于@sourceloop/cli用户，这个选项可以在搭建过程中通过选择“数据源类型”选项来指定。
命令
由任务服务运行的工作流可以具有用于各种操作的服务任务，但是任务服务提供了两个常见的服务任务
创建任务
这个命令需要一个包含任务列表的输入变量。该命令创建该变量中的所有任务，并使用创建的任务触发一个传出事件。要在节点上触发此命令，请使用主题- create-tasks 并提供一个结构为-的变量
{
"tasks": {
"values": [
{
// task data
...
}
]
}
}
结束任务
此命令预计将用作任务工作流结束事件的主题。例如，在camunda的情况下，它将是topic For 消息结束事件 节点。要使用此命令，请使用topic - end-task
输入和输出连接器
任务服务需要 IIncomingConnector 和一个 IOutgoingConnectors 实现从外部源发送和接收事件。默认情况下，任务服务附带两组不同的连接器。注意，你可以使用不同类型的传入和传出连接器（例如，传入事件通过Kafka接收，但传出事件通过Http发送给webhook订阅）。
卡夫卡
这组连接器实现了通过Kafka接收和发送事件的连接器。它可以被绑定为传入和传出连接器，并且需要一个额外的适配器绑定，以使kafka事件适应任务服务所期望的类型。你需要安装 kafkajs 要使用这些连接器- npm i kafkajs
// Bind the config for Kafka connectors
this.bind(TaskServiceKafkaModule.CONFIG).to({
// this part is required for both incoming and outgoing connectors
connection: {
clientId: process.env.KAFKA_CLIENT_ID,
brokers: [...(process.env.KAFKA_SERVER?.split(',') ?? ['localhost:9092'])],
...(process.env.KAFKA_SSL === 'true'
? {
ssl: true,
}
: {}),
},
// this part is required only if you use it as an incoming connector
consumer: {
groupId: process.env.KAFKA_GROUP_ID ?? 'task-service',
},
topics: ['sow.update'], // topics for receiving events
// this part is required only if you use it as an outgoing connector
producer: {},
output: {
topic: 'test', // topic for output events
},
});
// bind the connector as an incoming connector if required
this.bind(TaskServiceBindings.INCOMING_CONNECTOR).toClass(KafkaStreamService);
// bind the connector as an outgoing connector if required
this.bind(TaskServiceBindings.OUTGOING_CONNECTOR).toClass(KafkaStreamService);
// bind the default adapter (it is available in `@sourceloop/task-service/kafka`)
this.bind(TaskServiceKafkaModule.ADAPTER).toClass(KafkaEventAdapter);
HTTP
这组连接器实现了通过HTTP调用发送和接收事件的代码。
// this line binds both the incoming and outgoing connectors plus some controllers required by the both
this.component(TaskHttpComponent);
// you can override either of the connectors by adding a new binding for them after the above
传入的连接器
- 它通过端点-接收事件
/events/trigger - 这个端点的有效载荷看起来像这样
{
"key": "event",
"payload": {
"name": "test",
"description": "description"
},
"source": "test-source",
"description": "description"
}
即将离任的连接器
- 传出连接器向所有webhook订阅发布事件
- webhook订阅是通过点击
/webhooks/subscribe端点的有效载荷看起来像这样
{
"url": "http://localhost:3000", // the url that will hit with the payload for every outgoing event
"key": "event-key" // the event keys for which this url would be hit
}
和一些必需的请求头 x-api-key 和 x-api-secret.
- 这些标头的值是通过另一个端点生成的
/client-apps。该端点应该被每个新客户端访问一次，并返回用于发送和验证webhook调用的新生成的密钥和秘密。呼唤/client-apps期望以下主体-
{
"clientName": "dummyName"
}
- 每个webhook调用也发送两个头
x-task-signature和x-task-timestamp帮助验证客户端webhook调用的真实性。客户端可以通过使用事件有效负载和时间戳生成HMAC来验证此签名。下面给出了一个关于如何做到这一点的示例node.js代码
function validateSignature(request) {
const signature = request.headers['x-task-signature'];
const timestamp = request.headers['x-task-timestampt'];
const payload = request.body;
// the secret in this line is the one generated by the /client-apps endpoint
const hmac = createHmac('sha256', yourApiSecret);
hmac.update(`${JSON.stringify(event)}:${timestamp}`);
const expectedSignature = hmac.digest('hex');
if (
// compare both the strings
!crypto.timingSafeEqual(
Buffer.from(expectedSignature),
Buffer.from(signature),
)
) {
// throw an error if signature does not match
throw new HttpErrors.Unauthorized(INVALID_WEBHOOK_SIGNATURE);
}
return true;
}
与Sequelize一起使用
此服务支持Sequelize作为底层ORM使用 @loopback / sequelize 扩展。为了使用它，您需要做以下更改。
1.要在应用程序中使用Sequelize，请在application.ts中添加以下内容：
this.bind(TaskServiceBindings.CONFIG).to({
useCustomSequence: false,
useSequelize: true,
});
- 使用
SequelizeDataSource在审计数据源中作为父类。参考 这 了解更多详情。
非对称令牌签名和验证
如果您正在使用非对称令牌签名和验证，则应该在服务中提供认证数据源，并在facade级别上提供认证数据源。auth数据库的数据源文件示例为：-
验证DB数据源：-
import {inject, lifeCycleObserver, LifeCycleObserver} from '@loopback/core';
import {juggler} from '@loopback/repository';
import {AuthDbSourceName} from '@sourceloop/core';
const DEFAULT_MAX_CONNECTIONS = 25;
const DEFAULT_DB_IDLE_TIMEOUT_MILLIS = 60000;
const DEFAULT_DB_CONNECTION_TIMEOUT_MILLIS = 2000;
const config = {
name: 'auth',
connector: 'postgresql',
host: process.env.DB_HOST,
port: process.env.DB_PORT,
user: process.env.DB_USER,
schema: process.env.DB_SCHEMA,
password: process.env.DB_PASSWORD,
database: process.env.AUTH_DB,
};
// Observe application's life cycle to disconnect the datasource when
// application is stopped. This allows the application to be shut down
// gracefully. The `stop()` method is inherited from `juggler.DataSource`.
// Learn more at https://loopback.io/doc/en/lb4/Life-cycle.html
@lifeCycleObserver('datasource')
export class AuthDataSource
extends juggler.DataSource
implements LifeCycleObserver
{
static dataSourceName = AuthDbSourceName;
static readonly defaultConfig = config;
constructor(
@inject('datasources.config.auth', {optional: true})
dsConfig: object = config,
) {
if (!!+(process.env.ENABLE_DB_CONNECTION_POOLING ?? 0)) {
const dbPool = {
max: +(process.env.DB_MAX_CONNECTIONS ?? DEFAULT_MAX_CONNECTIONS),
idleTimeoutMillis: +(
process.env.DB_IDLE_TIMEOUT_MILLIS ?? DEFAULT_DB_IDLE_TIMEOUT_MILLIS
),
connectionTimeoutMillis: +(
process.env.DB_CONNECTION_TIMEOUT_MILLIS ??
DEFAULT_DB_CONNECTION_TIMEOUT_MILLIS
),
};
dsConfig = {...dsConfig, ...dbPool};
}
super(dsConfig);
}
}
验证Cache Redis数据源：-
import {inject, lifeCycleObserver, LifeCycleObserver} from '@loopback/core';
import {AnyObject, juggler} from '@loopback/repository';
import {readFileSync} from 'fs';
import {AuthCacheSourceName} from '@sourceloop/core';
const config = {
name: process.env.REDIS_NAME,
connector: 'kv-redis',
host: process.env.REDIS_HOST,
port: process.env.REDIS_PORT,
password: process.env.REDIS_PASSWORD,
db: process.env.REDIS_DATABASE,
url: process.env.REDIS_URL,
tls:
+process.env.REDIS_TLS_ENABLED! && process.env.REDIS_TLS_CERT
? {
ca: readFileSync(process.env.REDIS_TLS_CERT),
}
: undefined,
sentinels:
+process.env.REDIS_HAS_SENTINELS! && process.env.REDIS_SENTINELS
? JSON.parse(process.env.REDIS_SENTINELS)
: undefined,
sentinelPassword:
+process.env.REDIS_HAS_SENTINELS! && process.env.REDIS_SENTINEL_PASSWORD
? process.env.REDIS_SENTINEL_PASSWORD
: undefined,
role:
+process.env.REDIS_HAS_SENTINELS! && process.env.REDIS_SENTINEL_ROLE
? process.env.REDIS_SENTINEL_ROLE
: undefined,
};
@lifeCycleObserver('datasource')
export class RedisDataSource
extends juggler.DataSource
implements LifeCycleObserver
{
static readonly dataSourceName = AuthCacheSourceName;
static readonly defaultConfig = config;
constructor(
@inject(`datasources.config.${process.env.REDIS_NAME}`, {optional: true})
dsConfig: AnyObject = config,
) {
if (
+process.env.REDIS_HAS_SENTINELS! &&
!!process.env.REDIS_SENTINEL_HOST &&
!!process.env.REDIS_SENTINEL_PORT
) {
dsConfig.sentinels = [
{
host: process.env.REDIS_SENTINEL_HOST,
port: +process.env.REDIS_SENTINEL_PORT,
},
];
}
super(dsConfig);
}
}