步骤3:流请求从服务器到客户端我们实现一个可写流Tunnel客户端发送请求数据:
JavaScript
const { Writable } = require(\'stream\');class SocketRequest extends Writable {constructor({ socket, requestId, request }) {super();this._socket = socket;this._requestId = requestId;this._socket.emit(\'request\', requestId, request);}_write(chunk, encoding, callback) {this._socket.emit(\'request-pipe\', this._requestId, chunk);this._socket.conn.once(\'drain\', () => {callback();});}_writev(chunks, callback) {this._socket.emit(\'request-pipes\', this._requestId, chunks);this._socket.conn.once(\'drain\', () => {callback();});}_final(callback) {this._socket.emit(\'request-pipe-end\', this._requestId);this._socket.conn.once(\'drain\', () => {callback();});}_destroy(e, callback) {if (e) {this._socket.emit(\'request-pipe-error\', this._requestId, e && e.message);this._socket.conn.once(\'drain\', () => {callback();});return;}callback();}}app.use(\'/\', (req, res) => {if (!connectedSocket) {res.status(404);res.send(\'Not Found\');return;}const requestId = uuidV4();const socketRequest = new SocketRequest({socket: connectedSocket,requestId,request: {method: req.method,headers: { ...req.headers },path: req.url,},});const onReqError = (e) => {socketRequest.destroy(new Error(e || \'Aborted\'));}req.once(\'aborted\', onReqError);req.once(\'error\', onReqError);req.pipe(socketRequest);req.once(\'finish\', () => {req.off(\'aborted\', onReqError);req.off(\'error\', onReqError);});// ...});实现一个流可读的get请求数据在客户端:
JavaScript
const stream = require(\'stream\');class SocketRequest extends stream.Readable {constructor({ socket, requestId }) {super();this._socket = socket;this._requestId = requestId;const onRequestPipe = (requestId, data) => {if (this._requestId === requestId) {this.push(data);}};const onRequestPipes = (requestId, data) => {if (this._requestId === requestId) {data.forEach((chunk) => {this.push(chunk);});}};const onRequestPipeError = (requestId, error) => {if (this._requestId === requestId) {this._socket.off(\'request-pipe\', onRequestPipe);this._socket.off(\'request-pipes\', onRequestPipes);this._socket.off(\'request-pipe-error\', onRequestPipeError);this._socket.off(\'request-pipe-end\', onRequestPipeEnd);this.destroy(new Error(error));}};const onRequestPipeEnd = (requestId, data) => {if (this._requestId === requestId) {this._socket.off(\'request-pipe\', onRequestPipe);this._socket.off(\'request-pipes\', onRequestPipes);this._socket.off(\'request-pipe-error\', onRequestPipeError);this._socket.off(\'request-pipe-end\', onRequestPipeEnd);if (data) {this.push(data);}this.push(null);}};this._socket.on(\'request-pipe\', onRequestPipe);this._socket.on(\'request-pipes\', onRequestPipes);this._socket.on(\'request-pipe-error\', onRequestPipeError);this._socket.on(\'request-pipe-end\', onRequestPipeEnd);}_read() {}}socket.on(\'request\', (requestId, request) => {console.log(`${request.method}: `, request.path);request.port = options.port;request.hostname = options.host;const socketRequest = new SocketRequest({requestId,socket: socket,});const localReq = http.request(request);socketRequest.pipe(localReq);const onSocketRequestError = (e) => {socketRequest.off(\'end\', onSocketRequestEnd);localReq.destroy(e);};const onSocketRequestEnd = () => {socketRequest.off(\'error\', onSocketRequestError);};socketRequest.once(\'error\', onSocketRequestError);socketRequest.once(\'end\', onSocketRequestEnd);// ...步骤4:流响应从客户端到服务器实现一个可写流将响应数据发送到Tunnel服务器:
JavaScript
const stream = require(\'stream\');class SocketResponse extends stream.Writable {constructor({ socket, responseId }) {super();this._socket = socket;this._responseId = responseId;}_write(chunk, encoding, callback) {this._socket.emit(\'response-pipe\', this._responseId, chunk);this._socket.io.engine.once(\'drain\', () => {callback();});}_writev(chunks, callback) {this._socket.emit(\'response-pipes\', this._responseId, chunks);this._socket.io.engine.once(\'drain\', () => {callback();});}_final(callback) {this._socket.emit(\'response-pipe-end\', this._responseId);this._socket.io.engine.once(\'drain\', () => {callback();});}_destroy(e, callback) {if (e) {this._socket.emit(\'response-pipe-error\', this._responseId, e && e.message);this._socket.io.engine.once(\'drain\', () => {callback();});return;}callback();}writeHead(statusCode, statusMessage, headers) {this._socket.emit(\'response\', this._responseId, {statusCode,statusMessage,headers,});}}socket.on(\'request\', (requestId, request) => {// ...stream request and send request to local server...const onLocalResponse = (localRes) => {localReq.off(\'error\', onLocalError);const socketResponse = new SocketResponse({responseId: requestId,socket: socket,});socketResponse.writeHead(localRes.statusCode,localRes.statusMessage,localRes.headers);localRes.pipe(socketResponse);};const onLocalError = (error) => {console.log(error);localReq.off(\'response\', onLocalResponse);socket.emit(\'request-error\', requestId, error && error.message);socketRequest.destroy(error);};localReq.once(\'error\', onLocalError);localReq.once(\'response\', onLocalResponse);});