Compare commits
1 Commits
23062dde58
...
bf7dd6646d
Author | SHA1 | Date |
---|---|---|
Vitaliy Filippov | bf7dd6646d |
62
antietcd.js
62
antietcd.js
|
@ -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 } }));
|
||||||
|
|
|
@ -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,
|
||||||
};
|
};
|
||||||
|
|
41
etctree.js
41
etctree.js
|
@ -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,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in New Issue