Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
PyNetworkTables2Js Client for Webpack projects
import {encode, decode} from 'cbor-js';
export default class NetworkTables {
_pynetworktables2jsAddress;
constructor(pynetworktables2jsAddress) {
this._pynetworktables2jsAddress = pynetworktables2jsAddress;
}
/**
* the active connection to the pynetworktables2js server
* @type {WebSocket || null}
* @private
*/
_socket = null;
_socketConnectionListeners = new Set();
addConnectionListener(listener, immediateNotify = true) {
this._socketConnectionListeners.add(listener);
if (immediateNotify) {
listener(this._socket !== null && this._socket.readyState === WebSocket.OPEN);
}
return () => this._socketConnectionListeners.delete(listener);
}
robotAddress = null;
_robotConnectionListeners = new Set();
addRobotConnectionListener(listener, immediateNotify = true) {
this._robotConnectionListeners.add(listener);
if (immediateNotify) {
listener(this._robotAddress);
}
return () => this._robotConnectionListeners.delete(listener);
}
table = new Map();
_globalUpdateListeners = new Set();
addGlobalUpdateListener(listener, immediateNotify = true) {
this._globalUpdateListeners.add(listener);
if (immediateNotify) {
this.table.forEach((value, key) => listener(key, value, true));
}
return () => this._globalUpdateListeners.delete(listener);
}
connect() {
const socket = this._socket = new WebSocket(`${this._pynetworktables2jsAddress}/networktables/ws`);
socket.binaryType = "arraybuffer";
socket.addEventListener("open", () => {
console.info(`Connected to pynetworktables2js server at ${this._pynetworktables2jsAddress}`);
this._socketConnectionListeners.forEach(it => it(true));
});
socket.addEventListener("message", message => {
const data = decode(message.data);
if (data.r) {
this._robotConnectionListeners.forEach(it => it(this._robotAddress = data.a || null));
} else {
const key = data.k;
const value = data.v;
const isNew = data.n;
this.table.set(key, value);
this._globalUpdateListeners.forEach(listener => listener(key, value, isNew));
}
});
socket.addEventListener("close", () => {
console.info(`Connection to pynetworktables2js server closed`);
this._robotConnectionListeners.forEach(it => it(null));
this._socketConnectionListeners.forEach(it => it(false));
});
socket.addEventListener("error", () => {
console.error(`An error occurred with the connection to ${this._pynetworktables2jsAddress}`);
this._robotConnectionListeners.forEach(it => it(null));
this._socketConnectionListeners.forEach(it => it(false));
});
}
set(key, value) {
this._socket.send(encode({
k: key,
v: value
}));
}
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment