soontao / cds-nats Goto Github PK
View Code? Open in Web Editor NEWNats message broker integration for CAP nodejs runtime
License: Other
Nats message broker integration for CAP nodejs runtime
License: Other
cds-nats/src/NatsRFCService.ts
Line 84 in b507a46
/* eslint-disable @typescript-eslint/ban-ts-comment */
import { cwdRequireCDS } from "cds-internal-tool";
import { QueryObject } from "cds-internal-tool/lib/types/ql";
import { Subscription } from "nats";
import path from "path";
import process from "process";
import { NatsMQBaseService } from "./NatsMQBaseService";
const HEADER_SERVICE_NAME = "servicename";
export class NatsRFCService extends NatsMQBaseService {
/**
* the app name of current app
*/
private appName!: string;
private appQueueGroup!: string;
async init(): Promise<any> {
await super.init();
if (this.options?.rfc?.enabled === true || this.options?.rfc?.enabled === "true") {
this.logger.info("rfc service enabled");
this.appName = this.options?.rfc?.name ?? path.basename(process.cwd());
this.appQueueGroup = this._toAppQueueGroup(this.appName);
this._listenServiceQueue();
}
}
private _toAppQueueGroup(serviceName: string) {
return [this.options.prefix ?? "", serviceName].join("");
}
private _listenServiceQueue() {
const sub = this.nc.subscribe(this.appQueueGroup, { queue: this.appQueueGroup });
this.logger.info("subscribe subject (Queue Group)", this.appQueueGroup, "for RFC");
this
._handleInboundCall(sub)
.catch(err => this.logger.error("receive error for subscription", "error", err));
}
private async _handleInboundCall(sub: Subscription) {
// use the service local event name, otherwise, framework could not found the handlers
const cds = cwdRequireCDS();
for await (const msg of sub) {
try {
const query = this.codec.decode(msg.data);
const headers = this._toHeader(msg);
const serviceName = headers[HEADER_SERVICE_NAME];
if (serviceName === undefined) {
this.logger.error("service name is not found for subject", msg.subject, "sid", msg.sid);
return;
}
const { user, tenant, id } = this._extractUserAndTenant(headers);
const srv = await cds.connect.to(serviceName);
// @ts-ignore
const tx = cds.context = srv.tx({ tenant, user, id });
try {
const response = await tx.run(query);
await tx.commit();
msg.respond(this.codec.encode(response));
}
catch (error) {
await tx.rollback();
}
}
catch (error) {
this.logger.error("process subject", msg.subject, "sid", msg.sid, "failed", error);
}
}
}
public async execute(appName: string, serviceName: string, query: QueryObject) {
const msg = await this.nc.request(
this._toAppQueueGroup(appName),
this.codec.encode(query),
{
headers: this._toNatsHeaders({ [HEADER_SERVICE_NAME]: serviceName }),
timeout: 60 * 1000, // TODO: options
},
);
return this.codec.decode(msg.data);
}
}
This issue lists Renovate updates and detected dependencies. Read the Dependency Dashboard docs to learn more.
Renovate tried to run on this repository, but found these problems.
These updates are currently rate-limited. Click on a checkbox below to force their creation now.
@typescript-eslint/eslint-plugin
, @typescript-eslint/parser
)These updates have all been created already. Click a checkbox below to force a retry/rebase of any.
@types/jest
, jest
)@typescript-eslint/eslint-plugin
, @typescript-eslint/parser
)docker/Dockerfile
.github/workflows/nodejs-coverage.yml
actions/checkout v3
actions/setup-node v3
actions/cache v3
codecov/codecov-action v3
.github/workflows/nodejs-lint.yml
actions/checkout v3
actions/setup-node v3
actions/cache v3
alstr/todo-to-issue-action v4.10.2
.github/workflows/nodejs-release.yml
actions/checkout v3
actions/setup-node v3
.github/workflows/nodejs.yml
actions/checkout v3
actions/setup-node v3
actions/cache v3
package.json
@newdash/newdash ^5.21.4
cds-internal-tool ^1.6.5
nats ^2.9.2
@types/jest 29.5.0
@types/node 18.16.12
@typescript-eslint/eslint-plugin 5.59.9
@typescript-eslint/parser 5.59.9
eslint 8.37.0
eslint-plugin-jest 27.2.1
jest 29.5.0
jest-junit 15.0.0
ts-jest 29.0.5
ts-node 10.9.1
typescript 5.0.2
node >=16
npm >=6
cds-nats/src/NatsMQBaseService.ts
Line 50 in b507a46
import { cwdRequireCDS, User } from "cds-internal-tool";
import { headers as MsgHeaders, JSONCodec, Msg } from "nats";
import os from "os";
import { FatalError } from "./errors";
import { NatsService } from "./NatsService";
interface HeaderObject {
id?: string;
type?: string;
source?: string;
time?: string;
datacontenttype?: string;
specversion?: string;
tenant?: string;
"user-class"?: string;
"user-attributes"?: string;
[key: string]: string | undefined
}
export class NatsMQBaseService extends NatsService {
protected codec = JSONCodec<any>();
// >> utils
/**
* convert nats `Headers` to general object
*
* @param msg nats message
* @returns
*/
protected _toHeader(msg: Msg): HeaderObject {
const headers: any = {};
if (msg.headers !== undefined) {
for (const key of msg.headers.keys()) {
headers[key] = msg.headers.get(key);
}
}
return headers;
}
protected _extractUserAndTenant(headers: HeaderObject): { user: User, id: string, tenant?: string } {
if ("user-class" in headers && "user-attributes" in headers) {
const cds = cwdRequireCDS() as any;
let user;
const tenant = headers["tenant"]; // TODO: warn if undefined
const className = headers["user-class"];
const id = headers["id"] ?? cds.utils.uuid();
const userAttrs = JSON.parse(headers["user-attributes"] ?? "{}");
if (className !== undefined) {
if (className === "User") {
user = new cds.User(userAttrs);
} else {
user = new cds.User[className](userAttrs);
}
}
return { user, tenant, id };
} else {
throw new FatalError("fatal: user context lost from mq");
}
}
/**
* convert `cds.Event` to nats `Headers`
*
* @param headers
* @param event
* @returns
*/
protected _toNatsHeaders(headers: any = {}, event?: string) {
const msgHeaders = MsgHeaders();
for (const [key, value] of Object.entries(headers)) {
msgHeaders.set(key, String(value));
}
const cds = cwdRequireCDS();
function setIfNotExit(key: string, value: () => string) {
if (!msgHeaders.has(key)) msgHeaders.set(key, value());
}
setIfNotExit("id", () => cds.context?.id ?? cds.utils.uuid());
setIfNotExit("type", () => event ?? "unknown");
setIfNotExit("source", () => `/default/community.cap/${os.hostname()}/${process.pid}`);
setIfNotExit("time", () => new Date().toISOString());
setIfNotExit("datacontenttype", () => "application/json");
setIfNotExit("specversion", () => "1.0");
setIfNotExit("user-class", () => {
let className = cds.context?.user?.constructor?.name;
if (className === undefined) className = "Anonymous";
return className;
});
setIfNotExit("user-attributes", () => JSON.stringify(cds.context?.user ?? "{}"));
if (cds.context?.tenant !== undefined) {
setIfNotExit("tenant", () => cds.context?.tenant);
}
return msgHeaders;
}
}
cds-nats/src/NatsRFCService.ts
Line 167 in f636356
* execute invocation remotely
*
* @param appName remote app name
* @param invocationInfo remote invocation info (full qualified name)
* @throws {RFCError|object} remote thrown message
* @returns result value from remote
*/
public async execute(appName: string, invocationInfo: RFCInvocationInfo) {
const msg = await this.nc.request(
this._toAppQueueGroup(appName),
this.codec.encode(invocationInfo),
{
headers: toNatsHeaders(),
timeout: this.timeout,
},
);
// process error
if (msg.headers?.get?.("throw") === "true") {
const throwObject = this.codec.decode(msg.data)
if (msg.headers?.get?.("error") === "true") {
const error = new RFCError(
throwObject?.message ?? "Unknown Error",
appName
);
Object.assign(error, throwObject)
throw error
}
else {
throw throwObject
}
}
return this.codec.decode(msg.data);
}
/**
* create an application proxy
*
* @param appName
* @returns
*/
public app(appName: string) {
return {
/**
* create a RFC service with limited methods
*
* @param serviceName the service name in remote app
* @returns {RFCService}
*/
service: (serviceName: string): RFCService => new Proxy({}, {
get: (_, prop) => {
if (typeof prop === "string") {
// emit/run/send
const methodName = prop;
return (...args: Array<any>) => {
return this.execute(appName, { methodName, serviceName, args });
};
}
// TODO: throw error
}
}) as any
};
}
}
export = NatsRFCService
Is your feature request related to a problem? Please describe.
A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]
Describe the solution you'd like
A clear and concise description of what you want to happen.
Describe alternatives you've considered
A clear and concise description of any alternative solutions or features you've considered.
Additional context
Add any other context or screenshots about the feature request here.
cds-nats/src/NatsRFCService.ts
Line 12 in f636356
/* eslint-disable @typescript-eslint/ban-ts-comment */
import { ApplicationService, cwdRequireCDS, TransactionMix } from "cds-internal-tool";
import { Subscription } from "nats";
import path from "path";
import process from "process";
import { RFCError } from "./errors";
import { NatsService } from "./NatsService";
import { RFCInvocationInfo, RFCService } from "./types";
import { createNatsHeaders, extractUserAndTenant, toHeader, toNatsHeaders } from "./utils";
// TODO: single instance validation
/**
* NatsRFCService
*
* enable RFC/RPC capability for CAP/Nats
*/
class NatsRFCService extends NatsService {
/**
* the app name of current app
A declarative, efficient, and flexible JavaScript library for building user interfaces.
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. ๐๐๐
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google โค๏ธ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.