Dispark/edgetunnel
1
0
Fork 0
mirror of https://github.com/lush2020/edgetunnel.git synced 2026-03-24 00:48:39 +08:00
Code Issues Packages Projects Releases Wiki Activity

update udp close

Browse Source
This commit is contained in:
zizifn
2023-03-07 02:12:43 +08:00
committed by zizifn
parent 0cbadf872a
commit 1d45c93e2c
4 changed files with 142 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
apps/node-vless/src/main.ts
View File

@@ -107,7 +107,7 @@ vlessWServer.on('connection', async function connection(ws, request) {
const writer = udpClientStream.writable.getWriter();
// nodejs buffer to ArrayBuffer issue
// https://nodejs.org/dist/latest-v18.x/docs/api/buffer.html#bufbuffer
writer.write(
await writer.write(
chunk.buffer.slice(
chunk.byteOffset,
chunk.byteOffset + chunk.length
@@ -149,7 +149,7 @@ vlessWServer.on('connection', async function connection(ws, request) {
if (isUDP) {
udpClientStream = makeUDPSocketStream(portRemote, address);
const writer = udpClientStream.writable.getWriter();
writer.write(rawClientData);
writer.write(rawClientData).catch(error=>console.log)
writer.releaseLock();
remoteConnectionReadyResolve(udpClientStream);
} else {
@@ -301,11 +301,12 @@ function makeUDPSocketStream(portRemote, address) {
);
});
udpClient.on('error', (error) => {
console.log('udpClient error event', error);
controller.error(error);
});
},
transform(chunk: ArrayBuffer, controller) {
async transform(chunk: ArrayBuffer, controller) {
//seems v2ray will use same web socket for dns query..
//And v2ray will combine A record and AAAA record into one ws message and use 2 btye for dns query length
for (let index = 0; index < chunk.byteLength; ) {
@@ -315,13 +316,16 @@ function makeUDPSocketStream(portRemote, address) {
chunk.slice(index + 2, index + 2 + udpPakcetLength)
);
index = index + 2 + udpPakcetLength;
await new Promise((resolve, reject)=>{
udpClient.send(udpData, portRemote, address, (err) => {
if (err) {
console.log(err);
controller.error('Failed to send UDP packet !!');
console.log('udps send error', err);
controller.error(`Failed to send UDP packet !! ${err}`);
safeCloseUDP(udpClient);
}
resolve(true)
});
})
index = index;
}

48
dist/apps/node-vless/main.js vendored
View File

@@ -6128,7 +6128,7 @@ vlessWServer.on('connection', function connection(ws, request) {
const writer = udpClientStream.writable.getWriter();
// nodejs buffer to ArrayBuffer issue
// https://nodejs.org/dist/latest-v18.x/docs/api/buffer.html#bufbuffer
writer.write(chunk.buffer.slice(chunk.byteOffset, chunk.byteOffset + chunk.length));
yield writer.write(chunk.buffer.slice(chunk.byteOffset, chunk.byteOffset + chunk.length));
writer.releaseLock();
return;
}
@@ -6152,7 +6152,7 @@ vlessWServer.on('connection', function connection(ws, request) {
if (isUDP) {
udpClientStream = makeUDPSocketStream(portRemote, address);
const writer = udpClientStream.writable.getWriter();
writer.write(rawClientData);
writer.write(rawClientData).catch(error => console.log);
writer.releaseLock();
remoteConnectionReadyResolve(udpClientStream);
}
@@ -6282,29 +6282,35 @@ function makeUDPSocketStream(portRemote, address) {
controller.enqueue(Buffer.concat([new Uint8Array([0, info.size]), message]));
});
udpClient.on('error', (error) => {
console.log('udpClient error event', error);
controller.error(error);
});
},
transform(chunk, controller) {
//seems v2ray will use same web socket for dns query..
//And v2ray will combine A record and AAAA record into one ws message and use 2 btye for dns query length
for (let index = 0; index < chunk.byteLength;) {
const lengthBuffer = chunk.slice(index, index + 2);
const udpPakcetLength = new DataView(lengthBuffer).getInt16(0);
const udpData = new Uint8Array(chunk.slice(index + 2, index + 2 + udpPakcetLength));
index = index + 2 + udpPakcetLength;
udpClient.send(udpData, portRemote, address, (err) => {
if (err) {
console.log(err);
controller.error('Failed to send UDP packet !!');
safeCloseUDP(udpClient);
}
});
index = index;
}
// console.log('dns chunk', chunk);
// console.log(portRemote, address);
// port is big-Endian in raw data etc 80 == 0x005d
return tslib_1.__awaiter(this, void 0, void 0, function* () {
//seems v2ray will use same web socket for dns query..
//And v2ray will combine A record and AAAA record into one ws message and use 2 btye for dns query length
for (let index = 0; index < chunk.byteLength;) {
const lengthBuffer = chunk.slice(index, index + 2);
const udpPakcetLength = new DataView(lengthBuffer).getInt16(0);
const udpData = new Uint8Array(chunk.slice(index + 2, index + 2 + udpPakcetLength));
index = index + 2 + udpPakcetLength;
yield new Promise((resolve, reject) => {
udpClient.send(udpData, portRemote, address, (err) => {
if (err) {
console.log('udps send error', err);
controller.error(`Failed to send UDP packet !! ${err}`);
safeCloseUDP(udpClient);
}
resolve(true);
});
});
index = index;
}
// console.log('dns chunk', chunk);
// console.log(portRemote, address);
// port is big-Endian in raw data etc 80 == 0x005d
});
},
flush(controller) {
safeCloseUDP(udpClient);

2
dist/apps/node-vless/main.js.map vendored
View File

File diff suppressed because one or more lines are too long

129
test.mjs
View File

@@ -9,10 +9,14 @@ try {
let i = 0;
const readableStream = new ReadableStream({
start(control) {
// throw 'pipeTo error';
control.enqueue(undefined);
control.enqueue(1);
setTimeout(()=>{
control.close()
}, 500)
// control.close();
control.error('eroro000-----readableStream--------');
// control.error('eroro000-----readableStream--------');
// setTimeout(() => {
// console.log('-----------------100');
// control.error('eroro000-----readableStream--------');
@@ -36,30 +40,50 @@ try {
},
});
setTimeout(() => {
console.log('cancel');
}, 2000);
// setTimeout(() => {
// console.log('cancel');
// }, 2000);
await readableStream.pipeTo(
new WritableStream({
async write(chunk, controller) {
console.log(chunk);
// throw 'pipeTo error';
// controller.error('pipeTo has error');
// await delay(1);
// controller.error('error');
// if (chunk === 7) {
// throw 'error';
// }
},
close() {
console.log('close------WritableStream');
},
abort(reason) {
console.log('abort--------', reason);
},
})
);
// readableStream.pipeThrough(new TransformStream({
// start(controller){
// // setTimeout(()=>{
// // controller.error('xxxxxx')
// // console.log('--transform--start-');
// // },3000)
// },
// transform(chunk, controller){
// // throw 'err'
// Promise.reject('xxxx')
// // setTimeout(()=>{
// // console.log('--transform---');
// // // throw '333'
// // }, 2000)
// // controller.enqueue(chunk)
// }
// }))
// await read1.pipeTo(
// new WritableStream({
// async write(chunk, controller) {
// console.log(chunk);
// // throw 'pipeTo error';
// // controller.error('pipeTo has error');
// // await delay(1);
// // controller.error('error');
// // if (chunk === 7) {
// // throw 'error';
// // }
// },
// close() {
// console.log('close------WritableStream');
// },
// abort(reason) {
// console.log('abort--------', reason);
// },
// })
// );
// console.log('end--------');
@@ -69,3 +93,60 @@ try {
} catch (error) {
console.log('---end---', error);
}
try{
console.log('----------');
const transform = new TransformStream({
start(controller){
// setInterval(()=>{
// controller.enqueue('1234')
// }, 1000)
},
async transform(chunk, controller){
console.log('----------', chunk);
// controller.error('xxxxxxxxxxxxxxxxxxxxx')
// throw 'xxxxxx'
// console.log('----------', chunk);
// Promise.reject('xxxx')
setTimeout(()=>{
controller.error('xxxxxxxxxxxxxxxxxxxxx')
// throw '333'
}, 2000)
controller.enqueue(chunk)
return '-======='
}
})
transform.readable.pipeTo(
new WritableStream({
async write(chunk, controller) {
console.log(chunk);
// throw 'pipeTo error';
// controller.error('pipeTo has error');
// await delay(1);
// controller.error('error');
// if (chunk === 7) {
// throw 'error';
// }
},
close() {
console.log('close------WritableStream');
},
abort(reason) {
console.log('abort--------', reason);
},
})).catch(error=>{
console.log(error);
})
const getWriter = transform.writable.getWriter()
await getWriter.write('abc').catch(error=>console.log(error))
getWriter.releaseLock()
console.log('xxxxxx');
}catch(errpr){
console.log(errpr);
}