Add bucket resolver interface and use it for put,update,remove and get messages in DocumentApiConverter.

6 files changed, 131 insertions, 38 deletions

diff --git a/storage/src/tests/storageserver/documentapiconvertertest.cpp b/storage/src/tests/storageserver/documentapiconvertertest.cpp
index 1744bfb6a79..7467bae0c41 100644
--- a/storage/src/tests/storageserver/documentapiconvertertest.cpp
+++ b/storage/src/tests/storageserver/documentapiconvertertest.cpp
@@ -1,20 +1,24 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 
-#include <vespa/document/base/testdocrepo.h>
 #include <cppunit/extensions/HelperMacros.h>
+#include <vespa/config/subscription/configuri.h>
+#include <vespa/document/base/testdocrepo.h>
+#include <vespa/document/bucket/bucketidfactory.h>
+#include <vespa/document/datatype/documenttype.h>
+#include <vespa/document/test/make_document_bucket.h>
+#include <vespa/documentapi/documentapi.h>
+#include <vespa/messagebus/emptyreply.h>
+#include <vespa/storage/common/bucket_resolver.h>
 #include <vespa/storage/storageserver/documentapiconverter.h>
 #include <vespa/storageapi/message/batch.h>
 #include <vespa/storageapi/message/datagram.h>
 #include <vespa/storageapi/message/multioperation.h>
 #include <vespa/storageapi/message/persistence.h>
-#include <vespa/documentapi/documentapi.h>
-#include <vespa/messagebus/emptyreply.h>
-#include <vespa/document/datatype/documenttype.h>
-#include <vespa/document/bucket/bucketidfactory.h>
-#include <vespa/config/subscription/configuri.h>
 #include <vespa/vespalib/testkit/test_kit.h>
-#include <vespa/document/test/make_document_bucket.h>
 
+using document::Bucket;
+using document::BucketId;
+using document::BucketSpace;
 using document::DataType;
 using document::DocIdString;
 using document::Document;
