Spaces:
Build error
Build error
File size: 5,557 Bytes
23a3b80 6e78e38 23a3b80 6e78e38 23a3b80 6e78e38 f9223d7 23a3b80 f9223d7 23a3b80 6e78e38 23a3b80 6e78e38 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 | import 'reflect-metadata';
import { container, singleton } from 'tsyringe';
import { KoaServer } from 'civkit/civ-rpc/koa';
import http2 from 'http2';
import http from 'http';
import { CrawlerHost } from '../api/crawler';
import { FsWalk, WalkOutEntity } from 'civkit/fswalk';
import path from 'path';
import fs from 'fs';
import { mimeOfExt } from 'civkit/mime';
import { Context, Next } from 'koa';
import { RPCRegistry } from '../services/registry';
import { AsyncResource } from 'async_hooks';
import { runOnce } from 'civkit/decorators';
import { randomUUID } from 'crypto';
import { ThreadedServiceRegistry } from '../services/threaded';
import { GlobalLogger } from '../services/logger';
import { AsyncLocalContext } from '../services/async-context';
import finalizer, { Finalizer } from '../services/finalizer';
import koaCompress from 'koa-compress';
@singleton()
export class CrawlStandAloneServer extends KoaServer {
logger = this.globalLogger.child({ service: this.constructor.name });
httpAlternativeServer?: typeof this['httpServer'];
assets = new Map<string, WalkOutEntity>();
constructor(
protected globalLogger: GlobalLogger,
protected registry: RPCRegistry,
protected crawlerHost: CrawlerHost,
protected threadLocal: AsyncLocalContext,
protected threads: ThreadedServiceRegistry,
) {
super(...arguments);
}
h2c() {
this.httpAlternativeServer = this.httpServer;
const fn = this.koaApp.callback();
this.httpServer = http2.createServer((req, res) => {
const ar = new AsyncResource('HTTP2ServerRequest');
ar.runInAsyncScope(fn, this.koaApp, req, res);
});
// useResourceBasedDefaultTracker();
return this;
}
override async init() {
await this.walkForAssets();
await super.init();
}
async walkForAssets() {
const files = await FsWalk.walkOut(path.resolve(__dirname, '..', '..', 'public'));
for (const file of files) {
if (file.type !== 'file') {
continue;
}
this.assets.set(file.relativePath.toString(), file);
}
}
override listen(port: number) {
const r = super.listen(port);
if (this.httpAlternativeServer) {
const altPort = port + 1;
this.httpAlternativeServer.listen(altPort, () => {
this.logger.info(`Alternative ${this.httpAlternativeServer!.constructor.name} listening on port ${altPort}`);
});
}
return r;
}
makeAssetsServingController() {
return (ctx: Context, next: Next) => {
const requestPath = ctx.path;
const file = requestPath.slice(1);
if (!file) {
return next();
}
const asset = this.assets.get(file);
if (asset?.type !== 'file') {
return next();
}
ctx.body = fs.createReadStream(asset.path);
ctx.type = mimeOfExt(path.extname(asset.path.toString())) || 'application/octet-stream';
ctx.set('Content-Length', asset.stats.size.toString());
return;
};
}
registerRoutes(): void {
this.koaApp.use(koaCompress({
filter(type) {
if (type.startsWith('text/')) {
return true;
}
if (type.includes('application/json') || type.includes('+json') || type.includes('+xml')) {
return true;
}
if (type.includes('application/x-ndjson')) {
return true;
}
return false;
}
}));
this.koaApp.use(this.makeAssetsServingController());
this.koaApp.use(this.registry.makeShimController());
}
// Using h2c server has an implication that multiple requests may share the same connection and x-cloud-trace-context
// TraceId is expected to be request-bound and unique. So these two has to be distinguished.
@runOnce()
override insertAsyncHookMiddleware() {
const asyncHookMiddleware = async (ctx: Context, next: () => Promise<void>) => {
const googleTraceId = ctx.get('x-cloud-trace-context').split('/')?.[0];
this.threadLocal.setup({
traceId: randomUUID(),
traceT0: new Date(),
googleTraceId,
});
return next();
};
this.koaApp.use(asyncHookMiddleware);
}
@Finalizer()
override async standDown() {
const tasks: Promise<any>[] = [];
if (this.httpAlternativeServer?.listening) {
(this.httpAlternativeServer as http.Server).closeIdleConnections?.();
this.httpAlternativeServer.close();
tasks.push(new Promise<void>((resolve, reject) => {
this.httpAlternativeServer!.close((err) => {
if (err) {
return reject(err);
}
resolve();
});
}));
}
tasks.push(super.standDown());
await Promise.all(tasks);
}
}
const instance = container.resolve(CrawlStandAloneServer);
export default instance;
if (process.env.NODE_ENV?.includes('dry-run')) {
instance.serviceReady().then(() => finalizer.terminate());
} else {
instance.serviceReady().then((s) => s.h2c().listen(parseInt(process.env.PORT || '') || 3000));
} |