🎉 pogsocket!

Author: melike2d

.gitignore

@@ -0,0 +1,6 @@
+# Editor Settings
+.vscode/
+.idea/
+
+# Environment Variables
+*.env

deps.ts

@@ -0,0 +1,3 @@
+export * from "https://deno.land/std@0.104.0/ws/mod.ts";
+export * from "https://deno.land/std@0.104.0/io/bufio.ts";
+export * from "https://deno.land/std@0.104.0/async/deferred.ts";

encoding.ts

@@ -0,0 +1,14 @@
+export type Encoder = (input?: string) => Uint8Array;
+
+export function encoder(): Encoder {
+    const _encoder = new TextEncoder();
+    return input => _encoder.encode(input);
+}
+
+export type Decoder = (input?: Uint8Array) => string;
+
+export function decoder(): Decoder {
+    const _decoder = new TextDecoder();
+    return input => _decoder.decode(input);
+}
+

helpers/close_socket.ts

@@ -0,0 +1,62 @@
+import { OpCode } from "../deps.ts";
+import { encoder } from "../encoding.ts";
+import { sendFrame } from "./send_frame.ts";
+
+import type { PogSocket } from "../socket.ts";
+
+const encode = encoder()
+
+/**
+ * Closes the provided pogsocket with a code and/or reason.
+ * @param socket pogsocket to clsoe
+ * @param code The close code
+ * @param reason The close reason.
+ */
+export async function closeSocket(socket: PogSocket, code = 1000, reason?: string) {
+    try { 
+        const header = [ code >>> 8, code & 0x00ff ];
+
+        let payload: Uint8Array;
+        if (reason) {
+            const bytes = encode(reason);
+            payload = new Uint8Array(2 + bytes.byteLength);
+            payload.set(header);
+            payload.set(bytes, 2);
+        } else {
+            payload = new Uint8Array(header);
+        }
+
+        await sendFrame(socket, OpCode.Close, payload);
+    } catch (e) {
+        throw e;
+    } finally {
+        closeSocketConnection(socket);
+    }
+}
+
+/**
+ * Closes the connection of the supplied pogsocket, only for internal use.
+ * @param socket The pogsocket to close.
+ */
+ export function closeSocketConnection(socket: PogSocket): void {
+    if (socket.isClosed) {
+        return;
+    }
+
+    try { 
+        socket.conn.close();
+    } catch (e) {
+        throw e;
+    } finally {
+        socket.isClosed = true;
+        while (socket.frameQueue.length > 0) {
+            const entry = socket.frameQueue.shift();
+            if (!entry) {
+                break;
+            }
+
+            const error = new Deno.errors.ConnectionReset("Socket has been closed.");
+            entry.deferred.reject(error);
+        }
+    }
+}

helpers/create_mask.ts

@@ -0,0 +1,3 @@
+export function createMask(length = 4): Uint8Array {
+    return crypto.getRandomValues(new Uint8Array(length));
+}

helpers/event_based.ts

@@ -0,0 +1,30 @@
+import { readSocket } from "./read_socket.ts";
+
+import type { WebSocketMessage } from "../deps.ts";
+import type { PogSocket } from "../socket.ts";
+
+export interface EventHandlers {
+    message?: (data: WebSocketMessage) => void;
+    close?: (code: number, reason?: string) => void;
+    ping?: (data: Uint8Array) => void;
+    pong?: (data: Uint8Array) => void;
+}
+
+export async function useEvents(socket: PogSocket, handlers: EventHandlers) {
+    for await (const event of readSocket(socket)) {
+        switch (event.type) {
+            case "message":
+                handlers.message?.(event.data);
+                break;
+            case "close":
+                handlers.close?.(event.code, event.reason);
+                break;
+            case "ping":
+                handlers.ping?.(event.data);
+                break;
+            case "pong":
+                handlers.pong?.(event.data);
+                break;
+        }
+    }
+}

helpers/is_binary.ts

@@ -0,0 +1,10 @@
+import { WebSocketMessage } from "../deps.ts";
+
+/**
+ * Whether a websocket message is binary.
+ * @param message The websocket isBinaryMessage
+ * @returns true, if the provided message is binary
+ */
+export function isBinaryMessage(message: WebSocketMessage): message is Uint8Array {
+    return typeof message !== "string";
+}

helpers/read_frame.ts

