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 AntiPersistence = require('./antipersistence.js');
|
||||
const AntiCluster = require('./anticluster.js');
|
||||
const { runCallbacks, RequestError } = require('./common.js');
|
||||
const { runCallbacks, de64, b64, RequestError } = require('./common.js');
|
||||
|
||||
class AntiEtcd extends EventEmitter
|
||||
{
|
||||
|
@ -435,24 +435,64 @@ class AntiEtcd extends EventEmitter
|
|||
// internal handlers
|
||||
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)
|
||||
{
|
||||
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 };
|
||||
}
|
||||
|
||||
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 };
|
||||
}
|
||||
|
||||
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 };
|
||||
}
|
||||
|
||||
|
@ -494,9 +534,15 @@ class AntiEtcd extends EventEmitter
|
|||
const create_request = msg.create_request;
|
||||
if (!create_request.watch_id || !client.watches[create_request.watch_id])
|
||||
{
|
||||
const watch = this.etctree.api_create_watch(
|
||||
{ ...create_request, watch_id: null }, (msg) => socket.send(JSON.stringify(msg))
|
||||
);
|
||||
const req = { ...create_request, watch_id: null };
|
||||
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)
|
||||
{
|
||||
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();
|
||||
}
|
||||
|
||||
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)
|
||||
{
|
||||
const cbs = obj[key];
|
||||
|
@ -34,5 +41,6 @@ function runCallbacks(obj, key, new_value)
|
|||
module.exports = {
|
||||
RequestError,
|
||||
de64,
|
||||
b64,
|
||||
runCallbacks,
|
||||
};
|
||||
|
|
41
etctree.js
41
etctree.js
|
@ -19,7 +19,7 @@ const { RequestError } = require('./common.js');
|
|||
|
||||
class EtcTree
|
||||
{
|
||||
constructor(use_base64)
|
||||
constructor()
|
||||
{
|
||||
this.state = {};
|
||||
this.leases = {};
|
||||
|
@ -27,7 +27,6 @@ class EtcTree
|
|||
this.watcher_id = 0;
|
||||
this.mod_revision = 0;
|
||||
this.compact_revision = 0;
|
||||
this.use_base64 = use_base64;
|
||||
this.replicate = null;
|
||||
this.paused = false;
|
||||
this.active_immediate = [];
|
||||
|
@ -51,23 +50,9 @@ class EtcTree
|
|||
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)
|
||||
{
|
||||
const parts = this._key_parts(this.de64(chk.key));
|
||||
const parts = this._key_parts(chk.key);
|
||||
const { cur } = this._get_subtree(parts, false, false);
|
||||
let check_value, ref_value;
|
||||
if (chk.target === 'MOD')
|
||||
|
@ -118,8 +103,8 @@ class EtcTree
|
|||
|
||||
_get_range(req)
|
||||
{
|
||||
const key = this.de64(req.key);
|
||||
const end = this.de64(req.range_end);
|
||||
const key = req.key;
|
||||
const end = req.range_end;
|
||||
if (end != null && (key[key.length-1] != '/' || end[end.length-1] != '0' ||
|
||||
end.substr(0, end.length-1) !== key.substr(0, key.length-1)))
|
||||
{
|
||||
|
@ -597,9 +582,9 @@ class EtcTree
|
|||
{
|
||||
const ev = {
|
||||
type: cur.value == null ? 'DELETE' : 'PUT',
|
||||
kv: cur.value == null ? { key: this.b64(prefix === null ? '' : prefix) } : {
|
||||
key: this.b64(prefix),
|
||||
value: this.b64(cur.value),
|
||||
kv: cur.value == null ? { key: (prefix === null ? '' : prefix) } : {
|
||||
key: prefix,
|
||||
value: cur.value,
|
||||
mod_revision: cur.mod_revision,
|
||||
},
|
||||
};
|
||||
|
@ -727,9 +712,9 @@ class EtcTree
|
|||
_put(request_put, cur_revision, notifications)
|
||||
{
|
||||
// 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 value = this.de64(request_put.value);
|
||||
const value = request_put.value;
|
||||
const { cur, watchers } = this._get_subtree(parts, true, true);
|
||||
if (cur.key_watchers)
|
||||
{
|
||||
|
@ -762,7 +747,7 @@ class EtcTree
|
|||
cur.create_revision = cur_revision;
|
||||
}
|
||||
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)
|
||||
{
|
||||
notify.lease = cur.lease;
|
||||
|
@ -793,10 +778,10 @@ class EtcTree
|
|||
}
|
||||
if (cur.value != null)
|
||||
{
|
||||
const item = { key: this.b64(prefix === null ? '' : prefix) };
|
||||
const item = { key: (prefix === null ? '' : prefix) };
|
||||
if (!req.keys_only)
|
||||
{
|
||||
item.value = this.b64(cur.value);
|
||||
item.value = cur.value;
|
||||
item.mod_revision = cur.mod_revision;
|
||||
//item.create_revision = cur.create_revision;
|
||||
//item.version = cur.version;
|
||||
|
@ -833,7 +818,7 @@ class EtcTree
|
|||
this.mod_revision = cur_revision;
|
||||
notifications.push({
|
||||
watchers: cur.key_watchers ? [ ...watchers, ...cur.key_watchers ] : watchers,
|
||||
key: this.b64(prefix === null ? '' : prefix),
|
||||
key: (prefix === null ? '' : prefix),
|
||||
mod_revision: cur_revision,
|
||||
});
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue