Deno 中的 Websocket 连接

Question

大家早上好，

我正在试验 Deno 和 Oak 框架。我想在前端和后端之间建立一个 WebSocket 连接。我知道如何使用 Deno 标准库 ('HTTP') 来做到这一点，但是在尝试使用 Oak 时，我总是遇到错误。

我当前的代码是：

import { Application, Router } from "https://deno.land/x/oak@v6.0.1/mod.ts";
import { WebSocket, acceptWebSocket, isWebSocketCloseEvent, acceptable } from 'https://deno.land/std@0.61.0/ws/mod.ts';
import { staticFileMiddleware } from './staticFileMiddleware.ts';

const app = new Application();
const router = new Router();

router
.get('/', (ctx) => {
    ctx.response.redirect('/index.html');
})

.get('/ws', async (ctx: any) => {
    // console.log(ctx.request.serverRequest);
    const sock = await ctx.upgrade();
    let id = socks.push(sock) - 1;
    for await (const ev of sock);
    socks.splice(id, 1);
    if (acceptable(ctx.request.serverRequest)) {
        const { conn, r: bufReader, h: bufWriter, headers } = ctx.request.serverRequest;
        const socket = await acceptWebSocket({
            conn,
            bufReader,
            bufWriter,
            headers
        });
        await chat(socket);
    } else {
        throw new Error('Error when connecting websocket');
    }
});

app.use(router.routes());
app.use(router.allowedMethods());
app.use(staticFileMiddleware);

app.addEventListener('listen', ({hostname, port, secure}) => {
    console.log(`Listening on ${secure ? 'https://' : 'http://'}${hostname || 'localhost'}:${port}`)
});

app.addEventListener('error', e => {
    console.log(e.error);
});

await app.listen({ port: 3000 });


async function chat(ws: WebSocket) {
    console.log(`Connected`);
    for await (let data of ws) {
        console.log(data, typeof data);
        ws.send('Your message was successfully received');
        if (isWebSocketCloseEvent(data)) {
            console.log('Goodbye');
            break;
        }
    }
}

我的静态文件中间件是：

import { Context, send } from "https://deno.land/x/oak@v6.0.1/mod.ts";

export const staticFileMiddleware = async (ctx: Context, next: Function) => {
    const path = `${Deno.cwd()}/public${ctx.request.url.pathname}`;
    if (await fileExists(path)) {
        await send(ctx, ctx.request.url.pathname, {
            root: `${Deno.cwd()}/public`
        })
    } else {
        await next();
    } 
}

async function fileExists(path: string) {
    try {
        const stats = await Deno.lstat(path);
        return stats && stats.isFile;
    } catch (e) {
        if (e && e instanceof Deno.errors.NotFound) {
            return false;
        } else {
            throw e;
        }
    }
}

客户端代码包括：

let ws;

window.addEventListener('DOMContentLoaded', () => {
    ws = new WebSocket(`ws://localhost:3000/ws`);
    ws.addEventListener('open', onConnectionOpen);
    ws.addEventListener('message', onMessageReceived);
});

function onConnectionOpen() {
    console.log('Connection Opened');
    ws.send('I am sending a message from the client side');
}

function onMessageReceived(event) {
    console.log('Message Received', event);
}

我的文件结构如下：

server.ts
staticFileMiddleware.ts
/public
   -client.js
   -index.html

如果您能帮助我并为我指明正确的方向，我将不胜感激。 提前感谢您的帮助！！！

Answer 1

我已经复制了上面的代码并尝试运行，我在sock.push() 块中遇到了错误

const sock = await ctx.upgrade();
let id = socks.push(sock) - 1;
for await (const ev of sock);
socks.splice(id, 1);

因此删除了不必要的代码。 server.ts

在破坏 ctx.request.serverRequest 时，我看到 h:bufWriter 在那里未定义并抛出类似的错误，

failed to accept websocket: TypeError: Cannot read property 'write' of undefined

这是更新后的代码，

import { Application, Router } from "https://deno.land/x/oak@v6.0.1/mod.ts";
import {
  acceptWebSocket,
  isWebSocketCloseEvent,
  isWebSocketPingEvent,
  WebSocket,
  acceptable
} from "https://deno.land/std/ws/mod.ts";
import { staticFileMiddleware } from './staticFileMiddleware.ts';

