mirror of
https://github.com/ragestudio/linebridge.git
synced 2025-06-09 10:34:17 +00:00
support on exit
This commit is contained in:
parent
d7fa42e53d
commit
4fe16f1a81
@ -6,254 +6,287 @@ import { EventEmitter } from "@foxify/events"
|
||||
import RedisMap from "../../lib/redis_map"
|
||||
|
||||
export default class RTEngineServer {
|
||||
constructor(params = {}) {
|
||||
this.params = params
|
||||
this.clusterMode = !!cluster.isWorker
|
||||
constructor(params = {}) {
|
||||
this.params = params
|
||||
this.clusterMode = !!cluster.isWorker
|
||||
|
||||
this.redisConnParams = {
|
||||
host: this.params.redisOptions?.host ?? process.env.REDIS_HOST ?? "localhost",
|
||||
port: this.params.redisOptions?.port ?? process.env.REDIS_PORT ?? 6379,
|
||||
username: this.params.redisOptions?.username ?? (process.env.REDIS_AUTH && process.env.REDIS_AUTH.split(":")[0]),
|
||||
password: this.params.redisOptions?.password ?? (process.env.REDIS_AUTH && process.env.REDIS_AUTH.split(":")[1]),
|
||||
db: this.params.redisOptions?.db ?? process.env.REDIS_DB ?? 0
|
||||
}
|
||||
this.redisConnParams = {
|
||||
host:
|
||||
this.params.redisOptions?.host ??
|
||||
process.env.REDIS_HOST ??
|
||||
"localhost",
|
||||
port:
|
||||
this.params.redisOptions?.port ??
|
||||
process.env.REDIS_PORT ??
|
||||
6379,
|
||||
username:
|
||||
this.params.redisOptions?.username ??
|
||||
(process.env.REDIS_AUTH &&
|
||||
process.env.REDIS_AUTH.split(":")[0]),
|
||||
password:
|
||||
this.params.redisOptions?.password ??
|
||||
(process.env.REDIS_AUTH &&
|
||||
process.env.REDIS_AUTH.split(":")[1]),
|
||||
db: this.params.redisOptions?.db ?? process.env.REDIS_DB ?? 0,
|
||||
}
|
||||
|
||||
this.redis = params.redis
|
||||
this.io = params.io
|
||||
}
|
||||
this.redis = params.redis
|
||||
this.io = params.io
|
||||
}
|
||||
|
||||
worker_id = nanoid()
|
||||
worker_id = nanoid()
|
||||
|
||||
io = null
|
||||
redis = null
|
||||
io = null
|
||||
redis = null
|
||||
|
||||
connections = null
|
||||
users = null
|
||||
connections = null
|
||||
users = null
|
||||
|
||||
events = new Map()
|
||||
events = new Map()
|
||||
|
||||
async initialize() {
|
||||
console.log("🌐 Initializing RTEngine server...")
|
||||
async initialize() {
|
||||
console.log("🌐 Initializing RTEngine server...")
|
||||
|
||||
if (!this.io) {
|
||||
this.io = new SocketIO.Server({
|
||||
path: this.params.root ?? "/",
|
||||
})
|
||||
}
|
||||
if (!this.io) {
|
||||
this.io = new SocketIO.Server({
|
||||
path: this.params.root ?? "/",
|
||||
})
|
||||
}
|
||||
|
||||
if (!this.redis) {
|
||||
this.redis = new redis({
|
||||
host: this.redisConnParams.host,
|
||||
port: this.redisConnParams.port,
|
||||
username: this.redisConnParams.username,
|
||||
password: this.redisConnParams.password,
|
||||
db: this.redisConnParams.db,
|
||||
})
|
||||
}
|
||||
if (!this.redis) {
|
||||
this.redis = new redis({
|
||||
lazyConnect: true,
|
||||
host: this.redisConnParams.host,
|
||||
port: this.redisConnParams.port,
|
||||
username: this.redisConnParams.username,
|
||||
password: this.redisConnParams.password,
|
||||
db: this.redisConnParams.db,
|
||||
maxRetriesPerRequest: null,
|
||||
})
|
||||
}
|
||||
|
||||
// create mappers
|
||||
this.connections = new RedisMap(this.redis, {
|
||||
refKey: "connections",
|
||||
worker_id: this.worker_id,
|
||||
})
|
||||
await this.redis.connect()
|
||||
|
||||
this.users = new RedisMap(this.redis, {
|
||||
refKey: "users",
|
||||
worker_id: this.worker_id,
|
||||
})
|
||||
// create mappers
|
||||
this.connections = new RedisMap(this.redis, {
|
||||
refKey: "connections",
|
||||
worker_id: this.worker_id,
|
||||
})
|
||||
|
||||
// register middlewares
|
||||
if (typeof this.middlewares === "object" && Array.isArray(this.middlewares)) {
|
||||
for (const middleware of this.middlewares) {
|
||||
this.io.use(middleware)
|
||||
}
|
||||
}
|
||||
this.users = new RedisMap(this.redis, {
|
||||
refKey: "users",
|
||||
worker_id: this.worker_id,
|
||||
})
|
||||
|
||||
// handle connection
|
||||
this.io.on("connection", (socket) => {
|
||||
this.eventHandler(this.onConnect, socket)
|
||||
})
|
||||
// register middlewares
|
||||
if (
|
||||
typeof this.middlewares === "object" &&
|
||||
Array.isArray(this.middlewares)
|
||||
) {
|
||||
for (const middleware of this.middlewares) {
|
||||
this.io.use(middleware)
|
||||
}
|
||||
}
|
||||
|
||||
console.log(`[RTEngine] Listening...`)
|
||||
console.log(`[RTEngine] Universal worker id [${this.worker_id}]`)
|
||||
// handle connection
|
||||
this.io.on("connection", (socket) => {
|
||||
this.eventHandler(this.onConnect, socket)
|
||||
})
|
||||
|
||||
return true
|
||||
}
|
||||
console.log(`[RTEngine] Listening...`)
|
||||
console.log(`[RTEngine] Universal worker id [${this.worker_id}]`)
|
||||
|
||||
close = () => {
|
||||
console.log(`Cleaning up RTEngine server...`)
|
||||
return true
|
||||
}
|
||||
|
||||
// WARN: Do not flush connections pls
|
||||
if (process.env.NODE_ENV !== "production") {
|
||||
console.log(`Flushing previus connections... (Only for dev mode)`)
|
||||
this.connections.flush()
|
||||
}
|
||||
close = () => {
|
||||
console.log(`Cleaning up RTEngine server...`)
|
||||
|
||||
if (this.clusterMode) {
|
||||
this.connections.flush(cluster.worker.id)
|
||||
}
|
||||
// WARN: Do not flush connections pls
|
||||
if (process.env.NODE_ENV !== "production") {
|
||||
console.log(`Flushing previus connections... (Only for dev mode)`)
|
||||
this.connections.flush()
|
||||
}
|
||||
|
||||
if (this.io) {
|
||||
this.io.close()
|
||||
}
|
||||
if (this.clusterMode) {
|
||||
this.connections.flush(cluster.worker.id)
|
||||
}
|
||||
|
||||
if (this.redis) {
|
||||
this.redis.quit()
|
||||
}
|
||||
}
|
||||
if (this.io) {
|
||||
this.io.close()
|
||||
}
|
||||
|
||||
onConnect = async (socket) => {
|
||||
console.log(`[RTEngine] new:client | id [${socket.id}]`)
|
||||
if (this.redis) {
|
||||
this.redis.quit()
|
||||
}
|
||||
}
|
||||
|
||||
// create eventBus
|
||||
socket.eventBus = new EventEmitter()
|
||||
socket.pendingTimeouts = new Set()
|
||||
onConnect = async (socket) => {
|
||||
console.log(`[RTEngine] new:client | id [${socket.id}]`)
|
||||
|
||||
// register events
|
||||
if (typeof this.events === "object") {
|
||||
for (const [key, handler] of this.events.entries()) {
|
||||
socket.on(key, (...args) => {
|
||||
this.eventHandler(handler, socket, ...args)
|
||||
})
|
||||
}
|
||||
}
|
||||
// create eventBus
|
||||
socket.eventBus = new EventEmitter()
|
||||
socket.pendingTimeouts = new Set()
|
||||
|
||||
// handle ping
|
||||
socket.on("ping", () => {
|
||||
socket.emit("pong")
|
||||
})
|
||||
// register events
|
||||
if (typeof this.events === "object") {
|
||||
console.log("registering events", this.events)
|
||||
for (const [key, handler] of this.events.entries()) {
|
||||
socket.on(key, (...args) => {
|
||||
this.eventHandler(handler, socket, ...args)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// handle disconnect
|
||||
socket.on("disconnect", () => {
|
||||
this.eventHandler(this.onDisconnect, socket)
|
||||
})
|
||||
// handle ping
|
||||
socket.on("ping", () => {
|
||||
socket.emit("pong")
|
||||
})
|
||||
|
||||
await this.connections.set(socket.id, socket)
|
||||
// handle disconnect
|
||||
socket.on("disconnect", () => {
|
||||
this.eventHandler(this.onDisconnect, socket)
|
||||
})
|
||||
|
||||
if (this.params.requireAuth) {
|
||||
await this.onAuth(socket, null, (this.params.handleAuth ?? this.handleAuth))
|
||||
} else if (socket.handshake.auth.token ?? socket.handshake.query.auth) {
|
||||
await this.onAuth(socket, (socket.handshake.auth.token ?? socket.handshake.query.auth), (this.params.handleAuth ?? this.handleAuth))
|
||||
}
|
||||
}
|
||||
await this.connections.set(socket.id, socket)
|
||||
|
||||
onDisconnect = async (socket,) => {
|
||||
console.log(`[RTEngine] disconnect:client | id [${socket.id}]`)
|
||||
if (this.params.requireAuth) {
|
||||
await this.onAuth(
|
||||
socket,
|
||||
null,
|
||||
this.params.handleAuth ?? this.handleAuth,
|
||||
)
|
||||
} else if (socket.handshake.auth.token ?? socket.handshake.query.auth) {
|
||||
await this.onAuth(
|
||||
socket,
|
||||
socket.handshake.auth.token ?? socket.handshake.query.auth,
|
||||
this.params.handleAuth ?? this.handleAuth,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
if (socket.eventBus.emit) {
|
||||
socket.eventBus.emit("disconnect")
|
||||
} else {
|
||||
console.warn(`[${socket.id}][@${socket.userData.username}] Cannot emit disconnect event`)
|
||||
}
|
||||
onDisconnect = async (socket) => {
|
||||
console.log(`[RTEngine] disconnect:client | id [${socket.id}]`)
|
||||
|
||||
const conn = await this.connections.get(socket.id)
|
||||
if (socket.eventBus.emit) {
|
||||
socket.eventBus.emit("disconnect")
|
||||
} else {
|
||||
console.warn(
|
||||
`[${socket.id}][@${socket.userData.username}] Cannot emit disconnect event`,
|
||||
)
|
||||
}
|
||||
|
||||
if (conn) {
|
||||
if (conn.user_id) {
|
||||
await this.users.del(conn.user_id)
|
||||
}
|
||||
}
|
||||
const conn = await this.connections.get(socket.id)
|
||||
|
||||
await this.connections.del(socket.id)
|
||||
}
|
||||
if (conn) {
|
||||
if (conn.user_id) {
|
||||
await this.users.del(conn.user_id)
|
||||
}
|
||||
}
|
||||
|
||||
onAuth = async (socket, token, handleAuth) => {
|
||||
if (typeof handleAuth !== "function") {
|
||||
console.log(`[RTEngine] [${socket.id}] No auth handler provided`)
|
||||
return false
|
||||
}
|
||||
await this.connections.del(socket.id)
|
||||
}
|
||||
|
||||
if (!token) {
|
||||
if (socket.handshake.auth.token) {
|
||||
token = socket.handshake.auth.token
|
||||
}
|
||||
if (socket.handshake.query.auth) {
|
||||
token = socket.handshake.query.auth
|
||||
}
|
||||
}
|
||||
onAuth = async (socket, token, handleAuth) => {
|
||||
if (typeof handleAuth !== "function") {
|
||||
console.log(`[RTEngine] [${socket.id}] No auth handler provided`)
|
||||
return false
|
||||
}
|
||||
|
||||
function err(code, message) {
|
||||
console.log(`[RTEngine] [${socket.id}] Auth error: ${code} >`, message)
|
||||
if (!token) {
|
||||
if (socket.handshake.auth.token) {
|
||||
token = socket.handshake.auth.token
|
||||
}
|
||||
if (socket.handshake.query.auth) {
|
||||
token = socket.handshake.query.auth
|
||||
}
|
||||
}
|
||||
|
||||
socket.emit("response:error", {
|
||||
code,
|
||||
message,
|
||||
})
|
||||
function err(code, message) {
|
||||
console.log(
|
||||
`[RTEngine] [${socket.id}] Auth error: ${code} >`,
|
||||
message,
|
||||
)
|
||||
|
||||
socket.disconnect()
|
||||
socket.emit("response:error", {
|
||||
code,
|
||||
message,
|
||||
})
|
||||
|
||||
return false
|
||||
}
|
||||
socket.disconnect()
|
||||
|
||||
if (!token) {
|
||||
return err(401, "auth:token_required")
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
const authResult = await handleAuth(socket, token, err)
|
||||
if (!token) {
|
||||
return err(401, "auth:token_required")
|
||||
}
|
||||
|
||||
if (authResult) {
|
||||
const conn = await this.connections.has(socket.id)
|
||||
const authResult = await handleAuth(socket, token, err)
|
||||
|
||||
// check if connection update is valid to avoid race condition(When user disconnect before auth verification is completed)
|
||||
if (!conn) {
|
||||
console.log(`Auth aborted`)
|
||||
return false
|
||||
}
|
||||
if (authResult) {
|
||||
const conn = await this.connections.has(socket.id)
|
||||
|
||||
this.users.set(authResult.user_id.toString(), {
|
||||
socket_id: socket.id,
|
||||
...authResult,
|
||||
})
|
||||
// check if connection update is valid to avoid race condition(When user disconnect before auth verification is completed)
|
||||
if (!conn) {
|
||||
console.log(`Auth aborted`)
|
||||
return false
|
||||
}
|
||||
|
||||
socket.emit("response:auth:ok")
|
||||
this.users.set(authResult.user_id.toString(), {
|
||||
socket_id: socket.id,
|
||||
...authResult,
|
||||
})
|
||||
|
||||
console.log(`[RTEngine] client:authenticated | socket_id [${socket.id}] | user_id [${authResult.user_id}] | username [@${authResult.username}]`)
|
||||
}
|
||||
}
|
||||
socket.emit("response:auth:ok")
|
||||
|
||||
eventHandler = async (fn, socket, payload) => {
|
||||
try {
|
||||
await fn(socket, payload, this)
|
||||
} catch (error) {
|
||||
console.error(error)
|
||||
console.log(
|
||||
`[RTEngine] client:authenticated | socket_id [${socket.id}] | user_id [${authResult.user_id}] | username [@${authResult.username}]`,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
if (typeof socket.emit === "function") {
|
||||
socket.emit("response:error", {
|
||||
code: 500,
|
||||
message: error.message,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
eventHandler = async (fn, socket, payload) => {
|
||||
try {
|
||||
await fn(socket, payload, this)
|
||||
} catch (error) {
|
||||
console.error(error)
|
||||
|
||||
find = {
|
||||
manyById: async (ids) => {
|
||||
if (typeof ids === "string") {
|
||||
ids = [ids]
|
||||
}
|
||||
if (typeof socket.emit === "function") {
|
||||
socket.emit("response:error", {
|
||||
code: 500,
|
||||
message: error.message,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const users = await this.users.getMany(ids)
|
||||
find = {
|
||||
manyById: async (ids) => {
|
||||
if (typeof ids === "string") {
|
||||
ids = [ids]
|
||||
}
|
||||
|
||||
return users
|
||||
},
|
||||
userBySocket: (socket_id) => {
|
||||
const users = await this.users.getMany(ids)
|
||||
|
||||
},
|
||||
userById: async (user_id) => {
|
||||
const user = await this.users.get(user_id)
|
||||
return users
|
||||
},
|
||||
userBySocket: (socket_id) => {},
|
||||
userById: async (user_id) => {
|
||||
const user = await this.users.get(user_id)
|
||||
|
||||
return user
|
||||
},
|
||||
socketByUserId: async (user_id) => {
|
||||
const user = await this.users.get(user_id)
|
||||
return user
|
||||
},
|
||||
socketByUserId: async (user_id) => {
|
||||
const user = await this.users.get(user_id)
|
||||
|
||||
if (!user) {
|
||||
return null
|
||||
}
|
||||
if (!user) {
|
||||
return null
|
||||
}
|
||||
|
||||
const socket = await this.connections.get(user.socket_id)
|
||||
const socket = await this.connections.get(user.socket_id)
|
||||
|
||||
return socket
|
||||
}
|
||||
}
|
||||
return socket
|
||||
},
|
||||
}
|
||||
}
|
||||
@ -2,140 +2,149 @@ import he from "hyper-express"
|
||||
import rtengine from "../../classes/rtengine"
|
||||
|
||||
export default class Engine {
|
||||
constructor(params) {
|
||||
this.params = params
|
||||
}
|
||||
constructor(params, ctx) {
|
||||
this.params = params
|
||||
this.ctx = ctx
|
||||
}
|
||||
|
||||
app = null
|
||||
router = null
|
||||
ws = null
|
||||
app = null
|
||||
router = null
|
||||
ws = null
|
||||
|
||||
initialize = async (params) => {
|
||||
const serverParams = {
|
||||
max_body_length: 50 * 1024 * 1024, //50MB in bytes,
|
||||
}
|
||||
initialize = async (params) => {
|
||||
const serverParams = {
|
||||
max_body_length: 50 * 1024 * 1024, //50MB in bytes,
|
||||
}
|
||||
|
||||
if (params.ssl) {
|
||||
serverParams.key_file_name = params.ssl?.key ?? null
|
||||
serverParams.cert_file_name = params.ssl?.cert ?? null
|
||||
}
|
||||
if (params.ssl) {
|
||||
serverParams.key_file_name = params.ssl?.key ?? null
|
||||
serverParams.cert_file_name = params.ssl?.cert ?? null
|
||||
}
|
||||
|
||||
this.app = new he.Server(serverParams)
|
||||
this.app = new he.Server(serverParams)
|
||||
|
||||
this.router = new he.Router()
|
||||
this.router = new he.Router()
|
||||
|
||||
// create a router map
|
||||
if (typeof this.router.map !== "object") {
|
||||
this.router.map = {}
|
||||
}
|
||||
// create a router map
|
||||
if (typeof this.router.map !== "object") {
|
||||
this.router.map = {}
|
||||
}
|
||||
|
||||
await this.router.any("*", (req, res) => {
|
||||
return res.status(404).json({
|
||||
code: 404,
|
||||
message: "Not found"
|
||||
})
|
||||
})
|
||||
await this.router.any("*", (req, res) => {
|
||||
return res.status(404).json({
|
||||
code: 404,
|
||||
message: "Not found",
|
||||
})
|
||||
})
|
||||
|
||||
await this.app.use(async (req, res, next) => {
|
||||
if (req.method === "OPTIONS") {
|
||||
// handle cors
|
||||
if (params.ignoreCors) {
|
||||
res.setHeader("Access-Control-Allow-Methods", "*")
|
||||
res.setHeader("Access-Control-Allow-Origin", "*")
|
||||
res.setHeader("Access-Control-Allow-Headers", "*")
|
||||
}
|
||||
await this.app.use(async (req, res, next) => {
|
||||
if (req.method === "OPTIONS") {
|
||||
// handle cors
|
||||
if (params.ignoreCors) {
|
||||
res.setHeader("Access-Control-Allow-Methods", "*")
|
||||
res.setHeader("Access-Control-Allow-Origin", "*")
|
||||
res.setHeader("Access-Control-Allow-Headers", "*")
|
||||
}
|
||||
|
||||
return res.status(204).end()
|
||||
}
|
||||
return res.status(204).end()
|
||||
}
|
||||
|
||||
// register body parser
|
||||
if (req.headers["content-type"]) {
|
||||
if (!req.headers["content-type"].startsWith("multipart/form-data")) {
|
||||
req.body = await req.urlencoded()
|
||||
req.body = await req.json(req.body)
|
||||
}
|
||||
}
|
||||
})
|
||||
// register body parser
|
||||
if (req.headers["content-type"]) {
|
||||
if (
|
||||
!req.headers["content-type"].startsWith(
|
||||
"multipart/form-data",
|
||||
)
|
||||
) {
|
||||
req.body = await req.urlencoded()
|
||||
req.body = await req.json(req.body)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
if (params.enableWebsockets) {
|
||||
this.ws = global.websocket = new rtengine({
|
||||
...params,
|
||||
handleAuth: params.handleWsAuth,
|
||||
root: `/${params.refName}`
|
||||
})
|
||||
if (params.enableWebsockets) {
|
||||
this.ws = global.websocket = new rtengine({
|
||||
...params,
|
||||
handleAuth: params.handleWsAuth,
|
||||
root: `/${params.refName}`,
|
||||
})
|
||||
|
||||
this.ws.initialize()
|
||||
this.ws.initialize()
|
||||
|
||||
await this.ws.io.attachApp(this.app.uws_instance)
|
||||
}
|
||||
}
|
||||
await this.ws.io.attachApp(this.app.uws_instance)
|
||||
}
|
||||
}
|
||||
|
||||
listen = async (params) => {
|
||||
if (process.env.lb_service) {
|
||||
let pathOverrides = Object.keys(this.router.map).map((key) => {
|
||||
return key.split("/")[1]
|
||||
})
|
||||
listen = async (params) => {
|
||||
if (process.env.lb_service) {
|
||||
let pathOverrides = Object.keys(this.router.map).map((key) => {
|
||||
return key.split("/")[1]
|
||||
})
|
||||
|
||||
// remove duplicates
|
||||
pathOverrides = [...new Set(pathOverrides)]
|
||||
// remove duplicates
|
||||
pathOverrides = [...new Set(pathOverrides)]
|
||||
|
||||
// remove "" and _map
|
||||
pathOverrides = pathOverrides.filter((key) => {
|
||||
if (key === "" || key === "_map") {
|
||||
return false
|
||||
}
|
||||
// remove "" and _map
|
||||
pathOverrides = pathOverrides.filter((key) => {
|
||||
if (key === "" || key === "_map") {
|
||||
return false
|
||||
}
|
||||
|
||||
return true
|
||||
})
|
||||
return true
|
||||
})
|
||||
|
||||
if (params.enableWebsockets) {
|
||||
process.send({
|
||||
type: "router:ws:register",
|
||||
id: process.env.lb_service.id,
|
||||
index: process.env.lb_service.index,
|
||||
data: {
|
||||
namespace: params.refName,
|
||||
listen: {
|
||||
ip: this.params.listen_ip,
|
||||
port: this.params.listen_port,
|
||||
},
|
||||
}
|
||||
})
|
||||
}
|
||||
if (params.enableWebsockets) {
|
||||
process.send({
|
||||
type: "router:ws:register",
|
||||
id: process.env.lb_service.id,
|
||||
index: process.env.lb_service.index,
|
||||
data: {
|
||||
namespace: params.refName,
|
||||
listen: {
|
||||
ip: this.params.listen_ip,
|
||||
port: this.params.listen_port,
|
||||
},
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
if (process.send) {
|
||||
// try to send router map to host
|
||||
process.send({
|
||||
type: "router:register",
|
||||
id: process.env.lb_service.id,
|
||||
index: process.env.lb_service.index,
|
||||
data: {
|
||||
router_map: this.router.map,
|
||||
path_overrides: pathOverrides,
|
||||
listen: {
|
||||
ip: this.params.listen_ip,
|
||||
port: this.params.listen_port,
|
||||
},
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
if (process.send) {
|
||||
// try to send router map to host
|
||||
process.send({
|
||||
type: "router:register",
|
||||
id: process.env.lb_service.id,
|
||||
index: process.env.lb_service.index,
|
||||
data: {
|
||||
router_map: this.router.map,
|
||||
path_overrides: pathOverrides,
|
||||
listen: {
|
||||
ip: this.params.listen_ip,
|
||||
port: this.params.listen_port,
|
||||
},
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
await this.app.listen(this.params.listen_port)
|
||||
}
|
||||
await this.app.listen(this.params.listen_port)
|
||||
}
|
||||
|
||||
// close should be synchronous
|
||||
close = () => {
|
||||
if (this.ws) {
|
||||
this.ws.clear()
|
||||
// close should be synchronous
|
||||
close = () => {
|
||||
if (this.ws) {
|
||||
this.ws.clear()
|
||||
|
||||
if (typeof this.ws?.close === "function") {
|
||||
this.ws.close()
|
||||
}
|
||||
}
|
||||
if (typeof this.ws?.close === "function") {
|
||||
this.ws.close()
|
||||
}
|
||||
}
|
||||
|
||||
if (typeof this.app?.close === "function") {
|
||||
this.app.close()
|
||||
}
|
||||
}
|
||||
if (typeof this.app?.close === "function") {
|
||||
this.app.close()
|
||||
}
|
||||
|
||||
if (typeof this.ctx.onClose === "function") {
|
||||
this.ctx.onClose()
|
||||
}
|
||||
}
|
||||
}
|
||||
481
src/server.js
481
src/server.js
@ -14,280 +14,333 @@ import registerWebsocketsEvents from "./initializators/registerWebsocketsEvents"
|
||||
import registerHttpRoutes from "./initializators/registerHttpRoutes"
|
||||
|
||||
async function loadEngine(engine) {
|
||||
const enginesPath = path.resolve(__dirname, "engines")
|
||||
const enginesPath = path.resolve(__dirname, "engines")
|
||||
|
||||
const selectedEnginePath = path.resolve(enginesPath, engine)
|
||||
const selectedEnginePath = path.resolve(enginesPath, engine)
|
||||
|
||||
if (!fs.existsSync(selectedEnginePath)) {
|
||||
throw new Error(`Engine ${engine} not found!`)
|
||||
}
|
||||
if (!fs.existsSync(selectedEnginePath)) {
|
||||
throw new Error(`Engine ${engine} not found!`)
|
||||
}
|
||||
|
||||
return require(selectedEnginePath).default
|
||||
return require(selectedEnginePath).default
|
||||
}
|
||||
|
||||
class Server {
|
||||
constructor(params = {}, controllers = {}, middlewares = {}, headers = {}) {
|
||||
this.isExperimental = defaults.isExperimental ?? false
|
||||
constructor(params = {}, controllers = {}, middlewares = {}, headers = {}) {
|
||||
this.isExperimental = defaults.isExperimental ?? false
|
||||
|
||||
if (this.isExperimental) {
|
||||
console.warn("\n🚧 This version of Linebridge is experimental! 🚧")
|
||||
console.warn(`Version: ${defaults.version}\n`)
|
||||
}
|
||||
if (this.isExperimental) {
|
||||
console.warn("\n🚧 This version of Linebridge is experimental! 🚧")
|
||||
console.warn(`Version: ${defaults.version}\n`)
|
||||
}
|
||||
|
||||
this.params = {
|
||||
...defaults.params,
|
||||
...params.default ?? params,
|
||||
}
|
||||
this.params = {
|
||||
...defaults.params,
|
||||
...(params.default ?? params),
|
||||
}
|
||||
|
||||
this.controllers = {
|
||||
...controllers.default ?? controllers,
|
||||
}
|
||||
this.controllers = {
|
||||
...(controllers.default ?? controllers),
|
||||
}
|
||||
|
||||
this.middlewares = {
|
||||
...middlewares.default ?? middlewares,
|
||||
}
|
||||
this.middlewares = {
|
||||
...(middlewares.default ?? middlewares),
|
||||
}
|
||||
|
||||
this.headers = {
|
||||
...defaults.headers,
|
||||
...headers.default ?? headers,
|
||||
}
|
||||
this.headers = {
|
||||
...defaults.headers,
|
||||
...(headers.default ?? headers),
|
||||
}
|
||||
|
||||
// fix and fulfill params
|
||||
this.params.useMiddlewares = this.params.useMiddlewares ?? []
|
||||
this.params.name = this.constructor.refName ?? this.params.refName
|
||||
this.params.useEngine = this.constructor.useEngine ?? this.params.useEngine ?? "hyper-express"
|
||||
this.params.listen_ip = this.constructor.listenIp ?? this.constructor.listen_ip ?? this.params.listen_ip ?? "0.0.0.0"
|
||||
this.params.listen_port = this.constructor.listenPort ?? this.constructor.listen_port ?? this.params.listen_port ?? 3000
|
||||
this.params.http_protocol = this.params.http_protocol ?? "http"
|
||||
this.params.http_address = `${this.params.http_protocol}://${defaults.localhost_address}:${this.params.listen_port}`
|
||||
this.params.enableWebsockets = this.constructor.enableWebsockets ?? this.params.enableWebsockets ?? false
|
||||
this.params.ignoreCors = this.constructor.ignoreCors ?? this.params.ignoreCors ?? true
|
||||
// fix and fulfill params
|
||||
this.params.useMiddlewares = this.params.useMiddlewares ?? []
|
||||
|
||||
this.params.routesPath = this.constructor.routesPath ?? this.params.routesPath ?? path.resolve(process.cwd(), "routes")
|
||||
this.params.wsRoutesPath = this.constructor.wsRoutesPath ?? this.params.wsRoutesPath ?? path.resolve(process.cwd(), "routes_ws")
|
||||
this.params.name = this.constructor.refName ?? this.params.refName
|
||||
|
||||
globalThis._linebridge = {
|
||||
name: this.params.name,
|
||||
useEngine: this.params.useEngine,
|
||||
listenIp: this.params.listen_ip,
|
||||
listenPort: this.params.listen_port,
|
||||
httpProtocol: this.params.http_protocol,
|
||||
httpAddress: this.params.http_address,
|
||||
enableWebsockets: this.params.enableWebsockets,
|
||||
ignoreCors: this.params.ignoreCors,
|
||||
routesPath: this.params.routesPath,
|
||||
validHttpMethods: defaults.valid_http_methods,
|
||||
}
|
||||
this.params.useEngine =
|
||||
this.constructor.useEngine ??
|
||||
this.params.useEngine ??
|
||||
"hyper-express"
|
||||
|
||||
return this
|
||||
}
|
||||
this.params.listen_ip =
|
||||
this.constructor.listenIp ??
|
||||
this.constructor.listen_ip ??
|
||||
this.params.listen_ip ??
|
||||
"0.0.0.0"
|
||||
|
||||
engine = null
|
||||
this.params.listen_port =
|
||||
this.constructor.listenPort ??
|
||||
this.constructor.listen_port ??
|
||||
this.params.listen_port ??
|
||||
3000
|
||||
|
||||
events = null
|
||||
this.params.http_protocol = this.params.http_protocol ?? "http"
|
||||
|
||||
ipc = null
|
||||
this.params.http_address = `${this.params.http_protocol}://${defaults.localhost_address}:${this.params.listen_port}`
|
||||
|
||||
ipcEvents = null
|
||||
this.params.enableWebsockets =
|
||||
this.constructor.enableWebsockets ??
|
||||
this.params.enableWebsockets ??
|
||||
false
|
||||
|
||||
eventBus = new EventEmitter()
|
||||
this.params.ignoreCors =
|
||||
this.constructor.ignoreCors ?? this.params.ignoreCors ?? true
|
||||
|
||||
initialize = async () => {
|
||||
const startHrTime = process.hrtime()
|
||||
this.params.disableBaseEndpoints =
|
||||
this.constructor.disableBaseEndpoints ??
|
||||
this.params.disableBaseEndpoints ??
|
||||
false
|
||||
|
||||
// register events
|
||||
if (this.events) {
|
||||
if (this.events.default) {
|
||||
this.events = this.events.default
|
||||
}
|
||||
this.params.routesPath =
|
||||
this.constructor.routesPath ??
|
||||
this.params.routesPath ??
|
||||
path.resolve(process.cwd(), "routes")
|
||||
|
||||
for (const [eventName, eventHandler] of Object.entries(this.events)) {
|
||||
this.eventBus.on(eventName, eventHandler)
|
||||
}
|
||||
}
|
||||
this.params.wsRoutesPath =
|
||||
this.constructor.wsRoutesPath ??
|
||||
this.params.wsRoutesPath ??
|
||||
path.resolve(process.cwd(), "routes_ws")
|
||||
|
||||
const engineParams = {
|
||||
...this.params,
|
||||
handleWsAuth: this.handleWsAuth,
|
||||
handleAuth: this.handleHttpAuth,
|
||||
requireAuth: this.constructor.requireHttpAuth,
|
||||
refName: this.constructor.refName ?? this.params.refName,
|
||||
ssl: this.ssl,
|
||||
}
|
||||
globalThis._linebridge = {
|
||||
name: this.params.name,
|
||||
useEngine: this.params.useEngine,
|
||||
listenIp: this.params.listen_ip,
|
||||
listenPort: this.params.listen_port,
|
||||
httpProtocol: this.params.http_protocol,
|
||||
httpAddress: this.params.http_address,
|
||||
enableWebsockets: this.params.enableWebsockets,
|
||||
ignoreCors: this.params.ignoreCors,
|
||||
routesPath: this.params.routesPath,
|
||||
validHttpMethods: defaults.valid_http_methods,
|
||||
}
|
||||
|
||||
// initialize engine
|
||||
this.engine = await loadEngine(this.params.useEngine)
|
||||
return this
|
||||
}
|
||||
|
||||
this.engine = new this.engine(engineParams)
|
||||
engine = null
|
||||
|
||||
if (typeof this.engine.initialize === "function") {
|
||||
await this.engine.initialize(engineParams)
|
||||
}
|
||||
events = null
|
||||
|
||||
// check if ws events are defined
|
||||
if (typeof this.wsEvents !== "undefined") {
|
||||
if (!this.engine.ws) {
|
||||
console.warn("`wsEvents` detected, but Websockets are not enabled! Ignoring...")
|
||||
} else {
|
||||
for (const [eventName, eventHandler] of Object.entries(this.wsEvents)) {
|
||||
this.engine.ws.events.set(eventName, eventHandler)
|
||||
}
|
||||
}
|
||||
}
|
||||
ipc = null
|
||||
|
||||
// try to execute onInitialize hook
|
||||
if (typeof this.onInitialize === "function") {
|
||||
try {
|
||||
await this.onInitialize()
|
||||
}
|
||||
catch (err) {
|
||||
console.error(err)
|
||||
process.exit(1)
|
||||
}
|
||||
}
|
||||
ipcEvents = null
|
||||
|
||||
// set defaults
|
||||
this.useDefaultHeaders()
|
||||
this.useDefaultMiddlewares()
|
||||
eventBus = new EventEmitter()
|
||||
|
||||
if (this.routes) {
|
||||
for (const [route, endpoint] of Object.entries(this.routes)) {
|
||||
this.engine.router.map[route] = new Endpoint(
|
||||
this,
|
||||
{
|
||||
...endpoint,
|
||||
route: route,
|
||||
handlers: {
|
||||
[endpoint.method]: endpoint.fn,
|
||||
},
|
||||
}
|
||||
)
|
||||
}
|
||||
}
|
||||
initialize = async () => {
|
||||
const startHrTime = process.hrtime()
|
||||
|
||||
// register http & ws routes
|
||||
this.engine = await registerHttpRoutes(this.params.routesPath, this.engine, this)
|
||||
this.engine = await registerWebsocketsEvents(this.params.wsRoutesPath, this.engine)
|
||||
// register events
|
||||
if (this.events) {
|
||||
if (this.events.default) {
|
||||
this.events = this.events.default
|
||||
}
|
||||
|
||||
// register base endpoints if enabled
|
||||
if (!this.params.disableBaseEndpoint) {
|
||||
await registerBaseEndpoints(this)
|
||||
}
|
||||
for (const [eventName, eventHandler] of Object.entries(
|
||||
this.events,
|
||||
)) {
|
||||
this.eventBus.on(eventName, eventHandler)
|
||||
}
|
||||
}
|
||||
|
||||
// use main router
|
||||
await this.engine.app.use(this.engine.router)
|
||||
const engineParams = {
|
||||
...this.params,
|
||||
handleWsAuth: this.handleWsAuth,
|
||||
handleAuth: this.handleHttpAuth,
|
||||
requireAuth: this.constructor.requireHttpAuth,
|
||||
refName: this.constructor.refName ?? this.params.refName,
|
||||
ssl: this.ssl,
|
||||
}
|
||||
|
||||
// if is a linebridge service then initialize IPC Channels
|
||||
if (process.env.lb_service) {
|
||||
await this.initializeIpc()
|
||||
}
|
||||
// initialize engine
|
||||
this.engine = await loadEngine(this.params.useEngine)
|
||||
|
||||
// try to execute beforeInitialize hook.
|
||||
if (typeof this.afterInitialize === "function") {
|
||||
await this.afterInitialize()
|
||||
}
|
||||
this.engine = new this.engine(engineParams, this)
|
||||
|
||||
// listen
|
||||
await this.engine.listen(engineParams)
|
||||
if (typeof this.engine.initialize === "function") {
|
||||
await this.engine.initialize(engineParams)
|
||||
}
|
||||
|
||||
// calculate elapsed time on ms, to fixed 2
|
||||
const elapsedHrTime = process.hrtime(startHrTime)
|
||||
const elapsedTimeInMs = elapsedHrTime[0] * 1e3 + elapsedHrTime[1] / 1e6
|
||||
// check if ws events are defined
|
||||
if (typeof this.wsEvents !== "undefined") {
|
||||
if (!this.engine.ws) {
|
||||
console.warn(
|
||||
"`wsEvents` detected, but Websockets are not enabled! Ignoring...",
|
||||
)
|
||||
} else {
|
||||
for (const [eventName, eventHandler] of Object.entries(
|
||||
this.wsEvents,
|
||||
)) {
|
||||
this.engine.ws.events.set(eventName, eventHandler)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
console.info(`🛰 Server ready!\n\t - ${this.params.http_protocol}://${this.params.listen_ip}:${this.params.listen_port} \n\t - Tooks ${elapsedTimeInMs.toFixed(2)}ms`)
|
||||
}
|
||||
// try to execute onInitialize hook
|
||||
if (typeof this.onInitialize === "function") {
|
||||
try {
|
||||
await this.onInitialize()
|
||||
} catch (err) {
|
||||
console.error(err)
|
||||
process.exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
initializeIpc = async () => {
|
||||
console.info("🚄 Starting IPC client")
|
||||
// set defaults
|
||||
this.useDefaultHeaders()
|
||||
this.useDefaultMiddlewares()
|
||||
|
||||
this.ipc = global.ipc = new IPCClient(this, process)
|
||||
}
|
||||
if (this.routes) {
|
||||
for (const [route, endpoint] of Object.entries(this.routes)) {
|
||||
this.engine.router.map[route] = new Endpoint(this, {
|
||||
...endpoint,
|
||||
route: route,
|
||||
handlers: {
|
||||
[endpoint.method]: endpoint.fn,
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
useDefaultHeaders = () => {
|
||||
this.engine.app.use((req, res, next) => {
|
||||
Object.keys(this.headers).forEach((key) => {
|
||||
res.setHeader(key, this.headers[key])
|
||||
})
|
||||
// register http & ws routes
|
||||
this.engine = await registerHttpRoutes(
|
||||
this.params.routesPath,
|
||||
this.engine,
|
||||
this,
|
||||
)
|
||||
this.engine = await registerWebsocketsEvents(
|
||||
this.params.wsRoutesPath,
|
||||
this.engine,
|
||||
)
|
||||
|
||||
next()
|
||||
})
|
||||
}
|
||||
// register base endpoints if enabled
|
||||
if (!this.params.disableBaseEndpoints) {
|
||||
await registerBaseEndpoints(this)
|
||||
}
|
||||
|
||||
useDefaultMiddlewares = async () => {
|
||||
const middlewares = await this.resolveMiddlewares([
|
||||
...this.params.useMiddlewares,
|
||||
...this.useMiddlewares ?? [],
|
||||
...defaults.useMiddlewares,
|
||||
])
|
||||
// use main router
|
||||
await this.engine.app.use(this.engine.router)
|
||||
|
||||
middlewares.forEach((middleware) => {
|
||||
this.engine.app.use(middleware)
|
||||
})
|
||||
}
|
||||
// if is a linebridge service then initialize IPC Channels
|
||||
if (process.env.lb_service) {
|
||||
await this.initializeIpc()
|
||||
}
|
||||
|
||||
register = {
|
||||
http: (endpoint, ..._middlewares) => {
|
||||
// check and fix method
|
||||
endpoint.method = endpoint.method?.toLowerCase() ?? "get"
|
||||
// try to execute beforeInitialize hook.
|
||||
if (typeof this.afterInitialize === "function") {
|
||||
await this.afterInitialize()
|
||||
}
|
||||
|
||||
if (defaults.fixed_http_methods[endpoint.method]) {
|
||||
endpoint.method = defaults.fixed_http_methods[endpoint.method]
|
||||
}
|
||||
// listen
|
||||
await this.engine.listen(engineParams)
|
||||
|
||||
// check if method is supported
|
||||
if (typeof this.engine.router[endpoint.method] !== "function") {
|
||||
throw new Error(`Method [${endpoint.method}] is not supported!`)
|
||||
}
|
||||
// calculate elapsed time on ms, to fixed 2
|
||||
const elapsedHrTime = process.hrtime(startHrTime)
|
||||
const elapsedTimeInMs = elapsedHrTime[0] * 1e3 + elapsedHrTime[1] / 1e6
|
||||
|
||||
// grab the middlewares
|
||||
let middlewares = [..._middlewares]
|
||||
console.info(
|
||||
`🛰 Server ready!\n\t - ${this.params.http_protocol}://${this.params.listen_ip}:${this.params.listen_port} \n\t - Tooks ${elapsedTimeInMs.toFixed(2)}ms`,
|
||||
)
|
||||
}
|
||||
|
||||
if (endpoint.middlewares) {
|
||||
if (!Array.isArray(endpoint.middlewares)) {
|
||||
endpoint.middlewares = [endpoint.middlewares]
|
||||
}
|
||||
initializeIpc = async () => {
|
||||
console.info("🚄 Starting IPC client")
|
||||
|
||||
middlewares = [...middlewares, ...this.resolveMiddlewares(endpoint.middlewares)]
|
||||
}
|
||||
this.ipc = global.ipc = new IPCClient(this, process)
|
||||
}
|
||||
|
||||
this.engine.router.map[endpoint.route] = {
|
||||
method: endpoint.method,
|
||||
path: endpoint.route,
|
||||
}
|
||||
useDefaultHeaders = () => {
|
||||
this.engine.app.use((req, res, next) => {
|
||||
Object.keys(this.headers).forEach((key) => {
|
||||
res.setHeader(key, this.headers[key])
|
||||
})
|
||||
|
||||
// register endpoint to http interface router
|
||||
this.engine.router[endpoint.method](endpoint.route, ...middlewares, endpoint.fn)
|
||||
},
|
||||
}
|
||||
next()
|
||||
})
|
||||
}
|
||||
|
||||
resolveMiddlewares = (requestedMiddlewares) => {
|
||||
const middlewares = {
|
||||
...this.middlewares,
|
||||
...defaults.middlewares,
|
||||
}
|
||||
useDefaultMiddlewares = async () => {
|
||||
const middlewares = await this.resolveMiddlewares([
|
||||
...this.params.useMiddlewares,
|
||||
...(this.useMiddlewares ?? []),
|
||||
...defaults.useMiddlewares,
|
||||
])
|
||||
|
||||
if (typeof requestedMiddlewares === "string") {
|
||||
requestedMiddlewares = [requestedMiddlewares]
|
||||
}
|
||||
middlewares.forEach((middleware) => {
|
||||
this.engine.app.use(middleware)
|
||||
})
|
||||
}
|
||||
|
||||
const execs = []
|
||||
register = {
|
||||
http: (endpoint, ..._middlewares) => {
|
||||
// check and fix method
|
||||
endpoint.method = endpoint.method?.toLowerCase() ?? "get"
|
||||
|
||||
requestedMiddlewares.forEach((middlewareKey) => {
|
||||
if (typeof middlewareKey === "string") {
|
||||
if (typeof middlewares[middlewareKey] !== "function") {
|
||||
throw new Error(`Middleware ${middlewareKey} not found!`)
|
||||
}
|
||||
if (defaults.fixed_http_methods[endpoint.method]) {
|
||||
endpoint.method = defaults.fixed_http_methods[endpoint.method]
|
||||
}
|
||||
|
||||
execs.push(middlewares[middlewareKey])
|
||||
}
|
||||
// check if method is supported
|
||||
if (typeof this.engine.router[endpoint.method] !== "function") {
|
||||
throw new Error(`Method [${endpoint.method}] is not supported!`)
|
||||
}
|
||||
|
||||
if (typeof middlewareKey === "function") {
|
||||
execs.push(middlewareKey)
|
||||
}
|
||||
})
|
||||
// grab the middlewares
|
||||
let middlewares = [..._middlewares]
|
||||
|
||||
return execs
|
||||
}
|
||||
if (endpoint.middlewares) {
|
||||
if (!Array.isArray(endpoint.middlewares)) {
|
||||
endpoint.middlewares = [endpoint.middlewares]
|
||||
}
|
||||
|
||||
middlewares = [
|
||||
...middlewares,
|
||||
...this.resolveMiddlewares(endpoint.middlewares),
|
||||
]
|
||||
}
|
||||
|
||||
this.engine.router.map[endpoint.route] = {
|
||||
method: endpoint.method,
|
||||
path: endpoint.route,
|
||||
}
|
||||
|
||||
// register endpoint to http interface router
|
||||
this.engine.router[endpoint.method](
|
||||
endpoint.route,
|
||||
...middlewares,
|
||||
endpoint.fn,
|
||||
)
|
||||
},
|
||||
}
|
||||
|
||||
resolveMiddlewares = (requestedMiddlewares) => {
|
||||
const middlewares = {
|
||||
...this.middlewares,
|
||||
...defaults.middlewares,
|
||||
}
|
||||
|
||||
if (typeof requestedMiddlewares === "string") {
|
||||
requestedMiddlewares = [requestedMiddlewares]
|
||||
}
|
||||
|
||||
const execs = []
|
||||
|
||||
requestedMiddlewares.forEach((middlewareKey) => {
|
||||
if (typeof middlewareKey === "string") {
|
||||
if (typeof middlewares[middlewareKey] !== "function") {
|
||||
throw new Error(`Middleware ${middlewareKey} not found!`)
|
||||
}
|
||||
|
||||
execs.push(middlewares[middlewareKey])
|
||||
}
|
||||
|
||||
if (typeof middlewareKey === "function") {
|
||||
execs.push(middlewareKey)
|
||||
}
|
||||
})
|
||||
|
||||
return execs
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = Server
|
||||
Loading…
x
Reference in New Issue
Block a user