baraba/Baradb
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix/nodebara compatibility (#2)

Browse Source
This commit is contained in:
Dimitar Gigov
2026-05-14 02:33:46 +03:00
committed by GitHub
parent a0d2ca7776 71dcffecce
commit 18f3c16b2a
3 changed files with 99 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
clients/javascript/baradb.js
+78 -48
View File
@@ -244,6 +244,8 @@ class Client {
this.requestId = 0;
this._buffer = Buffer.alloc(0);
this._pendingResolve = null;
this._requestQueue = [];
this._requestLock = false;
}
async connect() {
@@ -478,70 +480,98 @@ class Client {
throw new Error(`Unexpected auth response: 0x${header.kind.toString(16)}`);
}
async _processQueue() {
if (this._requestLock || this._requestQueue.length === 0) return;
this._requestLock = true;
const { task, resolve, reject } = this._requestQueue.shift();
try {
const result = await task();
resolve(result);
} catch (err) {
reject(err);
} finally {
this._requestLock = false;
setImmediate(() => this._processQueue());
}
}
_enqueue(task) {
return new Promise((resolve, reject) => {
this._requestQueue.push({ task, resolve, reject });
this._processQueue();
});
}
async ping() {
if (!this.connected) throw new Error('Not connected');
const msg = this._build(MsgKind.PING, Buffer.alloc(0));
this.socket.write(msg);
return this._enqueue(async () => {
const msg = this._build(MsgKind.PING, Buffer.alloc(0));
this.socket.write(msg);
const header = await this._readHeader();
if (header.kind === MsgKind.PONG) return true;
if (header.kind === MsgKind.ERROR) throw await this._readError(header.length);
return false;
const header = await this._readHeader();
if (header.kind === MsgKind.PONG) return true;
if (header.kind === MsgKind.ERROR) throw await this._readError(header.length);
return false;
});
}
async query(sql) {
if (!this.connected) throw new Error('Not connected');
const queryBuf = Buffer.from(sql, 'utf-8');
const payload = Buffer.alloc(4 + queryBuf.length + 1);
payload.writeUInt32BE(queryBuf.length, 0);
queryBuf.copy(payload, 4);
payload[4 + queryBuf.length] = ResultFormat.BINARY;
return this._enqueue(async () => {
const queryBuf = Buffer.from(sql, 'utf-8');
const payload = Buffer.alloc(4 + queryBuf.length + 1);
payload.writeUInt32BE(queryBuf.length, 0);
queryBuf.copy(payload, 4);
payload[4 + queryBuf.length] = ResultFormat.BINARY;
const msg = this._build(MsgKind.QUERY, payload);
this.socket.write(msg);
const msg = this._build(MsgKind.QUERY, payload);
this.socket.write(msg);
const header = await this._readHeader();
if (header.kind === MsgKind.ERROR) throw await this._readError(header.length);
if (header.kind === MsgKind.DATA) return await this._readDataResponse(header.length);
if (header.kind === MsgKind.COMPLETE) {
const data = await this._recvExact(header.length);
const result = new QueryResult();
result.affectedRows = data.readUInt32BE(0);
return result;
}
return new QueryResult();
const header = await this._readHeader();
if (header.kind === MsgKind.ERROR) throw await this._readError(header.length);
if (header.kind === MsgKind.DATA) return await this._readDataResponse(header.length);
if (header.kind === MsgKind.COMPLETE) {
const data = await this._recvExact(header.length);
const result = new QueryResult();
result.affectedRows = data.readUInt32BE(0);
return result;
}
return new QueryResult();
});
}
async queryParams(sql, params = []) {
if (!this.connected) throw new Error('Not connected');
const queryBuf = Buffer.from(sql, 'utf-8');
const paramParts = [];
for (const p of params) {
paramParts.push(p.serialize());
}
const paramsBuf = Buffer.concat(paramParts);
return this._enqueue(async () => {
const queryBuf = Buffer.from(sql, 'utf-8');
const paramParts = [];
for (const p of params) {
paramParts.push(p.serialize());
}
const paramsBuf = Buffer.concat(paramParts);
const payload = Buffer.alloc(4 + queryBuf.length + 1 + 4 + paramsBuf.length);
let pos = 0;
payload.writeUInt32BE(queryBuf.length, pos); pos += 4;
queryBuf.copy(payload, pos); pos += queryBuf.length;
payload[pos] = ResultFormat.BINARY; pos++;
payload.writeUInt32BE(params.length, pos); pos += 4;
paramsBuf.copy(payload, pos);
const payload = Buffer.alloc(4 + queryBuf.length + 1 + 4 + paramsBuf.length);
let pos = 0;
payload.writeUInt32BE(queryBuf.length, pos); pos += 4;
queryBuf.copy(payload, pos); pos += queryBuf.length;
payload[pos] = ResultFormat.BINARY; pos++;
payload.writeUInt32BE(params.length, pos); pos += 4;
paramsBuf.copy(payload, pos);
const msg = this._build(MsgKind.QUERY_PARAMS, payload);
this.socket.write(msg);
const msg = this._build(MsgKind.QUERY_PARAMS, payload);
this.socket.write(msg);
const header = await this._readHeader();
if (header.kind === MsgKind.ERROR) throw await this._readError(header.length);
if (header.kind === MsgKind.DATA) return await this._readDataResponse(header.length);
if (header.kind === MsgKind.COMPLETE) {
const data = await this._recvExact(header.length);
const result = new QueryResult();
result.affectedRows = data.readUInt32BE(0);
return result;
}
return new QueryResult();
const header = await this._readHeader();
if (header.kind === MsgKind.ERROR) throw await this._readError(header.length);
if (header.kind === MsgKind.DATA) return await this._readDataResponse(header.length);
if (header.kind === MsgKind.COMPLETE) {
const data = await this._recvExact(header.length);
const result = new QueryResult();
result.affectedRows = data.readUInt32BE(0);
return result;
}
return new QueryResult();
});
}
async execute(sql) {
src/barabadb/core/gossip.nim
+6 -9
View File
@@ -4,6 +4,7 @@ import std/random
import std/monotimes
import std/asyncdispatch
import std/net
import std/asyncnet
import std/strutils
import std/streams
import std/os
@@ -36,7 +37,7 @@ type
fanout*: int # number of nodes to gossip to per round
gossipPort*: int
running*: bool
sock*: Socket
sock*: AsyncSocket
onJoin*: proc(node: GossipNode) {.gcsafe.}
onLeave*: proc(nodeId: string) {.gcsafe.}
onSuspect*: proc(nodeId: string) {.gcsafe.}
@@ -305,25 +306,21 @@ proc startGossipRound*(gp: GossipProtocol, intervalMs: int = 2000) {.async.} =
gp.broadcastGossip()
proc startGossipListener*(gp: GossipProtocol) {.async.} =
gp.sock = newSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)
gp.sock = newAsyncSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)
gp.sock.setSockOpt(OptReuseAddr, true)
gp.sock.bindAddr(Port(gp.gossipPort))
gp.running = true
while gp.running:
try:
var data = newString(65535)
var senderAddr = ""
var senderPort: Port
let bytesRead = gp.sock.recvFrom(data, 65535, senderAddr, senderPort)
if bytesRead > 0:
data.setLen(bytesRead)
let (data, senderAddr, senderPort) = await gp.sock.recvFrom(65535)
if data.len > 0:
gp.handleIncomingGossip(data, senderAddr)
except:
# Small sleep on error to avoid spin
gp.sock.close()
# Recreate socket for next iteration
try:
gp.sock = newSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)
gp.sock = newAsyncSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)
gp.sock.setSockOpt(OptReuseAddr, true)
gp.sock.bindAddr(Port(gp.gossipPort))
except:
src/barabadb/protocol/wire.nim
+15 -6
View File
@@ -169,12 +169,14 @@ proc serializeValue*(buf: var seq[byte], val: WireValue) =
of fkFloat32:
var fl = val.float32Val
var bytes: array[4, byte]
copyMem(addr bytes, unsafeAddr fl, 4)
var i32 = cast[int32](fl)
bigEndian32(addr bytes, unsafeAddr i32)
buf.add(bytes)
of fkFloat64:
var fl = val.float64Val
var bytes: array[8, byte]
copyMem(addr bytes, unsafeAddr fl, 8)
var i64 = cast[int64](fl)
bigEndian64(addr bytes, unsafeAddr i64)
buf.add(bytes)
of fkString: buf.writeString(val.strVal)
of fkBytes: buf.writeBytes(val.bytesVal)
@@ -192,7 +194,8 @@ proc serializeValue*(buf: var seq[byte], val: WireValue) =
for f in val.vecVal:
var fl = f
var bytes: array[4, byte]
copyMem(addr bytes, unsafeAddr fl, 4)
var i32 = cast[int32](fl)
bigEndian32(addr bytes, unsafeAddr i32)
buf.add(bytes)
of fkJson:
buf.writeString(val.jsonVal)
@@ -227,14 +230,18 @@ proc deserializeValue*(buf: openArray[byte], pos: var int, depth: int = 0): Wire
var fl: float32
var bytes: array[4, byte]
for i in 0..3: bytes[i] = buf[pos + i]
copyMem(addr fl, unsafeAddr bytes, 4)
var i32: int32
bigEndian32(addr i32, unsafeAddr bytes)
fl = cast[float32](i32)
pos += 4
result = WireValue(kind: fkFloat32, float32Val: fl)
of fkFloat64:
var fl: float64
var bytes: array[8, byte]
for i in 0..7: bytes[i] = buf[pos + i]
copyMem(addr fl, unsafeAddr bytes, 8)
var i64: int64
bigEndian64(addr i64, unsafeAddr bytes)
fl = cast[float64](i64)
pos += 8
result = WireValue(kind: fkFloat64, float64Val: fl)
of fkString:
@@ -270,7 +277,9 @@ proc deserializeValue*(buf: openArray[byte], pos: var int, depth: int = 0): Wire
var fl: float32
var bytes: array[4, byte]
for j in 0..3: bytes[j] = buf[pos + j]
copyMem(addr fl, unsafeAddr bytes, 4)
var i32: int32
bigEndian32(addr i32, unsafeAddr bytes)
fl = cast[float32](i32)
pos += 4
vec.add(fl)
result = WireValue(kind: fkVector, vecVal: vec)