bobbydigitales/dandelion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

working datachannel and inital RPC test

Browse Source
This commit is contained in:
bobbydigitales
2025-04-17 23:54:12 -07:00
parent aade48a7b8
commit f6bf55f13c
5 changed files with 522 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
deno/ddln_server.ts
View File

@@ -9,6 +9,7 @@
import { brotli } from "jsr:@deno-library/compress";
const superLog=true;
const memoryCache = true;
const filepathResponseCache: Map<string, Response> = new Map();
@@ -254,6 +255,8 @@ function connectWebsocket(request: Request) {
return null;
}
superLog && console.log(message);
const dispatchHandler = messageDispatch.get(message?.type)
if (!dispatchHandler) {
console.log("Got message I don't understand: ", event.data);

570
src/PeerManager.ts
View File

@@ -5,6 +5,7 @@
// how? do "perfect negotiation" with bootstrap peer. All logic here moves to BP.
import { generateID } from "IDUtils";
import { log } from "log";
// Use a broadcast channel to only have one peer manager for multiple tabs,
// then we won't need to have a session ID as all queries for a peerID will be coming from the same peer manager
@@ -12,44 +13,226 @@ import { generateID } from "IDUtils";
export class PeerManager {
routingTable: Map<string, string>;
private peers: Map<string, PeerConnection>;
private signaler: Signaler;
peers: Map<string, PeerConnection>;
// private signaler: Signaler;
searchQueryFunctions: Map<string, Function> = new Map();
RPC_remote: Map<string, Function> = new Map();
rpc: { [key: string]: Function } = {};
isBootstrapPeer: boolean = false;
bootstrapPeerConnection: PeerConnection | null = null;
sessionID = generateID();
userID: string;
peerID: string;
websocket: WebSocket | null = null;
bootstrapPeerID: string | null = null;
connectPromise: { resolve: Function, reject: Function } | null = null;
pingPeers: RTCPeerConnection[] = [];
websocketSend(message: any) {
if (!this.websocket) {
throw new Error();
}
let messageJSON = "";
try {
messageJSON = JSON.stringify(message);
} catch (e) {
log(e);
return;
}
log("<-signaler:", message);
this.websocket.send(messageJSON);
}
onWebsocketMessage(event: MessageEvent) {
let messageJSON = event.data;
let message: any = null;
try {
message = JSON.parse(messageJSON);
} catch (e) {
log(e);
throw new Error();
}
log("->signaler:", message);
if (message.type === "hello2") {
if (!this.isBootstrapPeer) {
this.bootstrapPeerID = message.bootstrapPeers[0];
}
this.onHello2Received(this.bootstrapPeerID as string);
}
if (message.type === "peer_message") {
let peerConnection = this.peers.get(message.from);
if (message.message.type === "rtc_description") {
// let existingConnection = this.peers.get(message.from);
// // We're already connected, so delete the existing connection and make a new one.
// if (peerConnection) {
// peerConnection.disconnect();
// this.peers.delete(message.from);
// peerConnection = undefined;
// }
if (!peerConnection) {
peerConnection = this.onConnectRequest(message);
}
}
if (!peerConnection) {
log("Can't find peer for peer message:", message);
return;
}
peerConnection.onWebsocketMessage(message.message);
}
}
onConnectRequest(message: any) {
let remotePeerID = message.from;
let newPeer = new PeerConnection(this, remotePeerID, this.websocketSendPeerMessage.bind(this));
if (this.isBootstrapPeer) {
newPeer.setPolite(false);
}
newPeer.connect();
this.peers.set(remotePeerID, newPeer);
return newPeer;
}
async onHello2Received(bootstrapPeerID: string) {
if (!this.isBootstrapPeer) {
this.bootstrapPeerConnection = await this.connectToPeer(bootstrapPeerID);
}
this.connectPromise?.resolve();
}
async sendHello2() {
this.websocketSend({
type: "hello2",
user_id: this.userID,
// user_name: app.username,
peer_id: this.peerID,
session_id: this.sessionID,
// peer_name: app.peername,
is_bootstrap_peer: this.isBootstrapPeer,
// peer_description: this.rtcPeerDescription
});
}
websocketSendPeerMessage(remotePeerID: string, peerMessage: { type: string; description: RTCSessionDescription; }) {
this.websocketSend({
type: "peer_message",
from: this.peerID,
to: remotePeerID,
from_username: "blah user",
from_peername: "blah peer",
message: peerMessage
});
// let responseMessage = { type: "peer_message",
// from: app.peerID,
// to: data.from,
// from_username: app.username,
// from_peername: app.peername,
// message: { type: "get_posts_for_user", post_ids: postIds, user_id: message.user_id } }
async onConnected(bootstrapPeerID: string) {
this.bootstrapPeerConnection = await this.connect(bootstrapPeerID);
}
constructor(userID: string, peerID: string, isBootstrapPeer: boolean) {
this.isBootstrapPeer = isBootstrapPeer;
this.peers = new Map();
this.routingTable = new Map();
this.userID = userID;
this.peerID = peerID;
this.signaler = new Signaler(userID, peerID, isBootstrapPeer, this.onConnected.bind(this));
// Testing
let dummyPeer = new PeerConnection(this, "dummy_peer", this.signaler);
this.peers.set("dummy_peer", dummyPeer);
}
async connect(remotePeerID: string) {
// Connect to the peer that has the peer id peerID
let peerConnection = new PeerConnection(this, remotePeerID, this.signaler);
await peerConnection.connect();
connect() {
let connectPromise = new Promise((resolve, reject) => {
this.connectPromise = { resolve, reject };
});
try {
this.websocket = new WebSocket(
`wss://${window.location.hostname}:${window.location.port}/ws`,
);
} catch (error: any) {
throw new Error(error.message);
}
this.websocket.onopen = async (event) => {
log("PeerManager:ws:onopen");
this.sendHello2();
};
this.websocket.onmessage = this.onWebsocketMessage.bind(this);
return connectPromise;
// this.signaler = new Signaler(userID, peerID, isBootstrapPeer, this.onConnected.bind(this));
// Testing
// let dummyPeer = new PeerConnection(this, "dummy_peer", this.websocketSendPeerMessage.bind(this));
// this.peers.set("dummy_peer", dummyPeer);
}
async connectToPeer(remotePeerID: string) {
// Connect to the peer that has the peer id remotePeerID.
// TODO how do multiple windows / tabs from the same peer and user work?
// Need to decide if they shold all get a unique connection. A peer should only be requesting and writing
// Data once though, so it probably need to be solved on the client side as the data is shared obv
// Maybe use BroadcastChannel to proxy all calls to peermanager? That will probably really complicate things.
// What if we just user session+peerID for the connections? Then we might have two windows making requests
// For IDs etc, it would probably be best to proxy everything.
// Maybe once we put this logic in a web worker, we'll need an interface to it that works over postMessage
// anyway, and at that point, we could just use that same interface over a broadcastChannel
// let's keep it simple for now and ignore the problem :)
let peerConnection = new PeerConnection(this, remotePeerID, this.websocketSendPeerMessage.bind(this));
this.peers.set(remotePeerID, peerConnection);
await peerConnection.connect();
return peerConnection;
}
onPeerDisconnected(remotePeerID: string) {
let deleted = this.peers.delete(remotePeerID);
if (!deleted) {
throw new Error(`Can't find peer that disconnected ${remotePeerID}`);
}
// TODO: What do we do if we lose connection to the bootstrap peer?
// If we have other connections, it probably doesn't matter.
// Eventually we want the bootstrap peer to be no different than any other peer anyway.
if (remotePeerID === this.bootstrapPeerID) {
this.bootstrapPeerID = null;
this.bootstrapPeerConnection = null;
}
}
async disconnect(remotePeerID: string) {
let peer = this.peers.get(remotePeerID);
if (!peer) {
console.log(`PeerManager.disconnect: couln't find peer ${remotePeerID}`);
log(`PeerManager.disconnect: couln't find peer ${remotePeerID}`);
return;
}
@@ -61,7 +244,7 @@ export class PeerManager {
let peer = this.peers.get(peerID);
if (!peer) {
console.log(`Can't find peer ${peerID}`);
log(`Can't find peer ${peerID}`);
return;
}
@@ -99,15 +282,11 @@ export class PeerManager {
}
onMessage(remotePeerID: string, message: any) {
console.log(remotePeerID, message);
log(remotePeerID, message);
}
}
function log(...args: any[]): void {
for (let arg of args) {
console.log("[LOG]", arg);
}
}
interface Message {
type: string;
@@ -118,125 +297,109 @@ interface Message {
peer_message: any;
}
// Initially this wil be the bootstrap peer, We'll keep a connection to it and it will keep a list of all connected peers.
// Eventually we will replace this with connecting via other peers.
class Signaler {
websocket: WebSocket;
// // Initially this wil be the bootstrap peer, We'll keep a connection to it and it will keep a list of all connected peers.
// // Eventually we will replace this with connecting via other peers.
// class Signaler {
// websocket: WebSocket;
sessionID: string;
userID: string;
peerID: string;
bootstrapPeerID: string = "";
private isBootstrapPeer: boolean = false;
private onConnected: Function;
constructor(userID: string, peerID: string, isBootstrapPeer: boolean, onConnected: Function) {
this.onConnected = onConnected;
this.isBootstrapPeer = isBootstrapPeer;
this.sessionID = generateID();
this.userID = userID;
this.peerID = peerID;
// sessionID: string;
// userID: string;
// peerID: string;
// bootstrapPeerID: string = "";
// private isBootstrapPeer: boolean = false;
// private onConnected: Function;
// peerRoutes: Map<string, PeerConnection> = new Map();
try {
this.websocket = new WebSocket(
`wss://${window.location.hostname}:${window.location.port}/ws`,
);
} catch (error: any) {
throw new Error(error.message);
}
this.websocket.onopen = async (event) => {
log("signaler:ws:onopen");
await this.sendHello2();
};
this.websocket.onmessage = this.onMessage.bind(this);
}
sendPeerMessage(remotePeerID: string, peerMessage: { type: string; description: RTCSessionDescription; }) {
this.send({
type: "peer_message",
from: this.peerID,
to: remotePeerID,
from_username: "blah user",
from_peername: "blah peer",
message: peerMessage
})
// let responseMessage = { type: "peer_message",
// from: app.peerID,
// to: data.from,
// from_username: app.username,
// from_peername: app.peername,
// message: { type: "get_posts_for_user", post_ids: postIds, user_id: message.user_id } }
}
// constructor(userID: string, peerID: string, isBootstrapPeer: boolean, onConnected: Function) {
// this.onConnected = onConnected;
// this.isBootstrapPeer = isBootstrapPeer;
// this.sessionID = generateID();
// this.userID = userID;
// this.peerID = peerID;
async sendHello2() {
this.send({
type: "hello2",
user_id: this.userID,
// user_name: app.username,
peer_id: this.peerID,
session_id: this.sessionID,
// peer_name: app.peername,
is_bootstrap_peer: this.isBootstrapPeer,
// peer_description: this.rtcPeerDescription
});
}
// try {
// this.websocket = new WebSocket(
// `wss://${window.location.hostname}:${window.location.port}/ws`,
// );
// } catch (error: any) {
// throw new Error(error.message);
// }
connect() {
}
// this.websocket.onopen = async (event) => {
// log("signaler:ws:onopen");
// await this.sendHello2();
// };
onMessage(event: MessageEvent) {
let messageJSON = event.data;
let message: any = null;
// this.websocket.onmessage = this.onMessage.bind(this);
// }
try {
message = JSON.parse(messageJSON);
} catch (e) {
console.log(e);
throw new Error();
}
console.log("->signaler:", message);
if (message.type === "hello2") {
if (!this.isBootstrapPeer) {
this.bootstrapPeerID = message.bootstrapPeers[0];
}
this.onConnected(this.bootstrapPeerID);
}
}
send(message: any) {
let messageJSON = "";
// connect() {
// }
try {
messageJSON = JSON.stringify(message);
} catch (e) {
console.log(e);
return;
}
// onMessage(event: MessageEvent) {
// let messageJSON = event.data;
// let message: any = null;
console.log("<-signaler:", message);
// try {
// message = JSON.parse(messageJSON);
// } catch (e) {
// log(e);
// throw new Error();
// }
this.websocket.send(messageJSON);
}
// log("->signaler:", message);
// sendPeerMessage
}
// if (message.type === "hello2") {
// if (!this.isBootstrapPeer) {
// this.bootstrapPeerID = message.bootstrapPeers[0];
// }
// this.onConnected(this.bootstrapPeerID);
// }
// if (message.type == "peer_message") {
// if (message.message.type = "rtc_connect") {
// }
// let connection = this.peerRoutes.get(message.from_peer);
// if (!connection) {
// log("Can't find peer for peer message:", message);
// return;
// }
// connection.onSignalerMessage(message);
// }
// }
// route(remotePeerID:string, peerConnection:PeerConnection) {
// this.peerRoutes.set(remotePeerID, peerConnection)
// }
// // sendPeerMessage
// }
class PeerConnection {
private remotePeerID: string;
private signaler: Signaler;
// private signaler: Signaler;
private peerManager: PeerManager;
private dataChannel: RTCDataChannel | null = null;
private messageHandlers: Map<string, Function> = new Map();
private sendPeerMessage: Function;
private makingOffer: boolean = false;
private ignoreOffer: boolean = false;
private isSettingRemoteAnswerPending: boolean = false;
private polite = true;
// private makingOffer:boolean = false;
// private ignoreOffer:boolean = false;
@@ -260,6 +423,7 @@ class PeerConnection {
{ resolve: Function; reject: Function; functionName: string }
> = new Map();
messageSuperlog: boolean = true;
connectionPromise: { resolve: (value?: unknown) => void; reject: (reason?: any) => void; } | null = null;
async RPCHandler(message: any) {
}
@@ -267,25 +431,100 @@ class PeerConnection {
constructor(
peerManager: PeerManager,
remotePeerID: string,
signaler: Signaler,
sendPeerMessage: Function,
) {
this.sendPeerMessage = sendPeerMessage;
this.peerManager = peerManager;
this.remotePeerID = remotePeerID;
this.signaler = signaler;
// this.signaler = signaler;
// this.signaler.route(remotePeerID, this);
}
setPolite(polite: boolean) {
this.polite = polite;
}
setupDataChannel() {
if (!this.dataChannel) {
throw new Error();
}
this.dataChannel.onopen = (e: any) => {
if (!this.dataChannel) {
throw new Error();
}
log("data channel is open!");
this.send({ type: "hello datachannel", from: this.peerManager.peerID });
// this.dataChannel?.send(`{typeHello datachannel from: ${this.peerManager.peerID}`);
log([...this.peerManager.peers.keys()]);
if (this.peerManager.isBootstrapPeer) {
this.send({ type: 'initial_peers', from: this.peerManager.peerID, peers: [...this.peerManager.peers.keys()].filter(entry => entry !== this.remotePeerID) })
// this.dataChannel.send(JSON.stringify());
}
this.connectionPromise?.resolve();
}
this.dataChannel.onmessage = (e: MessageEvent) => {
log("data channel message: ", e.data)
this.onMessage(e.data);
}
}
async connect() {
let connectionPromise = new Promise((resolve, reject) => this.connectionPromise = { resolve, reject });
this.rtcPeer = new RTCPeerConnection(PeerConnection.config);
this.dataChannel = this.rtcPeer.createDataChannel("ddln_main");
this.rtcPeer.onconnectionstatechange = async (e: any) => {
log("rtcPeer: onconnectionstatechange:", this.rtcPeer?.connectionState)
if (!this.rtcPeer) {
throw new Error("onconnectionstatechange");
}
// When the connection is closed, tell the peer manager that this connection has gone away
if (this.rtcPeer.connectionState === "disconnected") {
this.peerManager.onPeerDisconnected(this.remotePeerID);
}
if (this.rtcPeer.connectionState === "connected") {
// let stats = await this.rtcPeer.getStats();
// for (const stat in stats) {
// log(stat);
// }
}
}
this.rtcPeer.ondatachannel = (e: any) => {
let dataChannel = e.channel;
this.dataChannel = dataChannel as RTCDataChannel;
this.setupDataChannel();
}
if (this.polite) {
this.dataChannel = this.rtcPeer.createDataChannel("ddln_main");
this.setupDataChannel();
}
if (this.rtcPeer === null) {
return;
}
// this.rtcPeer.onicecandidate = ({ candidate }) => this.signaler.send(JSON.stringify({ candidate }));
this.rtcPeer.onicecandidate = ({ candidate }) => console.log(candidate);
// this.rtcPeer.onicecandidate = ({ candidate }) => log(candidate);
this.rtcPeer.onicecandidate = ({ candidate }) => {
log(candidate);
this.sendPeerMessage(this.remotePeerID, { type: "rtc_candidate", candidate: candidate });
}
this.rtcPeer.onnegotiationneeded = async (event) => {
log("on negotiation needed fired");
@@ -293,35 +532,104 @@ class PeerConnection {
throw new Error();
}
let makingOffer = false;
try {
makingOffer = true;
this.makingOffer = true;
await this.rtcPeer.setLocalDescription();
if (!this.rtcPeer.localDescription) {
return;
}
this.signaler.sendPeerMessage(this.remotePeerID, { type: "rtcDescription", description: this.rtcPeer.localDescription });
this.sendPeerMessage(this.remotePeerID, { type: "rtc_description", description: this.rtcPeer.localDescription });
} catch (err) {
console.error(err);
} finally {
makingOffer = false;
this.makingOffer = false;
}
};
return connectionPromise;
}
async onWebsocketMessage(message: any) {
if (message.type == "rtc_connect") {
this.rtcPeer?.setRemoteDescription(message.description);
}
// /*
// let ignoreOffer = false;
// let isSettingRemoteAnswerPending = false;
// signaler.onmessage = async ({ data: { description, candidate } }) => {
if (!this.rtcPeer) {
throw new Error();
}
let description = null;
if (message.type == "rtc_description") {
description = message.description;
}
let candidate = null;
if (message.type == "rtc_candidate") {
candidate = message.candidate;
}
try {
if (description) {
const readyForOffer =
!this.makingOffer &&
(this.rtcPeer.signalingState === "stable" || this.isSettingRemoteAnswerPending);
const offerCollision = description.type === "offer" && !readyForOffer;
this.ignoreOffer = !this.polite && offerCollision;
if (this.ignoreOffer) {
console.warn(">>>>>>>>>>>>>>>>>IGNORING OFFER");
return;
}
this.isSettingRemoteAnswerPending = description.type == "answer";
await this.rtcPeer.setRemoteDescription(description);
this.isSettingRemoteAnswerPending = false;
if (description.type === "offer") {
await this.rtcPeer.setLocalDescription();
this.sendPeerMessage(this.remotePeerID, { type: "rtc_description", description: this.rtcPeer.localDescription });
}
} else if (candidate) {
try {
await this.rtcPeer.addIceCandidate(candidate);
} catch (err) {
if (!this.ignoreOffer) {
throw err;
}
}
}
} catch (err) {
console.error(err);
}
// };
// */
}
async disconnect() {
// this.rtcPeer?.close();
}
send(message: any) {
this.messageSuperlog && console.log("<-", message.type, message);
this.messageSuperlog && log("<-", message.type, message);
let messageJSON = JSON.stringify(message);
// this.dataChannel?.send();
this.dataChannel?.send(messageJSON);
this.onMessage(messageJSON);
// this.onMessage(messageJSON);
}
call(functionName: string, args: any) {
@@ -350,10 +658,10 @@ class PeerConnection {
try {
message = JSON.parse(messageJSON);
} catch (e) {
console.log("PeerConnection.onMessage:", e);
log("PeerConnection.onMessage:", e);
}
this.messageSuperlog && console.log("->", message.type, message);
this.messageSuperlog && log("->", message.type, message);
let type = message.type;
if (type === "rpc_response") {

13
src/log.ts
View File

@@ -17,9 +17,16 @@ export function renderLog() {
}
log.innerText = logLines.join("\n");
}
export function log(message: string) {
console.log(message);
logLines.push(`${new Date().toLocaleTimeString()}: ${message}`);
export function log(...args: any[]): void {
console.log(...args);
let logLine = `[${new Date().toLocaleTimeString()}]: `;
for (let arg of args) {
logLine += (typeof arg === "string" || arg instanceof String) ? arg : JSON.stringify(arg, null, 4);
}
logLines.push(logLine + "\n");
if (logLines.length > logLength) {
logLines = logLines.slice(logLines.length - logLength);
}

93
src/main2.ts
View File

@@ -35,6 +35,8 @@ import { openDatabase, getData, addData, addDataArray, clearData, deleteData, me
import { generateID } from "IDUtils";
import { PeerManager } from "PeerManager";
import {log, renderLog, setLogVisibility} from "log"
// import {PeerConnection} from "webRTC";
// declare let WebTorrent: any;
@@ -122,29 +124,7 @@ function logID(ID: string) {
// }
let logLines: string[] = [];
let logLength = 30;
let logVisible = false;
function renderLog() {
if (!logVisible) {
return;
}
let log = document.getElementById("log");
if (!log) {
throw new Error();
}
log.innerText = logLines.join("\n");
}
function log(message: string) {
console.log(message);
logLines.push(`${new Date().toLocaleTimeString()}: ${message}`);
if (logLines.length > logLength) {
logLines = logLines.slice(logLines.length - logLength);
}
renderLog();
}
@@ -839,6 +819,36 @@ class App {
firstRun = false;
peerManager: PeerManager | null = null;
async connect() {
this.peerManager = new PeerManager(this.userID, this.peerID, this.isBootstrapPeer);
log("*************** before peerManager.connect");
// We use promises here to only return from this call once we're connected to the boostrap peer
// and the datachannel is open.
// Might want to take this a step further and only return once we're connected to an initial set of peers?
// we could return progress information as we connect and have the app subscribe to that?
// Would be lovely to show a little display of peers connecting, whether you're connected directly to a friend's peer etc.
// Basically that live "dandelion" display.
this.peerManager.registerRPC('getPostIDsForUser', (userID: any) => {
return [1, 2, 3, 4, 5]
});
await this.peerManager.connect();
log("*************** after peerManager.connect");
if (!this.isBootstrapPeer) {
let postIDs = await this.peerManager.rpc.getPostIDsForUser(this.peerManager.bootstrapPeerID, this.userID);
console.log("peerManager.rpc.getPostIDsForUser", postIDs);
}
}
getPreferentialUserID() {
return this.router.userID.length !== 0 ? this.router.userID : this.userID;
}
@@ -1338,7 +1348,7 @@ class App {
return;
}
infoElement.style.display == 'none' ? infoElement.style.display = 'block' : infoElement.style.display = 'none';
logVisible = infoElement.style.display == 'block';
setLogVisibility(infoElement.style.display == 'block');
renderLog();
this.lazyCreateQRCode();
(document.querySelector('#qrcode > img') as HTMLImageElement).classList.add('qrcode_image');
@@ -1592,6 +1602,29 @@ class App {
return false;
}
async registerRPCs() {
if (!this.peerManager) {
throw new Error();
}
this.peerManager.registerRPC('ping', (args: any) => {
return {id:this.peerID};
});
if (!this.isBootstrapPeer) {
let pong = await this.peerManager.rpc.ping(this.peerManager.bootstrapPeerID);
console.log(pong);
}
// this.peerManager.registerRPC('getPostIDsForUser', (args: any) => {
// this.sync.getPostsForUser
// });
}
async testPeerManager() {
if (!this.peerManager) {
throw new Error();
@@ -1631,8 +1664,10 @@ class App {
this.userID = this.getUserID();
this.username = this.getUsername();
this.peerManager = new PeerManager(this.userID, this.peerID, this.isBootstrapPeer);
this.testPeerManager();
this.connect();
this.registerRPCs();
// this.testPeerManager();
// let peer: RTCPeerConnection | null = null;
// // if (window.RTCPeerConnection) {
@@ -2105,6 +2140,14 @@ namespace App {
HOME,
CONNECT,
};
// export function connect() {
// throw new Error("Function not implemented.");
// }
// export function connect() {
// throw new Error("Function not implemented.");
// }
}

3
static/index.html
View File

@@ -12,7 +12,8 @@
"imports": {
"db": "/static/db.js",
"IDUtils": "/static/IDUtils.js",
"PeerManager": "/static/PeerManager.js"
"PeerManager": "/static/PeerManager.js",
"log": "/static/log.js"
}
}
</script>