@@ -0,0 +1,14 @@
+import { readFrame as _readFrame, unmask, WebSocketFrame } from "../deps.ts";
+
+import type { PogSocket } from "../socket.ts";
+
+export async function readFrame(socket: PogSocket): Promise<WebSocketFrame|null> {
+    try {
+        const frame = await _readFrame(socket.reader);
+        unmask(frame.payload, frame.mask);
+    
+        return frame;
+    } catch {
+        return null;
+    }
+}

helpers/read_socket.ts

@@ -0,0 +1,105 @@
+import { OpCode } from "../deps.ts";
+import { decoder } from "../encoding.ts";
+import { closeSocket, closeSocketConnection } from "./close_socket.ts";
+import { readFrame } from "./read_frame.ts";
+import { sendFrame } from "./send_frame.ts";
+
+import type { PogSocket } from "../socket.ts";
+import type { WebSocketFrame } from "../deps.ts";
+
+function getLength(frames: WebSocketFrame[]): number {
+    return frames.reduce((length, frame) => length + frame.payload.length, 0);
+}
+
+function getCloseEvent(frame: WebSocketFrame, decode = decoder()): CloseEvent {
+    return { 
+        // [0x12, 0x34] -> 0x1234
+        code: (frame.payload[0] << 8) | frame.payload[1], 
+        reason: decode(frame.payload.subarray(2, frame.payload.length)), 
+        type: "close"
+    }
+}
+
+export async function* readSocket(socket: PogSocket): AsyncIterableIterator<PogSocketEvent> {
+    const decode = decoder();
+
+    /* handle frames. */
+    let frames: WebSocketFrame[] = [];
+    while (!socket.isClosed) {
+        const frame = await readFrame(socket);
+        if (!frame) {
+            closeSocketConnection(socket);
+            break;
+        }
+
+        switch (frame.opcode) {
+            case OpCode.TextFrame:
+            case OpCode.BinaryFrame:
+            case OpCode.Continue:
+                frames.push(frame);
+
+                /* merge all of the received frames. */
+                if (frame.isLastFrame) {
+                    const message = new Uint8Array(getLength(frames));
+                    let pos = 0;
+                    for (const frame of frames) {
+                        message.set(frame.payload, pos);
+                        pos += frame.payload.length;
+                    }
+
+                    yield {
+                        data: frames[0].opcode === OpCode.TextFrame
+                            ? decode(message)
+                            : message,
+                        type: "message"
+                    }
+
+                    frames = [];
+                }
+
+                break;
+            case OpCode.Close: {
+                const event = getCloseEvent(frame, decode);
+                await closeSocket(socket, event.code, event.reason);
+                yield event;
+                break;
+            }
+            case OpCode.Ping:
+                await sendFrame(socket, OpCode.Pong, frame.payload);
+                yield { type: "ping", data: frame.payload }
+                break;
+            case OpCode.Pong:
+                yield { type: "pong", data: frame.payload }
+                break;
+        }
+    }
+}
+
+export type PogSocketEvent = MessageEvent | PingEvent | PongEvent | CloseEvent;
+export type PogSocketEventType = "message" | "ping" | "pong" | "close";
+
+interface IPogSocketEvent {
+    type: PogSocketEventType;
+}
+
+export interface MessageEvent extends IPogSocketEvent {
+    type: "message";
+    data: string | Uint8Array;
+}
+
+export interface PingEvent extends IPogSocketEvent {
+    type: "ping";
+    data: Uint8Array;
+}
+
+export interface PongEvent extends IPogSocketEvent {
+    type: "pong";
+    data: Uint8Array;
+}
+
+export interface CloseEvent extends IPogSocketEvent {
+    type: "close";
+    code: number;
+    reason?: string;
+}
+

helpers/send_frame.ts

@@ -0,0 +1,15 @@
+import { OpCode, WebSocketMessage } from "../deps.ts";
+import { encoder } from "../encoding.ts";
+import { enqueueFrame, PogSocket } from "../socket.ts";
+import { isBinaryMessage } from "./is_binary.ts";
+
+const encode = encoder()
+
+export function sendFrame(socket: PogSocket, opCode: OpCode, payload: WebSocketMessage) {
+    return enqueueFrame(socket, {
+        mask: socket.mask,
+        isLastFrame: true,
+        opcode: opCode,
+        payload: isBinaryMessage(payload) ? payload : encode(payload)
+    });
+}