Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 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 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 | 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 10x 10x 10x 10x 10x 1x 1x 1x 1x 1x 10x 10x 10x 10x 10x 2x 2x 2x 2x 2x 2x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 10x 10x 10x 10x 10x 10x 10x 10x 10x 10x 10x 1x 1x 1x 10x 10x 10x 10x 10x 10x | import {printTree} from 'tree-dump/lib/printTree';
import {type Http1CreateServerOpts, Http1Server, type Http1ServerOpts} from './Http1Server';
import {RpcError} from '@jsonjoy.com/rpc-error';
import {gzip} from '@jsonjoy.com/util/lib/compression/gzip';
import {RxLogicalChannelBase} from '@jsonjoy.com/rpc-calls/lib/channel/RxLogicalChannelBase';
import {RxLogicalChannelBaseDispatcher} from '@jsonjoy.com/rpc-calls/lib/dispatcher/RxLogicalChannelBaseDispatcher';
import {BatchDispatcher} from '@jsonjoy.com/rpc-calls/lib/dispatcher/BatchDispatcher';
import {RpcCodec} from '@jsonjoy.com/rpc-codec';
import type {Printable} from 'tree-dump/lib/types';
import type {AnyCallee} from '@jsonjoy.com/rpc-calls';
import type {RpcLogger} from '@jsonjoy.com/rpc-error';
import type {Http1ConnectionContext} from './Http1ConnectionContext';
import type {CorsOpts} from './Http1Cors';
const DEFAULT_MAX_PAYLOAD = 4 * 1024 * 1024;
export interface RpcServerOpts {
http1: Http1Server;
callee: AnyCallee;
logger?: RpcLogger;
cors?: CorsOpts;
}
export interface RpcServerStartOpts extends Omit<RpcServerOpts, 'http1'> {
port?: number;
host?: string;
server?: Omit<Http1ServerOpts, 'server'>;
create?: Http1CreateServerOpts;
}
export class RpcServer implements Printable {
public static readonly startWithDefaults = async (opts: RpcServerStartOpts): Promise<RpcServer> => {
const port = opts.port || 8080;
const logger = opts.logger ?? console;
const server = await Http1Server.create(opts.create);
const http1 = new Http1Server({...opts.server, server});
const rpc = new RpcServer({
callee: opts.callee,
http1,
logger,
cors: opts.cors,
});
rpc.enableDefaults();
await http1.start();
const listenArgs: any[] = [port];
if (opts.host) listenArgs.push(opts.host);
listenArgs.push(() => {
let host = server.address() || 'localhost';
if (typeof host === 'object') host = (host as any).address;
logger.log({msg: 'SERVER_STARTED', host, port});
});
server.listen(...listenArgs);
return rpc;
};
public readonly http1: Http1Server;
public readonly dispatcher: BatchDispatcher;
constructor(protected readonly opts: RpcServerOpts) {
this.dispatcher = new BatchDispatcher({callee: opts.callee as any});
const http1 = (this.http1 = opts.http1);
const onInternalError = http1.oninternalerror;
http1.oninternalerror = (error, res, req) => {
Eif (error instanceof RpcError) {
res.statusCode = 400;
const data = JSON.stringify(error.toJson());
res.end(data);
return;
}
onInternalError(error, res, req);
};
}
public enableHttpPing(): void {
const http1 = this.http1;
http1.enableHttpPing();
http1.enableKamalPing();
}
public enableCors(opts: CorsOpts | undefined = this.opts.cors): void {
this.http1.enableCors(opts);
}
private processHttpRpcRequest = async (ctx: Http1ConnectionContext) => {
const res = ctx.res;
const body = await ctx.body(DEFAULT_MAX_PAYLOAD);
Iif (!res.socket) return;
try {
const messageCodec = ctx.msgCodec;
const incomingMessages = messageCodec.decode(ctx.reqCodec, body);
try {
const outgoing = await this.dispatcher.onBatch(incomingMessages as any, ctx);
Iif (!res.socket) return;
const resCodec = ctx.resCodec;
messageCodec.writeBatch(resCodec, outgoing as any);
const buf = resCodec.encoder.writer.flush();
Iif (!res.socket) return;
res.end(buf);
} catch (error) {
const logger = this.opts.logger ?? console;
logger.error('HTTP_RPC_PROCESSING', error, {messages: incomingMessages});
throw RpcError.from(error);
}
} catch (error) {
Eif (typeof error === 'object' && error)
Iif ((error as any).message === 'Invalid JSON') throw RpcError.badRequest();
throw RpcError.from(error);
}
};
public enableHttpRpc(path = '/rx'): void {
const http1 = this.http1;
http1.route({
method: 'POST',
path,
handler: this.processHttpRpcRequest,
msgCodec: http1.codecs.msg.compact,
});
}
public enableJsonRcp2HttpRpc(path = '/rpc'): void {
const http1 = this.http1;
http1.route({
method: 'POST',
path,
handler: this.processHttpRpcRequest,
msgCodec: http1.codecs.msg.jsonRpc2,
});
}
public enableWsRpc(path = '/rx'): void {
const opts = this.opts;
const callee = opts.callee;
this.http1.ws({
path,
maxIncomingMessage: 2 * 1024 * 1024,
maxOutgoingBackpressure: 2 * 1024 * 1024,
handler: (ctx) => {
const codec = new RpcCodec(ctx.msgCodec, ctx.reqCodec, ctx.resCodec);
const logicalChannel = new RxLogicalChannelBase(ctx.connection, codec);
new RxLogicalChannelBaseDispatcher(logicalChannel, callee as any, ctx);
},
});
}
/**
* Exposes JSON Type schema under the GET /schema endpoint.
*/
public enableSchema(path: string = '/schema', method: string = 'GET'): void {
const responseBody: Uint8Array = Buffer.from('{}');
let responseBodyCompressed: Uint8Array = new Uint8Array(0);
gzip(responseBody).then((compressed) => (responseBodyCompressed = compressed));
this.http1.route({
method,
path,
handler: (ctx) => {
const res = ctx.res;
res.writeHead(200, 'OK', {
'Content-Type': 'application/json',
'Content-Encoding': 'gzip',
'Cache-Control': 'public, max-age=3600, immutable',
'Content-Length': responseBodyCompressed.length,
});
res.end(responseBodyCompressed);
},
});
}
public enableDefaults(): void {
this.enableCors();
this.enableHttpPing();
this.enableHttpRpc();
this.enableJsonRcp2HttpRpc();
this.enableWsRpc();
this.enableSchema();
}
// ---------------------------------------------------------------- Printable
public toString(tab = ''): string {
return (
`${this.constructor.name}` +
printTree(tab, [
(tab) => this.http1.toString(tab),
() => '',
(tab) => (this.opts.callee as unknown as Printable).toString(tab),
])
);
}
}
|