add channel communication

This commit is contained in:
marlonerler
2024-07-04 17:27:46 +02:00
parent 2770d9b07d
commit 9649e3e522
5 changed files with 239 additions and 56 deletions
+24 -22
View File
@@ -1,38 +1,40 @@
export class TwoWayMap<A, B> { import { WS } from "./index";
private mapAtoB = new Map<A, Set<B>>();
private mapBtoA = new Map<B, Set<A>>(); export class SubscriptionMap {
private channelsPerClient = new Map<WS, Set<string>>();
private clientsPerChannel = new Map<string, Set<WS>>();
// prepare // prepare
private prepareA(a: A) { private prepareChannelSet(ws: WS) {
if (this.mapAtoB.has(a)) return; if (this.channelsPerClient.has(ws)) return;
this.mapAtoB.set(a, new Set()); this.channelsPerClient.set(ws, new Set());
} }
private prepareB(b: B) { private prepareClientSet(channel: string) {
if (this.mapBtoA.has(b)) return; if (this.clientsPerChannel.has(channel)) return;
this.mapBtoA.set(b, new Set()); this.clientsPerChannel.set(channel, new Set());
} }
// utility // utility
private getListOnA(a: A) { getChannelList(ws: WS) {
return this.mapAtoB.get(a); return this.channelsPerClient.get(ws);
} }
private getListOnB(b: B) { getClientList(channel: string) {
return this.mapBtoA.get(b); return this.clientsPerChannel.get(channel);
} }
// public // modifications
set(a: A, b: B) { set(ws: WS, channel: string) {
this.prepareA(a); this.prepareChannelSet(ws);
this.prepareB(b); this.prepareClientSet(channel);
this.getListOnA(a)?.add(b); this.getChannelList(ws)?.add(channel);
this.getListOnB(b)?.add(a); this.getClientList(channel)?.add(ws);
} }
delete(a: A, b: B) { delete(ws: WS, channel: string) {
this.getListOnA(a)?.delete(b); this.getChannelList(ws)?.delete(channel);
this.getListOnB(b)?.delete(a); this.getClientList(channel)?.delete(ws);
} }
} }
+10 -2
View File
@@ -1,10 +1,16 @@
import { forgetConnection, processMessage } from "./websocket-handler";
import Path from "path"; import Path from "path";
import { ServerWebSocket } from "bun";
// CONFIG // CONFIG
const staticDir = "static/dist"; const staticDir = "static/dist";
const defaultPage = "index.html"; const defaultPage = "index.html";
const errorPage = "404.html"; const errorPage = "404.html";
// TYPES
export type WS = ServerWebSocket<unknown>;
// SERVER // SERVER
const server = Bun.serve({ const server = Bun.serve({
fetch(request, server) { fetch(request, server) {
@@ -27,9 +33,11 @@ const server = Bun.serve({
websocket: { websocket: {
message(ws, message) { message(ws, message) {
processMessage(ws, message.toString());
},
close(ws, code, reason) {
forgetConnection(ws);
}, },
close(ws, code, reason) {},
}, },
}); });
+54 -3
View File
@@ -1,5 +1,27 @@
(() => { (() => {
// node_modules/bloatless-react/index.ts // node_modules/bloatless-react/index.ts
var State = class {
_value;
_bindings = /* @__PURE__ */ new Set();
constructor(initialValue) {
this._value = initialValue;
}
get value() {
return this._value;
}
set value(newValue) {
if (this._value == newValue) return;
this._value = newValue;
this.callSubscriptions();
}
callSubscriptions() {
this._bindings.forEach((fn) => fn(this._value));
}
subscribe(fn) {
this._bindings.add(fn);
fn(this._value);
}
};
function createElement(tagName, attributes = {}, ...children) { function createElement(tagName, attributes = {}, ...children) {
const element = document.createElement(tagName); const element = document.createElement(tagName);
if (attributes != null) if (attributes != null)
@@ -58,7 +80,36 @@
// src/index.tsx // src/index.tsx
var ws = new WebSocket(`ws://${window.location.host}`); var ws = new WebSocket(`ws://${window.location.host}`);
ws.onopen = () => ws.send("Hello!"); ws.addEventListener("message", (message) => {
ws.onmessage = (data) => console.log(data); receivedMessage.value = message.data;
document.querySelector("main").append(/* @__PURE__ */ createElement("h1", null, "Hello there!")); });
function send(object) {
const stringified = JSON.stringify(object);
ws.send(stringified);
}
var receivedMessage = new State("");
var subscriptionChannel = new State("");
var messageChannel = new State("");
var messageBody = new State("");
function subscribe() {
send({ subscribeChannel: subscriptionChannel.value });
}
function unsubscribe() {
send({ unsubscribeChannel: subscriptionChannel.value });
}
function sendMessage() {
send({
messageChannel: messageChannel.value,
messageBody: messageBody.value
});
}
document.querySelector("main").append(
/* @__PURE__ */ createElement("article", null, /* @__PURE__ */ createElement("header", null, "UDN Test"), /* @__PURE__ */ createElement("div", null, /* @__PURE__ */ createElement("div", { class: "tile" }, /* @__PURE__ */ createElement("div", null, /* @__PURE__ */ createElement("b", null, "Received message"), /* @__PURE__ */ createElement("p", { class: "secondary", "subscribe:innerText": receivedMessage }))), /* @__PURE__ */ createElement("hr", null), /* @__PURE__ */ createElement("label", { class: "tile" }, /* @__PURE__ */ createElement("div", null, /* @__PURE__ */ createElement("span", null, "Subscribe to channel"), /* @__PURE__ */ createElement(
"input",
{
placeholder: "my-channel",
"bind:value": subscriptionChannel
}
))), /* @__PURE__ */ createElement("div", { class: "flex-row width-input justify-end" }, /* @__PURE__ */ createElement("button", { class: "danger width-50", "on:click": unsubscribe }, "Unsubscribe"), /* @__PURE__ */ createElement("button", { class: "primary width-50", "on:click": subscribe }, "Subscribe", /* @__PURE__ */ createElement("span", { class: "icon" }, "arrow_forward"))), /* @__PURE__ */ createElement("hr", null), /* @__PURE__ */ createElement("label", { class: "tile" }, /* @__PURE__ */ createElement("div", null, /* @__PURE__ */ createElement("span", null, "Channel"), /* @__PURE__ */ createElement("input", { placeholder: "my-channel", "bind:value": messageChannel }))), /* @__PURE__ */ createElement("label", { class: "tile" }, /* @__PURE__ */ createElement("div", null, /* @__PURE__ */ createElement("span", null, "Message"), /* @__PURE__ */ createElement("input", { placeholder: "Hello, world!", "bind:value": messageBody }))), /* @__PURE__ */ createElement("div", { class: "flex-row width-input justify-end" }, /* @__PURE__ */ createElement("button", { class: "primary width-50", "on:click": sendMessage }, "Send", /* @__PURE__ */ createElement("span", { class: "icon" }, "arrow_forward")))))
);
})(); })();
+90 -3
View File
@@ -1,8 +1,95 @@
import * as React from "bloatless-react"; import * as React from "bloatless-react";
// WS
const ws = new WebSocket(`ws://${window.location.host}`); const ws = new WebSocket(`ws://${window.location.host}`);
ws.onopen = () => ws.send("Hello!") ws.addEventListener("message", (message) => {
ws.onmessage = (data) => console.log(data); receivedMessage.value = message.data;
});
document.querySelector("main")!.append(<h1>Hello there!</h1>); function send(object: any) {
const stringified = JSON.stringify(object);
ws.send(stringified);
}
// STATE
const receivedMessage = new React.State("");
const subscriptionChannel = new React.State("");
const messageChannel = new React.State("");
const messageBody = new React.State("");
function subscribe() {
send({ subscribeChannel: subscriptionChannel.value });
}
function unsubscribe() {
send({ unsubscribeChannel: subscriptionChannel.value });
}
function sendMessage() {
send({
messageChannel: messageChannel.value,
messageBody: messageBody.value,
});
}
// UI
document.querySelector("main")!.append(
<article>
<header>UDN Test</header>
<div>
<div class="tile">
<div>
<b>Received message</b>
<p class="secondary" subscribe:innerText={receivedMessage}></p>
</div>
</div>
<hr></hr>
<label class="tile">
<div>
<span>Subscribe to channel</span>
<input
placeholder="my-channel"
bind:value={subscriptionChannel}
></input>
</div>
</label>
<div class="flex-row width-input justify-end">
<button class="danger width-50" on:click={unsubscribe}>
Unsubscribe
</button>
<button class="primary width-50" on:click={subscribe}>
Subscribe
<span class="icon">arrow_forward</span>
</button>
</div>
<hr></hr>
<label class="tile">
<div>
<span>Channel</span>
<input placeholder="my-channel" bind:value={messageChannel}></input>
</div>
</label>
<label class="tile">
<div>
<span>Message</span>
<input placeholder="Hello, world!" bind:value={messageBody}></input>
</div>
</label>
<div class="flex-row width-input justify-end">
<button class="primary width-50" on:click={sendMessage}>
Send
<span class="icon">arrow_forward</span>
</button>
</div>
</div>
</article>
);
+61 -26
View File
@@ -1,42 +1,77 @@
// TYPES // TYPES
// requests
interface IncomingMessage {
//sent on new connections
connectionType?: ConnectionType;
import { SubscriptionMap } from "./auxilary";
import { WS } from "./index";
// requests
export interface IncomingMessage {
// subscribing to channel // subscribing to channel
channelSubscription?: string; subscribeChannel?: string;
unsubscribeChannel?: string;
//sending message //sending message
messageChannel?: string; messageChannel?: string;
messageBody?: string; messageBody?: string;
} }
// connections
enum ConnectionType {
Server = "server",
Client = "client",
}
// TRACKING // TRACKING
const serversOnEntireNetwork = new Set<WebSocket>(); const subscriptions = new SubscriptionMap();
const socketsPerChannel = new Map<string, Set<WebSocket>>(); export function trackSubscription(ws: WS, channel: string): void {
const channelsPerSocket = new Map<WebSocket, Set<string>>(); subscriptions.set(ws, channel);
confirmSubscription(ws, channel);
export function trackServer(ws: WebSocket): void {
serversOnEntireNetwork.add(ws);
} }
export function trackSubscription(channel: string, ws: WebSocket): void { export function removeSubscription(ws: WS, channel: string): void {
if (!socketsPerChannel.has(channel)) socketsPerChannel.set(channel, new Set()); subscriptions.delete(ws, channel);
if (!channelsPerSocket.has(ws)) channelsPerSocket.set(ws, new Set()); confirmSubscription(ws, channel);
socketsPerChannel.get(channel)!.add(ws);
channelsPerSocket.get(ws)!.add(channel);
} }
export function forgetConnection(ws: WebSocket): void { export function forgetConnection(ws: WS): void {
serversOnEntireNetwork.delete(ws); const channels = subscriptions.getChannelList(ws);
serversOnEntireNetwork.delete(ws); channels?.forEach((channel) => {
subscriptions.delete(ws, channel);
});
}
// PROCESS MESSAGE
export function processMessage(ws: WS, messageString: string): void {
const messageObject: IncomingMessage = parseMessage(messageString);
if (typeof messageObject.subscribeChannel == "string") {
trackSubscription(ws, messageObject.subscribeChannel);
} else if (typeof messageObject.unsubscribeChannel == "string") {
removeSubscription(ws, messageObject.unsubscribeChannel);
} else if (
typeof messageObject.messageChannel == "string" &&
typeof messageObject.messageBody == "string"
) {
sendMessage(messageObject.messageChannel, messageObject.messageBody);
}
}
// MESSAGING
export function stringifyMessage(messageObject: Object): string {
return JSON.stringify(messageObject);
}
export function parseMessage(messageString: string): Object {
return JSON.parse(messageString);
}
export function sendMessage(channel: string, body: string): void {
const messageObject = { channel, body };
const messageString = stringifyMessage(messageObject);
const subscribers = subscriptions.getClientList(channel);
subscribers?.forEach((ws) => ws.send(messageString));
}
function confirmSubscription(ws: WS, channel: string): void {
const messageObject = {
channel: channel,
subscribed: subscriptions.getChannelList(ws)?.has(channel),
};
const messageString = stringifyMessage(messageObject);
ws.send(messageString);
} }