Compare commits
4 Commits
396a03c229
...
master
Author | SHA1 | Date | |
---|---|---|---|
|
f41bfeaa65 | ||
|
31f5adfcc0 | ||
|
95f3c821a5 | ||
|
9cab041af4 |
@@ -37,3 +37,20 @@ CRDTNotes::Doc* CRDTNotes::addDoc(const CRDTAgent& self_agent, const DocID& id)
|
||||
return &doc;
|
||||
}
|
||||
|
||||
void CRDTNotes::writeLockRelease(const DocID& id) {
|
||||
assert(_doc_write_locks.count(id) > 0);
|
||||
_doc_write_locks.erase(id);
|
||||
}
|
||||
|
||||
bool CRDTNotes::isWriteLocked(const DocID& id) const {
|
||||
return _doc_write_locks.count(id);
|
||||
}
|
||||
|
||||
std::optional<CRDTNotes::DocWriteLock> CRDTNotes::writeLockAquire(const DocID& id) {
|
||||
if (_doc_write_locks.count(id)) {
|
||||
return std::nullopt; // replace with exception instead?
|
||||
}
|
||||
|
||||
_doc_write_locks.emplace(id);
|
||||
return DocWriteLock{*this, id};
|
||||
}
|
||||
|
@@ -6,6 +6,8 @@
|
||||
#include <cstdint>
|
||||
#include <functional>
|
||||
#include <unordered_map>
|
||||
#include <unordered_set>
|
||||
#include <optional>
|
||||
|
||||
using ID32 = std::array<uint8_t, 32>;
|
||||
|
||||
@@ -35,11 +37,26 @@ class CRDTNotes {
|
||||
struct Frontier { // newest known seq for given agent
|
||||
CRDTAgent agent;
|
||||
uint64_t seq{0};
|
||||
uint64_t del_num{0};
|
||||
};
|
||||
|
||||
// RAII lock wrapper
|
||||
struct DocWriteLock {
|
||||
CRDTNotes* notes;
|
||||
DocID id;
|
||||
|
||||
// ctr assumes lock
|
||||
DocWriteLock(CRDTNotes& notes, const DocID& id) : notes(¬es), id(id) {}
|
||||
DocWriteLock(const DocWriteLock&) = delete;
|
||||
DocWriteLock(DocWriteLock&& other) : notes(other.notes), id(other.id) { other.notes = nullptr; }
|
||||
~DocWriteLock(void) { if (notes) { notes->writeLockRelease(id); } }
|
||||
bool operator==(const DocWriteLock& other) const { return id == other.id; }
|
||||
};
|
||||
|
||||
private:
|
||||
// TODO: add metadata to docs
|
||||
std::unordered_map<DocID, Doc> _docs;
|
||||
std::unordered_set<DocID> _doc_write_locks;
|
||||
|
||||
public:
|
||||
// config?
|
||||
@@ -52,5 +69,16 @@ class CRDTNotes {
|
||||
Doc* getDoc(const DocID& id);
|
||||
|
||||
Doc* addDoc(const CRDTAgent& self_agent, const DocID& doc);
|
||||
|
||||
void writeLockRelease(const DocID& id);
|
||||
bool isWriteLocked(const DocID& id) const;
|
||||
std::optional<DocWriteLock> writeLockAquire(const DocID& id);
|
||||
};
|
||||
|
||||
template<>
|
||||
struct std::hash<CRDTNotes::DocWriteLock> {
|
||||
std::uint64_t operator()(const CRDTNotes::DocWriteLock& s) const noexcept {
|
||||
return std::hash<ID32>{}(s.id);
|
||||
}
|
||||
};
|
||||
|
||||
|
@@ -19,7 +19,7 @@ struct CRDTNotesContactSyncModelI {
|
||||
virtual void SendGossip(
|
||||
ContactHandle4 c,
|
||||
const CRDTNotes::DocID& doc_id,
|
||||
const std::vector<CRDTNotes::Frontier>& selected_frontier
|
||||
const std::vector<CRDTNotes::Frontier>& frontier
|
||||
) = 0;
|
||||
|
||||
// fetch
|
||||
@@ -31,7 +31,7 @@ struct CRDTNotesContactSyncModelI {
|
||||
) = 0;
|
||||
|
||||
// action range request
|
||||
virtual void SendFetchOps(
|
||||
virtual void SendFetchAddRange(
|
||||
ContactHandle4 c,
|
||||
const CRDTNotes::DocID& doc_id,
|
||||
const CRDTNotes::CRDTAgent& agent,
|
||||
@@ -39,6 +39,12 @@ struct CRDTNotesContactSyncModelI {
|
||||
const uint64_t seq_to
|
||||
) = 0;
|
||||
|
||||
virtual void SendFetchDel(
|
||||
ContactHandle4 c,
|
||||
const CRDTNotes::DocID& doc_id,
|
||||
const CRDTNotes::CRDTAgent& agent
|
||||
) = 0;
|
||||
|
||||
public: // ops response
|
||||
virtual void SendOps(
|
||||
ContactHandle4 c,
|
||||
|
@@ -5,10 +5,23 @@
|
||||
|
||||
#include <solanaceae/contact/components.hpp>
|
||||
|
||||
#include <entt/container/dense_set.hpp>
|
||||
|
||||
#include <cstdint>
|
||||
#include <vector>
|
||||
#include <iostream>
|
||||
|
||||
namespace Components {
|
||||
|
||||
// attached to contact
|
||||
struct OpSendQueue {
|
||||
std::map<CRDTNotes::DocID, std::vector<CRDTNotes::Doc::Op>> ops;
|
||||
// HACK: limit to 5 ops per packet for now
|
||||
// TODO: ft based alternative for >5 ops
|
||||
};
|
||||
|
||||
} // Components
|
||||
|
||||
static ID32 id_from_vec(const std::vector<uint8_t>& vec) {
|
||||
ID32 new_id;
|
||||
for (size_t i = 0; i < new_id.size() && i < vec.size(); i++) {
|
||||
@@ -27,6 +40,25 @@ CRDTNotesSync::~CRDTNotesSync(void) {
|
||||
}
|
||||
|
||||
float CRDTNotesSync::iterate(float time_delta) {
|
||||
for (auto doc_it = _docs_incoming_ops.begin(); doc_it != _docs_incoming_ops.end();) {
|
||||
if (_notes.isWriteLocked(doc_it->first)) {
|
||||
doc_it++;
|
||||
continue;
|
||||
}
|
||||
|
||||
auto lock_opt = _notes.writeLockAquire(doc_it->first);
|
||||
assert(lock_opt);
|
||||
|
||||
auto* doc_ptr = getDoc(doc_it->first);
|
||||
// TODO: record every applied op and throw event, so eg gui can react better
|
||||
// , or better yet, edit events in string space (imgui can consume them)
|
||||
doc_ptr->apply(doc_it->second);
|
||||
|
||||
std::cout << "CRDTNotesSync: applied " << doc_it->second.size() << " ops\n";
|
||||
|
||||
doc_it = _docs_incoming_ops.erase(doc_it);
|
||||
}
|
||||
|
||||
if (!_gossip_queue.empty()) {
|
||||
// TODO: set is sorted by id, not by order added
|
||||
// only one per iterate *should* be enough
|
||||
@@ -89,7 +121,40 @@ float CRDTNotesSync::iterate(float time_delta) {
|
||||
_fetch_frontier_queue.erase(it);
|
||||
}
|
||||
|
||||
return 2.f;
|
||||
bool sending_ops {false};
|
||||
{ // send ops in queue
|
||||
std::vector<Contact4> empty_queue;
|
||||
for (const auto& [c, op_comp, sync_model] : _cs.registry().view<Components::OpSendQueue, CRDTNotesContactSyncModelI*>().each()) {
|
||||
// HACK: one pkg with up to 5 ops per tick per peer
|
||||
//for (const auto& [doc_id, op_vec] : op_comp.ops) {
|
||||
for (auto it = op_comp.ops.begin(); it != op_comp.ops.end();) {
|
||||
if (it->second.empty()) {
|
||||
it = op_comp.ops.erase(it);
|
||||
continue;
|
||||
} else if (it->second.size() <= 5) {
|
||||
std::cout << "sending " << it->second.size() << " ops\n";
|
||||
sync_model->SendOps(_cs.contactHandle(c), it->first, it->second);
|
||||
it = op_comp.ops.erase(it);
|
||||
//sending_ops = true;
|
||||
} else {
|
||||
std::vector<CRDTNotes::Doc::Op> tmp_ops {it->second.cbegin(), it->second.cbegin()+5};
|
||||
assert(tmp_ops.size() == 5);
|
||||
sync_model->SendOps(_cs.contactHandle(c), it->first, tmp_ops);
|
||||
it->second.erase(it->second.cbegin(), it->second.cbegin()+5);
|
||||
sending_ops = true;
|
||||
}
|
||||
|
||||
break; // single update only
|
||||
}
|
||||
|
||||
if (op_comp.ops.empty()) {
|
||||
empty_queue.push_back(c);
|
||||
}
|
||||
}
|
||||
_cs.registry().remove<Components::OpSendQueue>(empty_queue.cbegin(), empty_queue.cend());
|
||||
}
|
||||
|
||||
return sending_ops ? 0.05f : 2.f;
|
||||
}
|
||||
|
||||
CRDTNotes::Doc* CRDTNotesSync::getDoc(const CRDTNotes::DocID& doc_id) {
|
||||
@@ -203,7 +268,50 @@ void CRDTNotesSync::merge(const CRDTNotes::DocID& doc_id, std::string_view new_t
|
||||
auto op_vec = doc_ptr->merge(new_text);
|
||||
std::cout << "doc changed " << op_vec.size() << " ops generated\n";
|
||||
|
||||
// USE OPS
|
||||
// attach OpSendQueue to every contact
|
||||
// needs to be placed at the contact with the sync model
|
||||
entt::dense_set<Contact4> handled_contacts;
|
||||
for (auto c : _docs_contacts.at(doc_id)) {
|
||||
if (handled_contacts.contains(c)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!c.all_of<CRDTNotesContactSyncModelI*>()) {
|
||||
|
||||
// TODO: this is a fallback, remove
|
||||
if (c.all_of<Contact::Components::ParentOf>()) {
|
||||
for (const auto child : c.get<Contact::Components::ParentOf>().subs) {
|
||||
if (handled_contacts.contains(child)) {
|
||||
continue;
|
||||
}
|
||||
if (c.registry()->all_of<Contact::Components::TagSelfStrong>(child)) {
|
||||
continue;
|
||||
}
|
||||
if (!c.registry()->all_of<CRDTNotesContactSyncModelI*>(child)) {
|
||||
std::cerr << "CRDTNotesSync error: fallback failed\n";
|
||||
continue;
|
||||
}
|
||||
|
||||
auto& op_queue = c.registry()->get_or_emplace<Components::OpSendQueue>(child).ops[doc_id];
|
||||
if (op_queue.empty()) {
|
||||
op_queue = op_vec;
|
||||
} else {
|
||||
op_queue.insert(op_queue.cend(), op_vec.cbegin(), op_vec.cend());
|
||||
}
|
||||
handled_contacts.emplace(child);
|
||||
}
|
||||
}
|
||||
continue; // skip, not impl
|
||||
}
|
||||
|
||||
auto& op_queue = c.get_or_emplace<Components::OpSendQueue>().ops[doc_id];
|
||||
if (op_queue.empty()) {
|
||||
op_queue = op_vec;
|
||||
} else {
|
||||
op_queue.insert(op_queue.cend(), op_vec.cbegin(), op_vec.cend());
|
||||
}
|
||||
handled_contacts.emplace(c);
|
||||
}
|
||||
}
|
||||
|
||||
void CRDTNotesSync::onCRDTNSyncEvent(Events::NGCEXT_crdtns_gossip&& e) {
|
||||
@@ -216,9 +324,38 @@ void CRDTNotesSync::onCRDTNSyncEvent(Events::NGCEXT_crdtns_gossip_frontier&& e)
|
||||
void CRDTNotesSync::onCRDTNSyncEvent(Events::NGCEXT_crdtns_fetch_complete_frontier&& e) {
|
||||
}
|
||||
|
||||
void CRDTNotesSync::onCRDTNSyncEvent(Events::NGCEXT_crdtns_fetch_op_range&& e) {
|
||||
void CRDTNotesSync::onCRDTNSyncEvent(Events::NGCEXT_crdtns_fetch_add_range&& e) {
|
||||
}
|
||||
|
||||
void CRDTNotesSync::onCRDTNSyncEvent(Events::NGCEXT_crdtns_fetch_del&& e) {
|
||||
}
|
||||
|
||||
void CRDTNotesSync::onCRDTNSyncEvent(Events::NGCEXT_crdtns_ops&& e) {
|
||||
addDoc(e.doc_id, e.c);
|
||||
|
||||
if (e.ops.empty()) {
|
||||
std::cerr << "CRDTNotesSync warning: got empty ops event/pkg\n";
|
||||
return;
|
||||
}
|
||||
|
||||
// TODO: deduplicate ops ?
|
||||
auto lock_opt = _notes.writeLockAquire(e.doc_id);
|
||||
if (lock_opt) {
|
||||
// TODO: perms n stuff
|
||||
// TODO: check if seq missing
|
||||
auto* doc_ptr = getDoc(e.doc_id);
|
||||
// TODO: record every applied op and throw event, so eg gui can react better
|
||||
// , or better yet, edit events in string space (imgui can consume them)
|
||||
doc_ptr->apply(e.ops);
|
||||
|
||||
// TODO: check if new frontier
|
||||
} else {
|
||||
auto& op_in_vec = _docs_incoming_ops[e.doc_id];
|
||||
if (op_in_vec.empty()) {
|
||||
op_in_vec = e.ops;
|
||||
} else {
|
||||
op_in_vec.insert(op_in_vec.cend(), e.ops.cbegin(), e.ops.cend());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
@@ -27,11 +27,12 @@ namespace Events {
|
||||
// - array [
|
||||
// - AgentID
|
||||
// - seq (frontier)
|
||||
// - del_num
|
||||
// - ]
|
||||
struct NGCEXT_crdtns_gossip_frontier {
|
||||
ContactHandle4 c;
|
||||
CRDTNotes::DocID doc_id;
|
||||
std::vector<CRDTNotes::Frontier> selected_frontier;
|
||||
std::vector<CRDTNotes::Frontier> frontier;
|
||||
};
|
||||
|
||||
// - DocID
|
||||
@@ -44,7 +45,7 @@ namespace Events {
|
||||
// - AgentID
|
||||
// - seq_from
|
||||
// - seq_to
|
||||
struct NGCEXT_crdtns_fetch_op_range {
|
||||
struct NGCEXT_crdtns_fetch_add_range {
|
||||
ContactHandle4 c;
|
||||
CRDTNotes::DocID doc_id;
|
||||
CRDTNotes::CRDTAgent agent;
|
||||
@@ -52,6 +53,14 @@ namespace Events {
|
||||
uint64_t seq_to;
|
||||
};
|
||||
|
||||
// - DocID
|
||||
// - AgentID
|
||||
struct NGCEXT_crdtns_fetch_del {
|
||||
ContactHandle4 c;
|
||||
CRDTNotes::DocID doc_id;
|
||||
CRDTNotes::CRDTAgent agent;
|
||||
};
|
||||
|
||||
// - DocID
|
||||
// - array [
|
||||
// - op
|
||||
@@ -76,7 +85,8 @@ struct CRDTNotesEventI {
|
||||
virtual void onCRDTNSyncEvent(Events::NGCEXT_crdtns_gossip&& e) = 0;
|
||||
virtual void onCRDTNSyncEvent(Events::NGCEXT_crdtns_gossip_frontier&& e) = 0;
|
||||
virtual void onCRDTNSyncEvent(Events::NGCEXT_crdtns_fetch_complete_frontier&& e) = 0;
|
||||
virtual void onCRDTNSyncEvent(Events::NGCEXT_crdtns_fetch_op_range&& e) = 0;
|
||||
virtual void onCRDTNSyncEvent(Events::NGCEXT_crdtns_fetch_add_range&& e) = 0;
|
||||
virtual void onCRDTNSyncEvent(Events::NGCEXT_crdtns_fetch_del&& e) = 0;
|
||||
virtual void onCRDTNSyncEvent(Events::NGCEXT_crdtns_ops&& e) = 0;
|
||||
};
|
||||
|
||||
@@ -98,6 +108,9 @@ class CRDTNotesSync final : public CRDTNotesEventI {
|
||||
};
|
||||
std::unordered_map<CRDTNotes::DocID, std::map<ContactHandle4, Peer>> _docs_peers;
|
||||
|
||||
// queue of unapplied ops, kept here until write lock can be aquired
|
||||
std::unordered_map<CRDTNotes::DocID, std::vector<CRDTNotes::Doc::Op>> _docs_incoming_ops;
|
||||
|
||||
// if a doc is eg new, it is added here
|
||||
std::set<CRDTNotes::DocID> _gossip_queue; // TODO: no
|
||||
std::set<CRDTNotes::DocID> _fetch_frontier_queue;
|
||||
@@ -130,7 +143,8 @@ class CRDTNotesSync final : public CRDTNotesEventI {
|
||||
void onCRDTNSyncEvent(Events::NGCEXT_crdtns_gossip&& e) override;
|
||||
void onCRDTNSyncEvent(Events::NGCEXT_crdtns_gossip_frontier&& e) override;
|
||||
void onCRDTNSyncEvent(Events::NGCEXT_crdtns_fetch_complete_frontier&& e) override;
|
||||
void onCRDTNSyncEvent(Events::NGCEXT_crdtns_fetch_op_range&& e) override;
|
||||
void onCRDTNSyncEvent(Events::NGCEXT_crdtns_fetch_add_range&& e) override;
|
||||
void onCRDTNSyncEvent(Events::NGCEXT_crdtns_fetch_del&& e) override;
|
||||
void onCRDTNSyncEvent(Events::NGCEXT_crdtns_ops&& e) override;
|
||||
};
|
||||
|
||||
|
@@ -9,7 +9,6 @@
|
||||
#include <imgui.h>
|
||||
#include <misc/cpp/imgui_stdlib.h>
|
||||
|
||||
#include <iostream>
|
||||
#include <cassert>
|
||||
|
||||
namespace detail {
|
||||
@@ -48,6 +47,12 @@ namespace detail {
|
||||
} // detail
|
||||
|
||||
|
||||
std::unordered_set<CRDTNotes::DocWriteLock>::iterator CRDTNotesImGui::findLock(const CRDTNotes::DocID& doc_id) {
|
||||
auto it = _held_locks.begin();
|
||||
for (; it != _held_locks.end() && it->id != doc_id; it++) {}
|
||||
return it;
|
||||
}
|
||||
|
||||
CRDTNotesImGui::CRDTNotesImGui(CRDTNotes& notes, CRDTNotesSync& notes_sync, ContactStore4I& cs) : _notes(notes), _notes_sync(notes_sync), _cs(cs) {
|
||||
}
|
||||
|
||||
@@ -142,17 +147,36 @@ bool CRDTNotesImGui::renderDoc(const CRDTNotes::DocID& doc_id) {
|
||||
return false;
|
||||
}
|
||||
|
||||
auto lock_it = findLock(doc_id);
|
||||
bool self_held = lock_it != _held_locks.end();
|
||||
const bool foreign_held = !self_held && _notes.isWriteLocked(doc_id);
|
||||
|
||||
auto text = doc->getText();
|
||||
if (renderDocText(text)) {
|
||||
ImGui::InputTextMultiline(
|
||||
"##doc",
|
||||
&text,
|
||||
{-1,-1},
|
||||
ImGuiInputTextFlags_AllowTabInput |
|
||||
(foreign_held ? ImGuiInputTextFlags_ReadOnly : ImGuiInputTextFlags_None) |
|
||||
ImGuiInputTextFlags_CallbackAlways
|
||||
//cb,
|
||||
//&text
|
||||
);
|
||||
if (!foreign_held && !self_held && (ImGui::IsItemActive() || ImGui::IsItemEdited())) {
|
||||
// TODO: check
|
||||
_held_locks.emplace(_notes.writeLockAquire(doc_id).value());
|
||||
self_held = true;
|
||||
//std::cout << "!!!! imgui lock aquired\n";
|
||||
} else if (!foreign_held && self_held && !(ImGui::IsItemActive() || ImGui::IsItemEdited())) {
|
||||
// release lock
|
||||
_held_locks.erase(lock_it);
|
||||
//std::cout << "!!!! imgui lock released\n";
|
||||
}
|
||||
|
||||
if (self_held && ImGui::IsItemEdited()) {
|
||||
_notes_sync.merge(doc_id, text);
|
||||
return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
bool CRDTNotesImGui::renderDocText(std::string& text) const {
|
||||
// TODO: replace with text editor (zep) or visualize stuff??
|
||||
return ImGui::InputTextMultiline("##doc", &text, {-1,-1}, ImGuiInputTextFlags_AllowTabInput);
|
||||
}
|
||||
|
||||
|
@@ -4,6 +4,7 @@
|
||||
#include <solanaceae/contact/fwd.hpp>
|
||||
|
||||
#include <set>
|
||||
#include <unordered_set>
|
||||
|
||||
class CRDTNotesImGui {
|
||||
CRDTNotes& _notes;
|
||||
@@ -13,6 +14,9 @@ class CRDTNotesImGui {
|
||||
bool _show_global_list {true};
|
||||
|
||||
std::set<CRDTNotes::DocID> _open_docs;
|
||||
std::unordered_set<CRDTNotes::DocWriteLock> _held_locks;
|
||||
|
||||
std::unordered_set<CRDTNotes::DocWriteLock>::iterator findLock(const CRDTNotes::DocID& doc_id);
|
||||
|
||||
public:
|
||||
CRDTNotesImGui(CRDTNotes& notes, CRDTNotesSync& notes_sync, ContactStore4I& cs);
|
||||
@@ -22,6 +26,5 @@ class CRDTNotesImGui {
|
||||
bool renderContactListContactSmall(const Contact4 c, const bool selected) const;
|
||||
|
||||
bool renderDoc(const CRDTNotes::DocID& doc_id);
|
||||
bool renderDocText(std::string& text) const;
|
||||
};
|
||||
|
||||
|
@@ -16,6 +16,7 @@ enum class NGCEXT_Event : uint8_t {
|
||||
// - array [
|
||||
// - AgentID
|
||||
// - seq (frontier)
|
||||
// - del_num
|
||||
// - ]
|
||||
CRDTN_GOSSIP_FRONTIER,
|
||||
|
||||
@@ -26,7 +27,11 @@ enum class NGCEXT_Event : uint8_t {
|
||||
// - AgentID
|
||||
// - seq_from
|
||||
// - seq_to
|
||||
CRDTN_FETCH_OP_RANGE,
|
||||
CRDTN_FETCH_ADD_RANGE,
|
||||
|
||||
// - DocID
|
||||
// - AgentID
|
||||
CRDTN_FETCH_DEL,
|
||||
|
||||
// - DocID
|
||||
// - array [
|
||||
@@ -101,7 +106,7 @@ void CRDTNotesToxSync::SendGossip(
|
||||
void CRDTNotesToxSync::SendGossip(
|
||||
ContactHandle4 c,
|
||||
const CRDTNotes::DocID& doc_id,
|
||||
const std::vector<CRDTNotes::Frontier>& selected_frontier
|
||||
const std::vector<CRDTNotes::Frontier>& frontier
|
||||
) {
|
||||
if (!c.all_of<Contact::Components::ToxGroupPeerEphemeral>()) {
|
||||
return;
|
||||
@@ -117,7 +122,7 @@ void CRDTNotesToxSync::SendGossip(
|
||||
}
|
||||
// +32
|
||||
|
||||
for (const auto& [f_id, f_seq] : selected_frontier) {
|
||||
for (const auto& [f_id, f_seq, del_num] : frontier) {
|
||||
for (const uint8_t v : f_id) {
|
||||
pkg.push_back(v);
|
||||
}
|
||||
@@ -127,8 +132,13 @@ void CRDTNotesToxSync::SendGossip(
|
||||
pkg.push_back((f_seq >> i*8) & 0xff);
|
||||
}
|
||||
// +8
|
||||
|
||||
for (size_t i = 0; i < sizeof(del_num); i++) {
|
||||
pkg.push_back((del_num >> i*8) & 0xff);
|
||||
}
|
||||
// +40
|
||||
// +8
|
||||
}
|
||||
// +48
|
||||
|
||||
// send
|
||||
const auto& gp = c.get<Contact::Components::ToxGroupPeerEphemeral>();
|
||||
@@ -164,7 +174,7 @@ void CRDTNotesToxSync::SendFetchCompleteFrontier(
|
||||
);
|
||||
}
|
||||
|
||||
void CRDTNotesToxSync::SendFetchOps(
|
||||
void CRDTNotesToxSync::SendFetchAddRange(
|
||||
ContactHandle4 c,
|
||||
const CRDTNotes::DocID& doc_id,
|
||||
const CRDTNotes::CRDTAgent& agent,
|
||||
@@ -177,7 +187,7 @@ void CRDTNotesToxSync::SendFetchOps(
|
||||
|
||||
std::vector<uint8_t> pkg;
|
||||
|
||||
pkg.push_back(static_cast<uint8_t>(NGCEXT_Event::CRDTN_FETCH_OP_RANGE));
|
||||
pkg.push_back(static_cast<uint8_t>(NGCEXT_Event::CRDTN_FETCH_ADD_RANGE));
|
||||
|
||||
for (const uint8_t v : doc_id) {
|
||||
pkg.push_back(v);
|
||||
@@ -207,6 +217,35 @@ void CRDTNotesToxSync::SendFetchOps(
|
||||
);
|
||||
}
|
||||
|
||||
void CRDTNotesToxSync::SendFetchDel(
|
||||
ContactHandle4 c,
|
||||
const CRDTNotes::DocID& doc_id,
|
||||
const CRDTNotes::CRDTAgent& agent
|
||||
) {
|
||||
if (!c.all_of<Contact::Components::ToxGroupPeerEphemeral>()) {
|
||||
return;
|
||||
}
|
||||
|
||||
std::vector<uint8_t> pkg;
|
||||
|
||||
pkg.push_back(static_cast<uint8_t>(NGCEXT_Event::CRDTN_FETCH_DEL));
|
||||
|
||||
for (const uint8_t v : doc_id) {
|
||||
pkg.push_back(v);
|
||||
}
|
||||
|
||||
for (const uint8_t v : agent) {
|
||||
pkg.push_back(v);
|
||||
}
|
||||
|
||||
const auto& gp = c.get<Contact::Components::ToxGroupPeerEphemeral>();
|
||||
_t.toxGroupSendCustomPrivatePacket(
|
||||
gp.group_number, gp.peer_number,
|
||||
true,
|
||||
pkg
|
||||
);
|
||||
}
|
||||
|
||||
void CRDTNotesToxSync::SendOps(
|
||||
ContactHandle4 c,
|
||||
const CRDTNotes::DocID& doc_id,
|
||||
@@ -228,6 +267,7 @@ void CRDTNotesToxSync::SendOps(
|
||||
|
||||
// this is very inefficent
|
||||
// a full add op is 124bytes like this
|
||||
// a full del op is 41bytes
|
||||
for (const auto& op : ops) {
|
||||
if(std::holds_alternative<CRDTNotes::Doc::OpAdd>(op)) {
|
||||
const auto& add_op = std::get<CRDTNotes::Doc::OpAdd>(op);
|
||||
@@ -332,7 +372,7 @@ bool CRDTNotesToxSync::parse_crdtn_gossip_frontier(
|
||||
|
||||
while (curser < data_size) {
|
||||
CRDTNotes::Frontier new_f;
|
||||
_DATA_HAVE(new_f.agent.size() * sizeof(CRDTNotes::CRDTAgent::value_type) + sizeof(new_f.seq), std::cerr << "NGCEXT: packet malformed, not enough data for forntier\n"; return false;)
|
||||
_DATA_HAVE(new_f.agent.size() * sizeof(CRDTNotes::CRDTAgent::value_type) + sizeof(new_f.seq) + sizeof(new_f.del_num), std::cerr << "NGCEXT: packet malformed, not enough data for frontier\n"; return false;)
|
||||
|
||||
for (size_t i = 0; i < new_f.agent.size(); i++, curser++) {
|
||||
new_f.agent[i] = data[curser];
|
||||
@@ -343,7 +383,12 @@ bool CRDTNotesToxSync::parse_crdtn_gossip_frontier(
|
||||
new_f.seq |= uint64_t(data[curser]) << i*8;
|
||||
}
|
||||
|
||||
e.selected_frontier.emplace_back(std::move(new_f));
|
||||
new_f.del_num = 0;
|
||||
for (size_t i = 0; i < sizeof(new_f.del_num); i++, curser++) {
|
||||
new_f.del_num |= uint64_t(data[curser]) << i*8;
|
||||
}
|
||||
|
||||
e.frontier.emplace_back(std::move(new_f));
|
||||
}
|
||||
|
||||
std::cout << "CRDTN gossip_frontier parsed\n";
|
||||
@@ -371,12 +416,12 @@ bool CRDTNotesToxSync::parse_crdtn_fetch_complete_frontier(
|
||||
return true;
|
||||
}
|
||||
|
||||
bool CRDTNotesToxSync::parse_crdtn_fetch_op_range(
|
||||
bool CRDTNotesToxSync::parse_crdtn_fetch_add_range(
|
||||
ContactHandle4 c,
|
||||
const uint8_t* data, size_t data_size,
|
||||
bool // dont care private
|
||||
) {
|
||||
Events::NGCEXT_crdtns_fetch_op_range e;
|
||||
Events::NGCEXT_crdtns_fetch_add_range e;
|
||||
e.c = c;
|
||||
|
||||
size_t curser = 0;
|
||||
@@ -403,7 +448,32 @@ bool CRDTNotesToxSync::parse_crdtn_fetch_op_range(
|
||||
e.seq_to |= uint64_t(data[curser]) << i*8;
|
||||
}
|
||||
|
||||
std::cout << "CRDTN fetch_op_range parsed\n";
|
||||
std::cout << "CRDTN fetch_add_range parsed\n";
|
||||
_notes_sync.onCRDTNSyncEvent(std::move(e));
|
||||
return true;
|
||||
}
|
||||
|
||||
bool CRDTNotesToxSync::parse_crdtn_fetch_del(
|
||||
ContactHandle4 c,
|
||||
const uint8_t* data, size_t data_size,
|
||||
bool // dont care private
|
||||
) {
|
||||
Events::NGCEXT_crdtns_fetch_del e;
|
||||
e.c = c;
|
||||
|
||||
size_t curser = 0;
|
||||
|
||||
_DATA_HAVE(e.doc_id.size() * sizeof(decltype(e.doc_id)::value_type), std::cerr << "NGCEXT: packet too small, missing doc_id\n"; return false;)
|
||||
for (size_t i = 0; i < e.doc_id.size(); i++, curser++) {
|
||||
e.doc_id[i] = data[curser];
|
||||
}
|
||||
|
||||
_DATA_HAVE(e.agent.size() * sizeof(decltype(e.agent)::value_type), std::cerr << "NGCEXT: packet too small, missing agent\n"; return false;)
|
||||
for (size_t i = 0; i < e.agent.size(); i++, curser++) {
|
||||
e.agent[i] = data[curser];
|
||||
}
|
||||
|
||||
std::cout << "CRDTN fetch_del parsed\n";
|
||||
_notes_sync.onCRDTNSyncEvent(std::move(e));
|
||||
return true;
|
||||
}
|
||||
@@ -531,8 +601,10 @@ bool CRDTNotesToxSync::handlePacket(
|
||||
return parse_crdtn_gossip_frontier(c, data+1, data_size-1, _private);
|
||||
case NGCEXT_Event::CRDTN_FETCH_COMPLETE_FRONTIER:
|
||||
return parse_crdtn_fetch_complete_frontier(c, data+1, data_size-1, _private);
|
||||
case NGCEXT_Event::CRDTN_FETCH_OP_RANGE:
|
||||
return parse_crdtn_fetch_op_range(c, data+1, data_size-1, _private);
|
||||
case NGCEXT_Event::CRDTN_FETCH_ADD_RANGE:
|
||||
return parse_crdtn_fetch_add_range(c, data+1, data_size-1, _private);
|
||||
case NGCEXT_Event::CRDTN_FETCH_DEL:
|
||||
return parse_crdtn_fetch_del(c, data+1, data_size-1, _private);
|
||||
case NGCEXT_Event::CRDTN_OPS:
|
||||
return parse_crdtn_ops(c, data+1, data_size-1, _private);
|
||||
default:
|
||||
|
@@ -39,7 +39,7 @@ class CRDTNotesToxSync : public CRDTNotesContactSyncModelI, public ToxEventI {
|
||||
void SendGossip(
|
||||
ContactHandle4 c,
|
||||
const CRDTNotes::DocID& doc_id,
|
||||
const std::vector<CRDTNotes::Frontier>& selected_frontier
|
||||
const std::vector<CRDTNotes::Frontier>& frontier
|
||||
) override;
|
||||
|
||||
void SendFetchCompleteFrontier(
|
||||
@@ -47,7 +47,7 @@ class CRDTNotesToxSync : public CRDTNotesContactSyncModelI, public ToxEventI {
|
||||
const CRDTNotes::DocID& doc_id
|
||||
) override;
|
||||
|
||||
void SendFetchOps(
|
||||
void SendFetchAddRange(
|
||||
ContactHandle4 c,
|
||||
const CRDTNotes::DocID& doc_id,
|
||||
const CRDTNotes::CRDTAgent& agent,
|
||||
@@ -55,6 +55,12 @@ class CRDTNotesToxSync : public CRDTNotesContactSyncModelI, public ToxEventI {
|
||||
const uint64_t seq_to
|
||||
) override;
|
||||
|
||||
void SendFetchDel(
|
||||
ContactHandle4 c,
|
||||
const CRDTNotes::DocID& doc_id,
|
||||
const CRDTNotes::CRDTAgent& agent
|
||||
) override;
|
||||
|
||||
void SendOps(
|
||||
ContactHandle4 c,
|
||||
const CRDTNotes::DocID& doc_id,
|
||||
@@ -77,7 +83,12 @@ class CRDTNotesToxSync : public CRDTNotesContactSyncModelI, public ToxEventI {
|
||||
const uint8_t* data, size_t data_size,
|
||||
bool _private
|
||||
);
|
||||
bool parse_crdtn_fetch_op_range(
|
||||
bool parse_crdtn_fetch_add_range(
|
||||
ContactHandle4 c,
|
||||
const uint8_t* data, size_t data_size,
|
||||
bool _private
|
||||
);
|
||||
bool parse_crdtn_fetch_del(
|
||||
ContactHandle4 c,
|
||||
const uint8_t* data, size_t data_size,
|
||||
bool _private
|
||||
|
Reference in New Issue
Block a user