const app = new Application();
const router = new Router();

router
  .get('/', (ctx) => {
    ctx.response.redirect('/index.html');
  })

  .get('/ws', async (ctx: any) => {
    await ctx.upgrade();
    if (acceptable(ctx.request.serverRequest)) {
      const { conn, r: bufReader, w: bufWriter, headers } = ctx.request.serverRequest;
      acceptWebSocket({
        conn,
        bufWriter,
        bufReader,
        headers,
      })
        .then(chat)
        .catch(async (err) => {
          console.error(`failed to accept websocket: ${err}`);
        });
    } else {
      throw new Error('Error when connecting websocket');
    }
  });

app.use(router.routes());
app.use(router.allowedMethods());
app.use(staticFileMiddleware);

app.addEventListener('listen', ({ hostname, port, secure }) => {
  console.log(`Listening on ${secure ? 'https://' : 'http://'}${hostname || 'localhost'}:${port}`)
});

app.addEventListener('error', e => {
  console.log(e.error);
});

await app.listen({ port: 3000 });


async function chat(ws: WebSocket) {
  console.log(`Connected`);
  for await (let data of ws) {
    console.log(data, typeof data);
    ws.send('Your message was successfully received');
    if (isWebSocketCloseEvent(data)) {
      console.log('Goodbye');
      break;
    }
  }
}

deno run --allow-net --allow-read server.ts

Answer 2

我采用了一种稍微不同的方法，它适用于向 Joe 发送消息，这应该可以适应您的需求和解决方案/问题。

/*
running with Deno 1.2
deno run --inspect --allow-net ./beautiful-socket.js
*/
import { Application, Router, HttpError, send, Status } from "https://deno.land/x/oak@v6.0.1/mod.ts";
import { acceptable } from "https://deno.land/std@0.61.0/ws/mod.ts";

const port = 8123;
const users = new Set();
const app = new Application({state:{users}});
const router = new Router();

function broadcastEach(user){
    user.send(this);
}
function broadcast(msg){
    console.log('---broadcasting--->', typeof msg, msg);
    users.forEach(broadcastEach, msg);
}

router.get('/socket', async (context, next) => {
    context.response.status = 204;
    if( !acceptable(context.request.serverRequest) ){
        context.response.status = 400;
        throw new Error(`not upgradable to WebSocket`);
    }

    const socket = await context.upgrade();
    users.add(socket);
    broadcast(`hello! ${ socket.conn.rid }`);
    for await (const ev of socket) {
        if(socket.isClosed){
            users.delete(socket);
            broadcast(`bye! ${ socket.conn.rid }`);
            break;
        }else{
            broadcast(ev);
        };
    }
});

router.get('/', async (context) => {
    context.response.body = `<!doctype html>
<html><body>
<p>let's chat...open the console to chat it up</p>
<script>
console.log(123);
const pipe = new WebSocket("ws://${context.request.url.host}/socket");
function fire(ev){
    switch(ev.type){
    case 'message':
        switch(typeof ev.data){
        case 'string':
            console.log('msg text', ev.data);
        break;
        case 'object':
            ev.data.arrayBuffer().then(ab=>{ console.log(new Uint8Array(ab)); });
        break;
        }
    break;
    default:
        console.log(ev.type ,ev);
    }
}
function hello(msg){
    if(msg === undefined){
        msg = new ArrayBuffer(4);
        const uint = new Uint8Array(msg);
        uint[0] = 4;
        uint[1] = 3;
        uint[2] = 2;
        uint[3] = 1;
    }
    pipe.send(msg);
}

pipe.addEventListener('open', fire);
pipe.addEventListener('close', fire);
pipe.addEventListener('message', fire);
pipe.addEventListener('error', fire);


</script>
</body></html>
`;
});

app.use(router.routes());
app.use(router.allowedMethods());

app.addEventListener('error', (ev)=>{
    console.error(ev);
    debugger;
});

app.addEventListener('listen', (server)=>{
    console.log(`open ${ server.secure ? 'https':'http' }://${ server.hostname }:${ server.port }`);
});
const whenClosed = app.listen(`:${port}`);
await whenClosed;
console.log(`closed http :${port}, bye`);