Compare commits

...

1 Commits

Author SHA1 Message Date
Vitaliy Filippov bf7dd6646d Move base64 handling into antietcd.js 2024-06-08 17:20:47 +03:00
3 changed files with 75 additions and 36 deletions

View File

@ -15,7 +15,7 @@ const ws = require('ws');
const EtcTree = require('./etctree.js'); const EtcTree = require('./etctree.js');
const AntiPersistence = require('./antipersistence.js'); const AntiPersistence = require('./antipersistence.js');
const AntiCluster = require('./anticluster.js'); const AntiCluster = require('./anticluster.js');
const { runCallbacks, RequestError } = require('./common.js'); const { runCallbacks, de64, b64, RequestError } = require('./common.js');
class AntiEtcd extends EventEmitter class AntiEtcd extends EventEmitter
{ {
@ -435,24 +435,64 @@ class AntiEtcd extends EventEmitter
// internal handlers // internal handlers
async _handle_kv_txn(data) async _handle_kv_txn(data)
{ {
return await this.etctree.api_txn(data); if (this.cfg.use_base64)
{
for (const item of data.compare||[])
{
if (item.key != null)
item.key = de64(item.key);
}
for (const items of [ data.success, data.failure ])
{
for (const item of items||[])
{
const req = item.request_range || item.requestRange ||
item.request_put || item.requestPut ||
item.request_delete_range || item.requestDeleteRange;
if (req.key != null)
req.key = de64(req.key);
if (req.range_end != null)
req.range_end = de64(req.range_end);
if (req.value != null)
req.value = de64(req.value);
}
}
}
const result = await this.etctree.api_txn(data);
if (this.cfg.use_base64)
{
for (const item of result.responses||[])
{
if (item.response_range)
{
for (const kv of item.response_range.kvs)
{
if (kv.key != null)
kv.key = b64(kv.key);
if (kv.value != null)
kv.value = b64(kv.value);
}
}
}
}
return result;
} }
async _handle_kv_range(data) async _handle_kv_range(data)
{ {
const r = await this.etctree.api_txn({ success: [ { request_range: data } ] }); const r = await this._handle_kv_txn({ success: [ { request_range: data } ] });
return { header: r.header, ...r.responses[0].response_range }; return { header: r.header, ...r.responses[0].response_range };
} }
async _handle_kv_put(data) async _handle_kv_put(data)
{ {
const r = await this.etctree.api_txn({ success: [ { request_put: data } ] }); const r = await this._handle_kv_txn({ success: [ { request_put: data } ] });
return { header: r.header, ...r.responses[0].response_put }; return { header: r.header, ...r.responses[0].response_put };
} }
async _handle_kv_deleterange(data) async _handle_kv_deleterange(data)
{ {
const r = await this.etctree.api_txn({ success: [ { request_delete_range: data } ] }); const r = await this._handle_kv_txn({ success: [ { request_delete_range: data } ] });
return { header: r.header, ...r.responses[0].response_delete_range }; return { header: r.header, ...r.responses[0].response_delete_range };
} }
@ -494,9 +534,15 @@ class AntiEtcd extends EventEmitter
const create_request = msg.create_request; const create_request = msg.create_request;
if (!create_request.watch_id || !client.watches[create_request.watch_id]) if (!create_request.watch_id || !client.watches[create_request.watch_id])
{ {
const watch = this.etctree.api_create_watch( const req = { ...create_request, watch_id: null };
{ ...create_request, watch_id: null }, (msg) => socket.send(JSON.stringify(msg)) if (this.cfg.use_base64)
); {
if (req.key != null)
req.key = de64(req.key);
if (req.range_end != null)
req.range_end = de64(req.range_end);
}
const watch = this.etctree.api_create_watch(req, (msg) => this.sendToSocket(socket, msg));
if (!watch.created) if (!watch.created)
{ {
socket.send(JSON.stringify({ result: { header: { revision: this.etctree.mod_revision }, watch_id: create_request.watch_id, ...watch } })); socket.send(JSON.stringify({ result: { header: { revision: this.etctree.mod_revision }, watch_id: create_request.watch_id, ...watch } }));

View File

@ -18,6 +18,13 @@ function de64(k)
return Buffer.from(k, 'base64').toString(); return Buffer.from(k, 'base64').toString();
} }
function b64(k)
{
if (k == null) // null or undefined
return k;
return this.use_base64 ? Buffer.from(k).toString('base64') : k;
}
function runCallbacks(obj, key, new_value) function runCallbacks(obj, key, new_value)
{ {
const cbs = obj[key]; const cbs = obj[key];
@ -34,5 +41,6 @@ function runCallbacks(obj, key, new_value)
module.exports = { module.exports = {
RequestError, RequestError,
de64, de64,
b64,
runCallbacks, runCallbacks,
}; };

View File

@ -19,7 +19,7 @@ const { RequestError } = require('./common.js');
class EtcTree class EtcTree
{ {
constructor(use_base64) constructor()
{ {
this.state = {}; this.state = {};
this.leases = {}; this.leases = {};
@ -27,7 +27,6 @@ class EtcTree
this.watcher_id = 0; this.watcher_id = 0;
this.mod_revision = 0; this.mod_revision = 0;
this.compact_revision = 0; this.compact_revision = 0;
this.use_base64 = use_base64;
this.replicate = null; this.replicate = null;
this.paused = false; this.paused = false;
this.active_immediate = []; this.active_immediate = [];
@ -51,23 +50,9 @@ class EtcTree
this.replicate = replicate; this.replicate = replicate;
} }
de64(k)
{
if (k == null) // null or undefined
return k;
return this.use_base64 ? Buffer.from(k, 'base64').toString() : k;
}
b64(k)
{
if (k == null) // null or undefined
return k;
return this.use_base64 ? Buffer.from(k).toString('base64') : k;
}
_check(chk) _check(chk)
{ {
const parts = this._key_parts(this.de64(chk.key)); const parts = this._key_parts(chk.key);
const { cur } = this._get_subtree(parts, false, false); const { cur } = this._get_subtree(parts, false, false);
let check_value, ref_value; let check_value, ref_value;
if (chk.target === 'MOD') if (chk.target === 'MOD')
@ -118,8 +103,8 @@ class EtcTree
_get_range(req) _get_range(req)
{ {
const key = this.de64(req.key); const key = req.key;
const end = this.de64(req.range_end); const end = req.range_end;
if (end != null && (key[key.length-1] != '/' || end[end.length-1] != '0' || if (end != null && (key[key.length-1] != '/' || end[end.length-1] != '0' ||
end.substr(0, end.length-1) !== key.substr(0, key.length-1))) end.substr(0, end.length-1) !== key.substr(0, key.length-1)))
{ {
@ -597,9 +582,9 @@ class EtcTree
{ {
const ev = { const ev = {
type: cur.value == null ? 'DELETE' : 'PUT', type: cur.value == null ? 'DELETE' : 'PUT',
kv: cur.value == null ? { key: this.b64(prefix === null ? '' : prefix) } : { kv: cur.value == null ? { key: (prefix === null ? '' : prefix) } : {
key: this.b64(prefix), key: prefix,
value: this.b64(cur.value), value: cur.value,
mod_revision: cur.mod_revision, mod_revision: cur.mod_revision,
}, },
}; };
@ -727,9 +712,9 @@ class EtcTree
_put(request_put, cur_revision, notifications) _put(request_put, cur_revision, notifications)
{ {
// FIXME: prev_kv, ignore_value(?), ignore_lease(?) // FIXME: prev_kv, ignore_value(?), ignore_lease(?)
const parts = this._key_parts(this.de64(request_put.key)); const parts = this._key_parts(request_put.key);
const key = parts.join('/'); const key = parts.join('/');
const value = this.de64(request_put.value); const value = request_put.value;
const { cur, watchers } = this._get_subtree(parts, true, true); const { cur, watchers } = this._get_subtree(parts, true, true);
if (cur.key_watchers) if (cur.key_watchers)
{ {
@ -762,7 +747,7 @@ class EtcTree
cur.create_revision = cur_revision; cur.create_revision = cur_revision;
} }
cur.value = value; cur.value = value;
const notify = { watchers, key: this.b64(key), value: this.b64(value), mod_revision: cur.mod_revision }; const notify = { watchers, key, value, mod_revision: cur.mod_revision };
if (cur.lease) if (cur.lease)
{ {
notify.lease = cur.lease; notify.lease = cur.lease;
@ -793,10 +778,10 @@ class EtcTree
} }
if (cur.value != null) if (cur.value != null)
{ {
const item = { key: this.b64(prefix === null ? '' : prefix) }; const item = { key: (prefix === null ? '' : prefix) };
if (!req.keys_only) if (!req.keys_only)
{ {
item.value = this.b64(cur.value); item.value = cur.value;
item.mod_revision = cur.mod_revision; item.mod_revision = cur.mod_revision;
//item.create_revision = cur.create_revision; //item.create_revision = cur.create_revision;
//item.version = cur.version; //item.version = cur.version;
@ -833,7 +818,7 @@ class EtcTree
this.mod_revision = cur_revision; this.mod_revision = cur_revision;
notifications.push({ notifications.push({
watchers: cur.key_watchers ? [ ...watchers, ...cur.key_watchers ] : watchers, watchers: cur.key_watchers ? [ ...watchers, ...cur.key_watchers ] : watchers,
key: this.b64(prefix === null ? '' : prefix), key: (prefix === null ? '' : prefix),
mod_revision: cur_revision, mod_revision: cur_revision,
}); });
} }