@@ -25,25 +29,40 @@ using document::test::makeDocumentBucket;
 
 namespace storage {
 
+DocumentId defaultDocId("id:test:text/html::0");
+const Bucket defaultBucket(BucketSpace(5), BucketId(0));
+
+struct MockBucketResolver : public BucketResolver {
+    virtual Bucket bucketFromId(const DocumentId &documentId) const override {
+        if (documentId.getDocType() == "text/html") {
+            return defaultBucket;
+        }
+        return Bucket(BucketSpace(0), BucketId(0));
+    }
+};
+
 struct DocumentApiConverterTest : public CppUnit::TestFixture
 {
+    MockBucketResolver _bucketResolver;
     std::unique_ptr<DocumentApiConverter> _converter;
     const DocumentTypeRepo::SP _repo;
     const DataType& _html_type;
 
     DocumentApiConverterTest()
-        : _repo(new DocumentTypeRepo(readDocumenttypesConfig(
+        : _bucketResolver(),
+          _repo(new DocumentTypeRepo(readDocumenttypesConfig(
                     TEST_PATH("config-doctypes.cfg")))),
           _html_type(*_repo->getDocumentType("text/html"))
     {
     }
 
     void setUp() override {
-        _converter.reset(new DocumentApiConverter("raw:"));
+        _converter.reset(new DocumentApiConverter("raw:", _bucketResolver));
     };
 
     void testPut();
     void testForwardedPut();
+    void testUpdate();
     void testRemove();
     void testGet();
     void testCreateVisitor();
@@ -58,6 +77,7 @@ struct DocumentApiConverterTest : public CppUnit::TestFixture
     CPPUNIT_TEST_SUITE(DocumentApiConverterTest);
     CPPUNIT_TEST(testPut);
     CPPUNIT_TEST(testForwardedPut);
+    CPPUNIT_TEST(testUpdate);
     CPPUNIT_TEST(testRemove);
     CPPUNIT_TEST(testGet);
     CPPUNIT_TEST(testCreateVisitor);
@@ -75,12 +95,13 @@ CPPUNIT_TEST_SUITE_REGISTRATION(DocumentApiConverterTest);
 
 void DocumentApiConverterTest::testPut()
 {
-    Document::SP doc(new Document(_html_type, DocumentId(DocIdString("test", "test"))));
+    Document::SP doc(new Document(_html_type, defaultDocId));
 
     documentapi::PutDocumentMessage putmsg(doc);
     putmsg.setTimestamp(1234);
 
     std::unique_ptr<storage::api::StorageCommand> cmd = _converter->toStorageAPI(putmsg, _repo);
+    CPPUNIT_ASSERT_EQUAL(defaultBucket, cmd->getBucket());
     api::PutCommand* pc = dynamic_cast<api::PutCommand*>(cmd.get());
 
     CPPUNIT_ASSERT(pc);
@@ -120,15 +141,46 @@ void DocumentApiConverterTest::testForwardedPut()
     _converter->transferReplyState(*pr, *reply);
 }
 
+void DocumentApiConverterTest::testUpdate()
+{
+    auto update = std::make_shared<document::DocumentUpdate>(_html_type, defaultDocId);
+    documentapi::UpdateDocumentMessage updateMsg(update);
+    updateMsg.setOldTimestamp(1234);
+    updateMsg.setNewTimestamp(5678);
+
+    auto storageCmd = _converter->toStorageAPI(updateMsg, _repo);
+    CPPUNIT_ASSERT_EQUAL(defaultBucket, storageCmd->getBucket());
+
+    auto updateCmd = dynamic_cast<api::UpdateCommand*>(storageCmd.get());
+    CPPUNIT_ASSERT(updateCmd);
+    CPPUNIT_ASSERT_EQUAL(update.get(), updateCmd->getUpdate().get());
+    CPPUNIT_ASSERT_EQUAL(api::Timestamp(1234), updateCmd->getOldTimestamp());
+    CPPUNIT_ASSERT_EQUAL(api::Timestamp(5678), updateCmd->getTimestamp());
+
+    auto mbusReply = updateMsg.createReply();
+    CPPUNIT_ASSERT(mbusReply.get());
+    auto storageReply = _converter->toStorageAPI(static_cast<documentapi::DocumentReply&>(*mbusReply), *storageCmd);
+    auto updateReply = dynamic_cast<api::UpdateReply*>(storageReply.get());
+    CPPUNIT_ASSERT(updateReply);
+
+    auto mbusMsg = _converter->toDocumentAPI(*updateCmd, _repo);
+    auto mbusUpdate = dynamic_cast<documentapi::UpdateDocumentMessage*>(mbusMsg.get());
+    CPPUNIT_ASSERT(mbusUpdate);
+    CPPUNIT_ASSERT((&mbusUpdate->getDocumentUpdate()) == update.get());
+    CPPUNIT_ASSERT_EQUAL(api::Timestamp(1234), mbusUpdate->getOldTimestamp());
+    CPPUNIT_ASSERT_EQUAL(api::Timestamp(5678), mbusUpdate->getNewTimestamp());
+}
+
 void DocumentApiConverterTest::testRemove()
 {
-    documentapi::RemoveDocumentMessage removemsg(document::DocumentId(document::DocIdString("test", "test")));
+    documentapi::RemoveDocumentMessage removemsg(defaultDocId);
     std::unique_ptr<storage::api::StorageCommand> cmd = _converter->toStorageAPI(removemsg, _repo);
+    CPPUNIT_ASSERT_EQUAL(defaultBucket, cmd->getBucket());
 
     api::RemoveCommand* rc = dynamic_cast<api::RemoveCommand*>(cmd.get());
 
     CPPUNIT_ASSERT(rc);
-    CPPUNIT_ASSERT_EQUAL(document::DocumentId(document::DocIdString("test", "test")), rc->getDocumentId());
+    CPPUNIT_ASSERT_EQUAL(defaultDocId, rc->getDocumentId());
 
     std::unique_ptr<mbus::Reply> reply = removemsg.createReply();
     CPPUNIT_ASSERT(reply.get());
@@ -142,20 +194,20 @@ void DocumentApiConverterTest::testRemove()
 
     documentapi::RemoveDocumentMessage* mbusremove = dynamic_cast<documentapi::RemoveDocumentMessage*>(mbusmsg.get());
     CPPUNIT_ASSERT(mbusremove);
-    CPPUNIT_ASSERT_EQUAL(document::DocumentId(document::DocIdString("test", "test")), mbusremove->getDocumentId());
+    CPPUNIT_ASSERT_EQUAL(defaultDocId, mbusremove->getDocumentId());
 };
 
 void DocumentApiConverterTest::testGet()
 {
-    documentapi::GetDocumentMessage getmsg(
-            document::DocumentId(document::DocIdString("test", "test")), "foo bar");
+    documentapi::GetDocumentMessage getmsg(defaultDocId, "foo bar");
 
     std::unique_ptr<storage::api::StorageCommand> cmd = _converter->toStorageAPI(getmsg, _repo);
+    CPPUNIT_ASSERT_EQUAL(defaultBucket, cmd->getBucket());
 
     api::GetCommand* rc = dynamic_cast<api::GetCommand*>(cmd.get());
 
     CPPUNIT_ASSERT(rc);
-    CPPUNIT_ASSERT_EQUAL(document::DocumentId(document::DocIdString("test", "test")), rc->getDocumentId());
+    CPPUNIT_ASSERT_EQUAL(defaultDocId, rc->getDocumentId());
     CPPUNIT_ASSERT_EQUAL(vespalib::string("foo bar"), rc->getFieldSet());
 };
 
diff --git a/storage/src/vespa/storage/common/bucket_resolver.h b/storage/src/vespa/storage/common/bucket_resolver.h
new file mode 100644
index 00000000000..7224a392e2f
--- /dev/null
+++ b/storage/src/vespa/storage/common/bucket_resolver.h
@@ -0,0 +1,18 @@
+// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+#pragma once
+
+#include <vespa/document/bucket/bucket.h>
+
+namespace document { class DocumentId; }
+
+namespace storage {
+
+/**
+ * Interface for resolving which bucket a given a document id belongs to.
+ */
+struct BucketResolver {
+    virtual ~BucketResolver() {}
+    virtual document::Bucket bucketFromId(const document::DocumentId &documentId) const = 0;
+};
+
+}
diff --git a/storage/src/vespa/storage/storageserver/communicationmanager.cpp b/storage/src/vespa/storage/storageserver/communicationmanager.cpp
index c90b18038c2..71a918610e6 100644
--- a/storage/src/vespa/storage/storageserver/communicationmanager.cpp
+++ b/storage/src/vespa/storage/storageserver/communicationmanager.cpp
@@ -1,18 +1,20 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
 #include "communicationmanager.h"
 #include "fnetlistener.h"
 #include "rpcrequestwrapper.h"
-#include <vespa/storage/config/config-stor-server.h>
-#include <vespa/storage/common/nodestateupdater.h>
-#include <vespa/storageframework/generic/clock/timer.h>
 #include <vespa/documentapi/messagebus/messages/wrongdistributionreply.h>
-#include <vespa/storageapi/message/state.h>
-#include <vespa/messagebus/rpcmessagebus.h>
-#include <vespa/messagebus/network/rpcnetworkparams.h>
 #include <vespa/messagebus/emptyreply.h>
+#include <vespa/messagebus/network/rpcnetworkparams.h>
+#include <vespa/messagebus/rpcmessagebus.h>
+#include <vespa/storage/common/bucket_resolver.h>
+#include <vespa/storage/common/nodestateupdater.h>
+#include <vespa/storage/config/config-stor-server.h>
+#include <vespa/storageapi/message/state.h>
+#include <vespa/storageframework/generic/clock/timer.h>
 #include <vespa/vespalib/stllike/asciistream.h>
-#include <vespa/vespalib/util/stringfmt.h>
 #include <vespa/vespalib/stllike/hash_map.hpp>
+#include <vespa/vespalib/util/stringfmt.h>
 
 #include <vespa/log/bufferedlogger.h>
 LOG_SETUP(".communication.manager");
@@ -267,6 +269,16 @@ CommunicationManager::handleReply(std::unique_ptr<mbus::Reply> reply)
     }
 }
 
+namespace {
+
+struct PlaceHolderBucketResolver : public BucketResolver {
+    virtual document::Bucket bucketFromId(const document::DocumentId &) const override {
+        return document::Bucket(document::BucketSpace::placeHolder(), document::BucketId(0));
+    }
+};
+
+}
+
 CommunicationManager::CommunicationManager(StorageComponentRegister& compReg, const config::ConfigUri & configUri)
     : StorageLink("Communication manager"),
       _component(compReg, "communicationmanager"),
@@ -277,7 +289,8 @@ CommunicationManager::CommunicationManager(StorageComponentRegister& compReg, co
       _count(0),
       _configUri(configUri),
       _closed(false),
-      _docApiConverter(configUri),
+      _bucketResolver(std::make_unique<PlaceHolderBucketResolver>()),
+      _docApiConverter(configUri, *_bucketResolver),
       _messageAllocTypes(_component.getMemoryManager())
 {
     _component.registerMetricUpdateHook(*this, framework::SecondTime(5));
diff --git a/storage/src/vespa/storage/storageserver/communicationmanager.h b/storage/src/vespa/storage/storageserver/communicationmanager.h
index e1f7888ac67..4cf3f33e6ea 100644
--- a/storage/src/vespa/storage/storageserver/communicationmanager.h
+++ b/storage/src/vespa/storage/storageserver/communicationmanager.h
@@ -35,6 +35,7 @@ namespace mbus {
 }
 namespace storage {
 
+class BucketResolver;
 class VisitorMbusSession;
 class Visitor;
 class VisitorThread;
@@ -167,6 +168,7 @@ private:
 
     config::ConfigUri _configUri;
     std::atomic<bool> _closed;
+    std::unique_ptr<BucketResolver> _bucketResolver;
     DocumentApiConverter _docApiConverter;
     framework::Thread::UP _thread;
     MessageAllocationTypes _messageAllocTypes;
diff --git a/storage/src/vespa/storage/storageserver/documentapiconverter.cpp b/storage/src/vespa/storage/storageserver/documentapiconverter.cpp
index ddc11e9ad77..331e32f43b8 100644
--- a/storage/src/vespa/storage/storageserver/documentapiconverter.cpp
+++ b/storage/src/vespa/storage/storageserver/documentapiconverter.cpp
@@ -1,18 +1,20 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
 #include "documentapiconverter.h"
 #include "priorityconverter.h"
+#include <vespa/document/bucket/bucketidfactory.h>
 #include <vespa/documentapi/documentapi.h>
-#include <vespa/storageapi/message/visitor.h>
+#include <vespa/storage/common/bucket_resolver.h>
+#include <vespa/storageapi/message/batch.h>
 #include <vespa/storageapi/message/datagram.h>
-#include <vespa/storageapi/message/persistence.h>
-#include <vespa/storageapi/message/searchresult.h>
-#include <vespa/storageapi/message/queryresult.h>
 #include <vespa/storageapi/message/documentsummary.h>
 #include <vespa/storageapi/message/multioperation.h>
+#include <vespa/storageapi/message/persistence.h>
+#include <vespa/storageapi/message/queryresult.h>
 #include <vespa/storageapi/message/removelocation.h>
+#include <vespa/storageapi/message/searchresult.h>
 #include <vespa/storageapi/message/stat.h>
-#include <vespa/storageapi/message/batch.h>
-#include <vespa/document/bucket/bucketidfactory.h>
+#include <vespa/storageapi/message/visitor.h>
 
 #include <vespa/log/log.h>
 LOG_SETUP(".documentapiconverter");
@@ -21,8 +23,10 @@ using document::BucketSpace;
 
 namespace storage {
 
-DocumentApiConverter::DocumentApiConverter(const config::ConfigUri & configUri)
-    : _priConverter(std::make_unique<PriorityConverter>(configUri))
+DocumentApiConverter::DocumentApiConverter(const config::ConfigUri &configUri,
+                                           const BucketResolver &bucketResolver)
+    : _priConverter(std::make_unique<PriorityConverter>(configUri)),
+      _bucketResolver(bucketResolver)
 {}
 
 DocumentApiConverter::~DocumentApiConverter() {}
@@ -38,7 +42,8 @@ DocumentApiConverter::toStorageAPI(documentapi::DocumentMessage& fromMsg,
     case DocumentProtocol::MESSAGE_PUTDOCUMENT:
     {
         documentapi::PutDocumentMessage& from(static_cast<documentapi::PutDocumentMessage&>(fromMsg));
-        auto to = std::make_unique<api::PutCommand>(document::Bucket(BucketSpace::placeHolder(), document::BucketId(0)), from.stealDocument(), from.getTimestamp());
+        document::Bucket bucket = _bucketResolver.bucketFromId(from.getDocument().getId());
+        auto to = std::make_unique<api::PutCommand>(bucket, from.stealDocument(), from.getTimestamp());
         to->setCondition(from.getCondition());
         toMsg = std::move(to);
         break;
@@ -46,8 +51,8 @@ DocumentApiConverter::toStorageAPI(documentapi::DocumentMessage& fromMsg,
     case DocumentProtocol::MESSAGE_UPDATEDOCUMENT:
     {
         documentapi::UpdateDocumentMessage& from(static_cast<documentapi::UpdateDocumentMessage&>(fromMsg));
-        auto to = std::make_unique<api::UpdateCommand>(document::Bucket(BucketSpace::placeHolder(), document::BucketId(0)), from.stealDocumentUpdate(),
-                                                       from.getNewTimestamp());
+        document::Bucket bucket = _bucketResolver.bucketFromId(from.getDocumentUpdate().getId());
+        auto to = std::make_unique<api::UpdateCommand>(bucket, from.stealDocumentUpdate(), from.getNewTimestamp());
         to->setOldTimestamp(from.getOldTimestamp());
         to->setCondition(from.getCondition());
         toMsg = std::move(to);
@@ -56,7 +61,7 @@ DocumentApiConverter::toStorageAPI(documentapi::DocumentMessage& fromMsg,
     case DocumentProtocol::MESSAGE_REMOVEDOCUMENT:
     {
         documentapi::RemoveDocumentMessage& from(static_cast<documentapi::RemoveDocumentMessage&>(fromMsg));
-        auto to = std::make_unique<api::RemoveCommand>(document::Bucket(BucketSpace::placeHolder(), document::BucketId(0)), from.getDocumentId(), 0);
+        auto to = std::make_unique<api::RemoveCommand>(_bucketResolver.bucketFromId(from.getDocumentId()), from.getDocumentId(), 0);
         to->setCondition(from.getCondition());
         toMsg = std::move(to);
         break;
@@ -64,7 +69,7 @@ DocumentApiConverter::toStorageAPI(documentapi::DocumentMessage& fromMsg,
     case DocumentProtocol::MESSAGE_GETDOCUMENT:
     {
         documentapi::GetDocumentMessage& from(static_cast<documentapi::GetDocumentMessage&>(fromMsg));
-        auto to = std::make_unique<api::GetCommand>(document::Bucket(BucketSpace::placeHolder(), document::BucketId(0)), from.getDocumentId(), from.getFieldSet());
+        auto to = std::make_unique<api::GetCommand>(_bucketResolver.bucketFromId(from.getDocumentId()), from.getDocumentId(), from.getFieldSet());
         toMsg.reset(to.release());
         break;
     }
diff --git a/storage/src/vespa/storage/storageserver/documentapiconverter.h b/storage/src/vespa/storage/storageserver/documentapiconverter.h
index 0cc2f3f3b9c..5310bcd0127 100644
--- a/storage/src/vespa/storage/storageserver/documentapiconverter.h
+++ b/storage/src/vespa/storage/storageserver/documentapiconverter.h
@@ -13,6 +13,7 @@ namespace api {
     class StorageReply;
 }
 
+class BucketResolver;
 class PriorityConverter;
 /**
    Converts messages from storageapi to documentapi and
@@ -21,7 +22,8 @@ class PriorityConverter;
 class DocumentApiConverter
 {
 public:
-    DocumentApiConverter(const config::ConfigUri & configUri);
+    DocumentApiConverter(const config::ConfigUri &configUri,
+                         const BucketResolver &bucketResolver);
     ~DocumentApiConverter();
 
     std::unique_ptr<api::StorageCommand> toStorageAPI(documentapi::DocumentMessage& msg, const document::DocumentTypeRepo::SP &repo);
@@ -31,6 +33,7 @@ public:
     const PriorityConverter& getPriorityConverter() const { return *_priConverter; }
 private:
     std::unique_ptr<PriorityConverter> _priConverter;
+    const BucketResolver &_bucketResolver;
 };
 
 }  // namespace storage