3 lines
11 KiB
JavaScript
3 lines
11 KiB
JavaScript
"use strict";var u=Object.defineProperty;var v=Object.getOwnPropertyDescriptor;var p=Object.getOwnPropertyNames;var b=Object.prototype.hasOwnProperty;var y=(h,s)=>{for(var e in s)u(h,e,{get:s[e],enumerable:!0})},E=(h,s,e,i)=>{if(s&&typeof s=="object"||typeof s=="function")for(let t of p(s))!b.call(h,t)&&t!==e&&u(h,t,{get:()=>s[t],enumerable:!(i=v(s,t))||i.enumerable});return h};var w=h=>E(u({},"__esModule",{value:!0}),h);var T={};y(T,{CONNECTION_QUEUE_TIMEOUT:()=>C,PGLiteSocketHandler:()=>c,PGLiteSocketServer:()=>m});module.exports=w(T);var f=require("net"),C=6e4,g=class{constructor(s,e=!1){this.queue=[];this.processing=!1;this.lastHandlerId=null;this.db=s,this.debug=e}log(s,...e){this.debug&&console.log(`[QueryQueueManager] ${s}`,...e)}async enqueue(s,e,i){return new Promise((t,r)=>{let o={handlerId:s,message:e,resolve:t,reject:r,timestamp:Date.now(),onData:i};this.queue.push(o),this.log(`enqueued query from handler #${s}, queue size: ${this.queue.length}`),this.processing||this.processQueue()})}async processQueue(){if(!(this.processing||this.queue.length===0)){for(this.processing=!0;this.queue.length>0;){let s;if(this.db.isInTransaction()&&this.lastHandlerId){let t=this.queue.findIndex(r=>r.handlerId===this.lastHandlerId);t===-1?(this.log("transaction started, but no query from the same handler id found in queue",this.lastHandlerId),s=null):s=this.queue.splice(t,1)[0]}else s=this.queue.shift();if(!s)break;let e=Date.now()-s.timestamp;this.log(`processing query from handler #${s.handlerId} (waited ${e}ms)`);let i=0;try{await this.db.runExclusive(async()=>await this.db.execProtocolRawStream(s.message,{onRawData:t=>{i+=t.length,s.onData(t)}}))}catch(t){this.log(`query from handler #${s.handlerId} failed:`,t),s.reject(t);return}this.log(`query from handler #${s.handlerId} completed, ${i} bytes`),this.lastHandlerId=s.handlerId,s.resolve(i)}this.processing=!1,this.log("queue processing complete, queue length is",this.queue.length)}}getQueueLength(){return this.queue.length}clearQueueForHandler(s){let e=this.queue.length;this.queue=this.queue.filter(t=>t.handlerId===s?(t.reject(new Error("Handler disconnected")),!1):!0);let i=e-this.queue.length;i>0&&this.log(`cleared ${i} queries for handler #${s}`)}async clearTransactionIfNeeded(s){this.db.isInTransaction()&&this.lastHandlerId===s&&(await this.db.exec("ROLLBACK"),this.lastHandlerId=null,await this.processQueue())}},d=class d extends EventTarget{constructor(e){super();this.socket=null;this.active=!1;this.messageBuffer=Buffer.alloc(0);this.lastActivityTime=Date.now();this.queryQueue=e.queryQueue,this.closeOnDetach=e.closeOnDetach??!1,this.inspect=e.inspect??!1,this.debug=e.debug??!1,this.idleTimeout=e.idleTimeout??0,this.id=d.nextHandlerId++,this.log("constructor: created new handler")}get handlerId(){return this.id}log(e,...i){this.debug&&console.log(`[PGLiteSocketHandler#${this.id}] ${e}`,...i)}async attach(e){if(this.log(`attach: attaching socket from ${e.remoteAddress}:${e.remotePort}`),this.socket)throw new Error("Socket already attached");return this.socket=e,this.active=!0,this.lastActivityTime=Date.now(),e.setNoDelay(!0),this.idleTimeout>0&&this.resetIdleTimer(),this.log("attach: setting up socket event handlers"),e.on("data",i=>{this.lastActivityTime=Date.now(),this.resetIdleTimer(),setImmediate(async()=>{try{await this.handleData(i)}catch(t){this.log("socket on data error: ",t),this.handleError(t)}})}),e.on("error",i=>{setImmediate(()=>this.handleError(i))}),e.on("close",()=>{setImmediate(()=>this.handleClose())}),this.log("attach: socket handler ready"),this}resetIdleTimer(){this.idleTimeout<=0||(this.idleTimer&&clearTimeout(this.idleTimer),this.idleTimer=setTimeout(()=>{let e=Date.now()-this.lastActivityTime;this.log(`idle timeout after ${e}ms`),this.handleError(new Error("Idle timeout"))},this.idleTimeout))}async detach(e){if(this.log(`detach: detaching socket, close=${e??this.closeOnDetach}`),this.idleTimer&&(clearTimeout(this.idleTimer),this.idleTimer=void 0),this.queryQueue.clearQueueForHandler(this.id),await this.queryQueue.clearTransactionIfNeeded(this.id),!this.socket)return this.log("detach: no socket attached, nothing to do"),this;if(this.socket.removeAllListeners("data"),this.socket.removeAllListeners("error"),this.socket.removeAllListeners("close"),(e??this.closeOnDetach)&&this.socket.writable){this.log("detach: closing socket");try{this.socket.end(),this.socket.destroy()}catch(i){this.log("detach: error closing socket:",i)}}return this.socket=null,this.active=!1,this.messageBuffer=Buffer.alloc(0),this.log("detach: handler cleaned up"),this}get isAttached(){return this.socket!==null}async handleData(e){if(!this.socket||!this.active)return this.log("handleData: no active socket, ignoring data"),0;this.log(`handleData: received ${e.length} bytes`),this.messageBuffer=Buffer.concat([this.messageBuffer,e]),this.inspectData("incoming",e);try{let i=0;for(;this.messageBuffer.length>0;){let t=0,r=!1;if(this.messageBuffer.length>=4){let n=this.messageBuffer.readInt32BE(0);if(this.messageBuffer.length>=8){let a=this.messageBuffer.readInt32BE(4);(a===196608||a===196608)&&(t=n,r=this.messageBuffer.length>=t)}!r&&this.messageBuffer.length>=5&&(t=1+this.messageBuffer.readInt32BE(1),r=this.messageBuffer.length>=t)}if(!r||t===0){this.log(`handleData: incomplete message, buffering ${this.messageBuffer.length} bytes`);break}let o=this.messageBuffer.slice(0,t);if(this.messageBuffer=this.messageBuffer.slice(t),this.log(`handleData: processing message of ${o.length} bytes`),!this.active||!this.socket){this.log("handleData: socket no longer active, stopping processing");break}let l;if(await this.queryQueue.enqueue(this.id,new Uint8Array(o),n=>{this.log(`handleData: received ${n.length} bytes from PGlite`),this.inspectData("outgoing",n),n.length>0&&this.socket&&this.socket.writable&&this.active&&(this.log("handleData: writing response to socket"),this.socket?.writable?this.socket.write(Buffer.from(n),a=>{a?(this.log("handleData: error writing to socket:",a),l=a):this.log(`handleData: socket sent: ${n.length} bytes`)}):this.log("handleData: socket no longer writable")),i+=n.length}),l)throw l}return this.dispatchEvent(new CustomEvent("data",{detail:{incoming:e.length,outgoing:i}})),i}catch(i){throw this.log("handleData: error processing data:",i),i}}handleError(e){if(!this.active){this.log("handleError: handler not active, ignoring error");return}e.message?.includes("ECONNRESET")?this.log("handleError: client disconnected (ECONNRESET) - normal behavior"):e.message?.includes("Idle timeout")?this.log("handleError: connection idle timeout"):this.log("handleError:",e),this.active=!1,this.dispatchEvent(new CustomEvent("error",{detail:e})),this.detach(!0)}handleClose(){this.log("handleClose: socket closed"),this.active=!1,this.dispatchEvent(new CustomEvent("close")),this.detach(!1)}inspectData(e,i){if(this.inspect){console.log("-".repeat(75)),console.log(e==="incoming"?"-> incoming":"<- outgoing",i.length,"bytes");for(let t=0;t<i.length;t+=16){let r=Math.min(16,i.length-t),o="";for(let n=0;n<16;n++)if(n<r){let a=i[t+n];o+=a.toString(16).padStart(2,"0")+" "}else o+=" ";let l="";for(let n=0;n<r;n++){let a=i[t+n];l+=a>=32&&a<=126?String.fromCharCode(a):"."}console.log(`${t.toString(16).padStart(8,"0")} ${o} ${l}`)}}}};d.nextHandlerId=1;var c=d,m=class extends EventTarget{constructor(e){super();this.server=null;this.active=!1;this.handlers=new Set;this.db=e.db,e.path?this.path=e.path:(typeof e.port=="number"?this.port=e.port??e.port:this.port=5432,this.host=e.host||"127.0.0.1"),this.inspect=e.inspect??!1,this.debug=e.debug??!1,this.idleTimeout=e.idleTimeout??0,this.maxConnections=e.maxConnections??1,this.queryQueue=new g(this.db,this.debug),this.log(`constructor: created server on ${this.getServerConn()}`),this.log(`constructor: max connections: ${this.maxConnections}`),this.idleTimeout>0&&this.log(`constructor: idle timeout: ${this.idleTimeout}ms`)}log(e,...i){this.debug&&console.log(`[PGLiteSocketServer] ${e}`,...i)}async start(){if(this.log(`start: starting server on ${this.getServerConn()}`),this.server)throw new Error("Socket server already started");return await this.db.waitReady,this.active=!0,this.server=(0,f.createServer)(e=>{setImmediate(()=>this.handleConnection(e))}),this.server.maxConnections=this.maxConnections,new Promise((e,i)=>{if(!this.server)return i(new Error("Server not initialized"));if(this.server.on("error",t=>{this.log("start: server error:",t),this.dispatchEvent(new CustomEvent("error",{detail:t})),this.active||i(t)}),this.path)this.server.listen(this.path,()=>{this.log(`start: server listening on ${this.getServerConn()}`),this.dispatchEvent(new CustomEvent("listening",{detail:{path:this.path}})),e()});else{let t=this.server;t.listen(this.port,this.host,()=>{let r=t.address();if(r===null||typeof r!="object")throw Error("Expected address info");this.port=r.port,this.log(`start: server listening on ${this.getServerConn()}`),this.dispatchEvent(new CustomEvent("listening",{detail:{port:this.port,host:this.host}})),e()})}})}getServerConn(){return this.path?this.path:`${this.host}:${this.port}`}async stop(){this.log("stop: stopping server"),this.active=!1,this.log(`stop: detaching ${this.handlers.size} handlers`);for(let e of this.handlers)e.detach(!0);return this.handlers.clear(),this.server?new Promise(e=>{if(!this.server)return e();this.server.close(()=>{this.log("stop: server closed"),this.server=null,this.dispatchEvent(new CustomEvent("close")),e()})}):(this.log("stop: server not running, nothing to do"),Promise.resolve())}async handleConnection(e){let i={clientAddress:e.remoteAddress||"unknown",clientPort:e.remotePort||0};if(this.log(`handleConnection: new connection from ${i.clientAddress}:${i.clientPort}`),this.log(`handleConnection: active connections: ${this.handlers.size}, queued queries: ${this.queryQueue.getQueueLength()}`),!this.active){this.log("handleConnection: server not active, closing connection");try{e.end()}catch(r){this.log("handleConnection: error closing socket:",r)}return}if(this.handlers.size>=this.maxConnections){this.log("handleConnection: max connections reached, rejecting"),e.write(Buffer.from(`Too many connections
|
|
`)),e.end();return}let t=new c({queryQueue:this.queryQueue,closeOnDetach:!0,inspect:this.inspect,debug:this.debug,idleTimeout:this.idleTimeout});this.handlers.add(t),t.addEventListener("error",r=>{let o=r.detail;o?.message?.includes("ECONNRESET")?this.log(`handler #${t.handlerId}: client disconnected (ECONNRESET)`):o?.message?.includes("Idle timeout")?this.log(`handler #${t.handlerId}: idle timeout`):this.log(`handler #${t.handlerId}: error:`,o)}),t.addEventListener("close",()=>{this.log(`handler #${t.handlerId}: closed`),this.handlers.delete(t),this.log(`handleConnection: active connections: ${this.handlers.size}`)});try{await t.attach(e),this.dispatchEvent(new CustomEvent("connection",{detail:i}))}catch(r){this.log("handleConnection: error attaching socket:",r),this.handlers.delete(t),this.dispatchEvent(new CustomEvent("error",{detail:r}));try{e.end()}catch(o){this.log("handleConnection: error closing socket:",o)}}}getStats(){return{activeConnections:this.handlers.size,queuedQueries:this.queryQueue.getQueueLength(),maxConnections:this.maxConnections}}};0&&(module.exports={CONNECTION_QUEUE_TIMEOUT,PGLiteSocketHandler,PGLiteSocketServer});
|
|
//# sourceMappingURL=index.cjs.map
|