From 469e57773e08f686cde992cc357948ffeda436fe Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Simon=20Gra=CC=88tzer?= <simon@graetzer.org>
Date: Wed, 31 May 2017 16:37:58 +0200
Subject: [PATCH] workaround for replication

---
 arangod/RocksDBEngine/RocksDBCollection.cpp   | 14 ++++++++--
 .../RocksDBReplicationContext.cpp             | 27 +++++++++++++------
 2 files changed, 31 insertions(+), 10 deletions(-)

diff --git a/arangod/RocksDBEngine/RocksDBCollection.cpp b/arangod/RocksDBEngine/RocksDBCollection.cpp
index 52812a0a71..ff723c630f 100644
--- a/arangod/RocksDBEngine/RocksDBCollection.cpp
+++ b/arangod/RocksDBEngine/RocksDBCollection.cpp
@@ -1333,6 +1333,11 @@ RocksDBOperationResult RocksDBCollection::insertDocument(
     res.keySize(key.string().size());
     return res;
   }
+  
+  /*LOG_TOPIC(ERR, Logger::FIXME) << "PUT rev: " << revisionId << " trx: " << trx->state()->id()
+  << " seq: " <<  mthd->readOptions().snapshot->GetSequenceNumber()
+  << " objectID " << _objectId
+  << " name: " << _logicalCollection->name();*/
 
   RocksDBOperationResult innerRes;
   READ_LOCKER(guard, _indexesLock);
@@ -1391,6 +1396,11 @@ RocksDBOperationResult RocksDBCollection::removeDocument(
   }
   //}
 
+  /*LOG_TOPIC(ERR, Logger::FIXME) << "Delete rev: " << revisionId << " trx: " << trx->state()->id()
+  << " seq: " <<  mthd->readOptions().snapshot->GetSequenceNumber()
+  << " objectID " << _objectId
+  << " name: " << _logicalCollection->name();*/
+  
   RocksDBOperationResult resInner;
   READ_LOCKER(guard, _indexesLock);
   for (std::shared_ptr<Index> const& idx : _indexes) {
@@ -1517,10 +1527,10 @@ arangodb::Result RocksDBCollection::lookupRevisionVPack(
 
     mdr.setManaged(std::move(value), revisionId);
   } else {
-    /*LOG_TOPIC(ERR, Logger::FIXME) << "NOT FOUND rev: " << revisionId << " trx: " << trx->state()->id()
+    LOG_TOPIC(ERR, Logger::FIXME) << "NOT FOUND rev: " << revisionId << " trx: " << trx->state()->id()
                                   << " seq: " <<  mthd->readOptions().snapshot->GetSequenceNumber()
                                   << " objectID " << _objectId
-    << " name: " << _logicalCollection->name();*/
+    << " name: " << _logicalCollection->name();
     mdr.reset();
   }
   return res;
diff --git a/arangod/RocksDBEngine/RocksDBReplicationContext.cpp b/arangod/RocksDBEngine/RocksDBReplicationContext.cpp
index 4850bc3ff6..c04b01b0a7 100644
--- a/arangod/RocksDBEngine/RocksDBReplicationContext.cpp
+++ b/arangod/RocksDBEngine/RocksDBReplicationContext.cpp
@@ -213,9 +213,14 @@ arangodb::Result RocksDBReplicationContext::dumpKeyChunks(VPackBuilder& b,
   VPackSlice highKey;  // FIXME: no good keeping this
 
   uint64_t hash = 0x012345678;
-  auto cb = [&](ManagedDocumentResult const& mdr) {
-
-    VPackSlice doc(mdr.vpack());
+  auto cb = [&](DocumentIdentifierToken const& token) {
+    bool ok = _collection->readDocument(_trx.get(), token, _mdr);
+    if (!ok) {
+      // TODO: do something here?
+      return;
+    }
+    
+    VPackSlice doc(_mdr.vpack());
     highKey = doc.get(StaticStrings::KeyString);
     // set type
     if (lowKey.empty()) {
@@ -231,7 +236,7 @@ arangodb::Result RocksDBReplicationContext::dumpKeyChunks(VPackBuilder& b,
   b.openArray();
   while (_hasMore) {
     try {
-      _hasMore = primary->nextDocument(cb, chunkSize);
+      _hasMore = primary->next(cb, chunkSize);
 
       b.add(VPackValue(VPackValueType::Object));
       b.add("low", VPackValue(lowKey));
@@ -282,20 +287,26 @@ arangodb::Result RocksDBReplicationContext::dumpKeys(
     }
   }
 
-  auto cb = [&](ManagedDocumentResult const& mdr) {
-    VPackSlice doc(mdr.vpack());
+  auto cb = [&](DocumentIdentifierToken const& token) {
+    bool ok = _collection->readDocument(_trx.get(), token, _mdr);
+    if (!ok) {
+      // TODO: do something here?
+      return;
+    }
+    
+    VPackSlice doc(_mdr.vpack());
     VPackSlice key = doc.get(StaticStrings::KeyString);
 
     b.openArray();
     b.add(key);
-    b.add(VPackValue(std::to_string(mdr.lastRevisionId())));
+    b.add(VPackValue(std::to_string(_mdr.lastRevisionId())));
     b.close();
   };
 
   b.openArray();
   // chunkSize is going to be ignored here
   try {
-    _hasMore = primary->nextDocument(cb, chunkSize);
+    _hasMore = primary->next(cb, chunkSize);
     _lastIteratorOffset++;
   } catch (std::exception const&) {
     return Result(TRI_ERROR_INTERNAL);