47 lines
1.4 KiB
TypeScript
47 lines
1.4 KiB
TypeScript
import { connect } from 'nats';
|
|
import type { NatsConnection } from 'nats';
|
|
import { config } from '../config';
|
|
|
|
let nc: NatsConnection | null = null;
|
|
/**
|
|
* 在途去重：nc 只在 connect() **resolve 之后**才被赋值，所以并发的首批调用者读到的
|
|
* 都还是 null，于是各自拨一条连接；最后一个赋值进 nc，前面那些谁也够不着——drain 不掉、
|
|
* 也不计入 closeNats，就那么挂在服务端的连接表上直到进程死。记住这次在途的 promise，
|
|
* 让并发调用者共享同一次拨号；成功与失败都清空，失败后下一次调用可以重新拨。
|
|
*/
|
|
let connecting: Promise<NatsConnection> | null = null;
|
|
|
|
export async function getNats(): Promise<NatsConnection> {
|
|
if (nc && !nc.isClosed()) return nc;
|
|
if (connecting) return await connecting;
|
|
const dial = connect({ servers: config.natsUrl, name: config.serviceName, timeout: 5000 })
|
|
.then((c) => {
|
|
nc = c;
|
|
connecting = null;
|
|
return c;
|
|
})
|
|
.catch((err: unknown) => {
|
|
connecting = null;
|
|
throw err;
|
|
});
|
|
connecting = dial;
|
|
return await dial;
|
|
}
|
|
|
|
export async function checkNats(): Promise<'ok' | 'fail'> {
|
|
try {
|
|
const c = await getNats();
|
|
await c.flush();
|
|
return 'ok';
|
|
} catch {
|
|
return 'fail';
|
|
}
|
|
}
|
|
|
|
export async function closeNats(): Promise<void> {
|
|
if (nc) {
|
|
await nc.drain().catch(() => {});
|
|
nc = null;
|
|
}
|
|
}
|