Get rid of mbus::reply leftovers.

3 files changed, 44 insertions, 142 deletions

diff --git a/searchcore/src/tests/proton/documentdb/feedhandler/feedhandler_test.cpp b/searchcore/src/tests/proton/documentdb/feedhandler/feedhandler_test.cpp
index 29f043e7caa..30c0556e812 100644
--- a/searchcore/src/tests/proton/documentdb/feedhandler/feedhandler_test.cpp
+++ b/searchcore/src/tests/proton/documentdb/feedhandler/feedhandler_test.cpp
@@ -305,7 +305,7 @@ struct MyTransport : public FeedToken::ITransport {
     bool documentWasFound;
     MyTransport();
     ~MyTransport();
-    virtual void send(Reply::UP, ResultUP res, bool documentWasFound_, double) override {
+    void send(ResultUP res, bool documentWasFound_) override {
         result = std::move(res);
         documentWasFound = documentWasFound_;
         gate.countDown();
@@ -315,21 +315,12 @@ struct MyTransport : public FeedToken::ITransport {
 MyTransport::MyTransport() : gate(), result(), documentWasFound(false) {}
 MyTransport::~MyTransport() {}
 
-Reply::UP getReply(uint32_t type) {
-    if (type == DocumentProtocol::REPLY_REMOVEDOCUMENT) {
-        return Reply::UP(new RemoveDocumentReply);
-    } else if (type == DocumentProtocol::REPLY_UPDATEDOCUMENT) {
-        return Reply::UP(new UpdateDocumentReply);
-    }
-    return Reply::UP(new DocumentReply(type));
-}
-
 struct FeedTokenContext {
     MyTransport transport;
     FeedToken::UP token_ap;
     FeedToken &token;
 
-    FeedTokenContext(uint32_t type = 0);
+    FeedTokenContext();
     ~FeedTokenContext();
     bool await(uint32_t timeout = 80000) { return transport.gate.await(timeout); }
     const Result *getResult() {
@@ -340,24 +331,23 @@ struct FeedTokenContext {
     }
 };
 
-FeedTokenContext::FeedTokenContext(uint32_t type)
+FeedTokenContext::FeedTokenContext()
     : transport(),
-      token_ap(new FeedToken(transport, getReply(type))),
+      token_ap(new FeedToken(transport)),
       token(*token_ap)
 {
-    token.getReply().getTrace().setLevel(9);
 }
-FeedTokenContext::~FeedTokenContext() {}
+
+FeedTokenContext::~FeedTokenContext() = default;
 
 struct PutContext {
     FeedTokenContext tokenCtx;
     DocumentContext  docCtx;
     typedef std::shared_ptr<PutContext> SP;
     PutContext(const vespalib::string &docId, DocBuilder &builder) :
-        tokenCtx(DocumentProtocol::REPLY_PUTDOCUMENT),
+        tokenCtx(),
         docCtx(docId, builder)
-    {
-    }
+    {}
 };
 
 
@@ -371,12 +361,10 @@ struct PutHandler {
         builder(db),
         timestamp(0),
         puts()
-    {
-    }
+    {}
     void put(const vespalib::string &docId) {
         PutContext::SP pc(new PutContext(docId, builder));
-        FeedOperation::UP op(new PutOperation(pc->docCtx.bucketId,
-                                              timestamp, pc->docCtx.doc));
+        FeedOperation::UP op(new PutOperation(pc->docCtx.bucketId, timestamp, pc->docCtx.doc));
         handler.handleOperation(pc->tokenCtx.token, std::move(op));
         timestamp = Timestamp(timestamp + 1);
         puts.push_back(pc);
@@ -492,12 +480,10 @@ TEST_F("require that heartBeat calls FeedView's heartBeat",
 TEST_F("require that outdated remove is ignored", FeedHandlerFixture)
 {
     DocumentContext doc_context("doc:test:foo", *f.schema.builder);
-    FeedOperation::UP op(new RemoveOperation(doc_context.bucketId,
-                                             Timestamp(10),
-                                             doc_context.doc->getId()));
+    FeedOperation::UP op(new RemoveOperation(doc_context.bucketId, Timestamp(10), doc_context.doc->getId()));
     static_cast<DocumentOperation &>(*op).setPrevDbDocumentId(DbDocumentId(4));
     static_cast<DocumentOperation &>(*op).setPrevTimestamp(Timestamp(10000));
-    FeedTokenContext token_context(DocumentProtocol::REPLY_REMOVEDOCUMENT);
+    FeedTokenContext token_context;
     f.handler.performOperation(std::move(token_context.token_ap), std::move(op));
     EXPECT_EQUAL(0, f.feedView.remove_count);
     EXPECT_EQUAL(0, f.tls_writer.store_count);
@@ -584,15 +570,11 @@ TEST_F("require that flush cannot unprune", FeedHandlerFixture)
     EXPECT_EQUAL(10u, f.handler.getPrunedSerialNum());
 }
 
-TEST_F("require that remove of unknown document with known data type "
-       "stores remove", FeedHandlerFixture)
+TEST_F("require that remove of unknown document with known data type stores remove", FeedHandlerFixture)
 {
-    DocumentContext doc_context("id:test:searchdocument::foo",
-                                *f.schema.builder);
-    FeedOperation::UP op(new RemoveOperation(doc_context.bucketId,
-                                             Timestamp(10),
-                                             doc_context.doc->getId()));
-    FeedTokenContext token_context(DocumentProtocol::REPLY_REMOVEDOCUMENT);
+    DocumentContext doc_context("id:test:searchdocument::foo", *f.schema.builder);
+    FeedOperation::UP op(new RemoveOperation(doc_context.bucketId, Timestamp(10), doc_context.doc->getId()));
+    FeedTokenContext token_context;
     f.handler.performOperation(std::move(token_context.token_ap), std::move(op));
     EXPECT_EQUAL(1, f.feedView.remove_count);
     EXPECT_EQUAL(1, f.tls_writer.store_count);
@@ -602,10 +584,8 @@ TEST_F("require that partial update for non-existing document is tagged as such"
        FeedHandlerFixture)
 {
     UpdateContext upCtx("id:test:searchdocument::foo", *f.schema.builder);
-    FeedOperation::UP op(new UpdateOperation(upCtx.bucketId,
-                                             Timestamp(10),
-                                             upCtx.update));
-    FeedTokenContext token_context(DocumentProtocol::REPLY_UPDATEDOCUMENT);
+    FeedOperation::UP op(new UpdateOperation(upCtx.bucketId, Timestamp(10), upCtx.update));
+    FeedTokenContext token_context;
     f.handler.performOperation(std::move(token_context.token_ap), std::move(op));
     const UpdateResult *result = static_cast<const UpdateResult *>(token_context.getResult());
 
@@ -622,12 +602,9 @@ TEST_F("require that partial update for non-existing document is created if spec
     f.handler.setSerialNum(15);
     UpdateContext upCtx("id:test:searchdocument::foo", *f.schema.builder);
     upCtx.update->setCreateIfNonExistent(true);
-    f.feedView.metaStore.insert(upCtx.update->getId().getGlobalId(),
-                                MyDocumentMetaStore::Entry(5, 5, Timestamp(10)));
-    FeedOperation::UP op(new UpdateOperation(upCtx.bucketId,
-                                             Timestamp(10),
-                                             upCtx.update));
-    FeedTokenContext token_context(DocumentProtocol::REPLY_UPDATEDOCUMENT);
+    f.feedView.metaStore.insert(upCtx.update->getId().getGlobalId(), MyDocumentMetaStore::Entry(5, 5, Timestamp(10)));
+    FeedOperation::UP op(new UpdateOperation(upCtx.bucketId, Timestamp(10), upCtx.update));
+    FeedTokenContext token_context;
     f.handler.performOperation(std::move(token_context.token_ap), std::move(op));
     const UpdateResult *result = static_cast<const UpdateResult *>(token_context.getResult());
 
@@ -663,7 +640,7 @@ TEST_F("require that update is rejected if resource limit is reached", FeedHandl
 
     UpdateContext updCtx("id:test:searchdocument::foo", *f.schema.builder);
     FeedOperation::UP op = std::make_unique<UpdateOperation>(updCtx.bucketId, Timestamp(10), updCtx.update);
-    FeedTokenContext token(DocumentProtocol::REPLY_UPDATEDOCUMENT);
+    FeedTokenContext token;
     f.handler.performOperation(std::move(token.token_ap), std::move(op));
     EXPECT_EQUAL(0, f.feedView.update_count);
     EXPECT_TRUE(dynamic_cast<const UpdateResult *>(token.getResult()));
@@ -679,7 +656,7 @@ TEST_F("require that remove is NOT rejected if resource limit is reached", FeedH
 
     DocumentContext docCtx("id:test:searchdocument::foo", *f.schema.builder);
     FeedOperation::UP op = std::make_unique<RemoveOperation>(docCtx.bucketId, Timestamp(10), docCtx.doc->getId());
-    FeedTokenContext token(DocumentProtocol::REPLY_REMOVEDOCUMENT);
+    FeedTokenContext token;
     f.handler.performOperation(std::move(token.token_ap), std::move(op));
     EXPECT_EQUAL(1, f.feedView.remove_count);
     EXPECT_EQUAL(Result::NONE, token.getResult()->getErrorCode());
diff --git a/searchcore/src/tests/proton/documentdb/feedview/feedview_test.cpp b/searchcore/src/tests/proton/documentdb/feedview/feedview_test.cpp
index 6d4e5ae2478..4eefbed0a53 100644
--- a/searchcore/src/tests/proton/documentdb/feedview/feedview_test.cpp
+++ b/searchcore/src/tests/proton/documentdb/feedview/feedview_test.cpp
@@ -412,11 +412,7 @@ struct MyTransport : public FeedToken::ITransport
     MyTracer &_tracer;
     MyTransport(MyTracer &tracer);
     ~MyTransport();
-    virtual void send(mbus::Reply::UP reply,
-                      ResultUP result,
-                      bool documentWasFound,
-                      double latency_ms) override {
-        (void) reply; (void) documentWasFound, (void) latency_ms;
+    void send(ResultUP result, bool ) override {
         lastResult = std::move(result);
         _tracer.traceAck(lastResult);
         _gate.countDown();
@@ -484,36 +480,20 @@ DocumentContext::DocumentContext(const vespalib::string &docId, uint64_t timesta
 {}
 DocumentContext::~DocumentContext() {}
 
-namespace {
-
-mbus::Reply::UP
-createReply(MessageType mtype)
-{
-    if (mtype == DocumentProtocol::REPLY_UPDATEDOCUMENT) {
-        return mbus::Reply::UP(new documentapi::UpdateDocumentReply);
-    } else if (mtype == DocumentProtocol::REPLY_REMOVEDOCUMENT) {
-        return mbus::Reply::UP(new documentapi::RemoveDocumentReply);
-    } else {
-        return mbus::Reply::UP(new documentapi::DocumentReply(mtype));
-    }
-}
-
-}  // namespace
-
 struct FeedTokenContext
 {
     MyTransport mt;
     FeedToken   ft;
     typedef std::shared_ptr<FeedTokenContext> SP;
     typedef std::vector<SP> List;
-    FeedTokenContext(MyTracer &tracer, MessageType mtype);
+    FeedTokenContext(MyTracer &tracer);
     ~FeedTokenContext();
 };
 
-FeedTokenContext::FeedTokenContext(MyTracer &tracer, MessageType mtype)
-    : mt(tracer), ft(mt, createReply(mtype))
+FeedTokenContext::FeedTokenContext(MyTracer &tracer)
+    : mt(tracer), ft(mt)
 {}
-FeedTokenContext::~FeedTokenContext() {}
+FeedTokenContext::~FeedTokenContext() = default;
 
 struct FixtureBase
 {
@@ -604,7 +584,7 @@ struct FixtureBase
     }
 
     void putAndWait(const DocumentContext &docCtx) {
-        FeedTokenContext token(_tracer, DocumentProtocol::REPLY_PUTDOCUMENT);
+        FeedTokenContext token(_tracer);
         PutOperation op(docCtx.bid, docCtx.ts, docCtx.doc);
         runInMaster([&] () { performPut(&token.ft, op); });
     }
@@ -616,7 +596,7 @@ struct FixtureBase
     }
 
     void updateAndWait(const DocumentContext &docCtx) {
-        FeedTokenContext token(_tracer, DocumentProtocol::REPLY_UPDATEDOCUMENT);
+        FeedTokenContext token(_tracer);
         UpdateOperation op(docCtx.bid, docCtx.ts, docCtx.upd);
         runInMaster([&] () { performUpdate(&token.ft, op); });
     }
@@ -634,7 +614,7 @@ struct FixtureBase
     }
 
     void removeAndWait(const DocumentContext &docCtx) {
-        FeedTokenContext token(_tracer, DocumentProtocol::REPLY_REMOVEDOCUMENT);
+        FeedTokenContext token(_tracer);
         RemoveOperation op(docCtx.bid, docCtx.ts, docCtx.doc->getId());
         runInMaster([&] () { performRemove(&token.ft, op); });
     }
diff --git a/searchcore/src/tests/proton/feedtoken/feedtoken.cpp b/searchcore/src/tests/proton/feedtoken/feedtoken.cpp
index 9df65ae3437..301f21d04a1 100644
--- a/searchcore/src/tests/proton/feedtoken/feedtoken.cpp
+++ b/searchcore/src/tests/proton/feedtoken/feedtoken.cpp
@@ -1,57 +1,39 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
-#include <vespa/messagebus/emptyreply.h>
-#include <vespa/messagebus/testlib/receptor.h>
-#include <vespa/documentapi/messagebus/messages/removedocumentreply.h>
-#include <vespa/searchcore/proton/common/feedtoken.h>
 #include <vespa/vespalib/testkit/testapp.h>
+#include <vespa/searchcore/proton/common/feedtoken.h>
 #include <vespa/vespalib/util/exceptions.h>
 
 using namespace proton;
 
 class LocalTransport : public FeedToken::ITransport {
 private:
-    mbus::Receptor _receptor;
-    double         _latency_ms;
+    size_t _receivedCount;
 
 public:
     LocalTransport()
-        : _receptor(),
-          _latency_ms(0.0)
-    {
-        // empty
-    }
-
-    void send(mbus::Reply::UP reply, ResultUP, bool, double latency_ms) override {
-        _receptor.handleReply(std::move(reply));
-        _latency_ms = latency_ms;
-    }
+        : _receivedCount(0)
+    { }
 
-    mbus::Reply::UP getReply() {
-        return _receptor.getReply();
+    void send(ResultUP, bool) override {
+        _receivedCount++;
     }
 
-    double getLatencyMs() const {
-        return _latency_ms;
-    }
+    size_t getReceivedCount() const { return _receivedCount; }
 };
 
 class Test : public vespalib::TestApp {
 private:
     void testAck();
-    void testAutoReply();
     void testFail();
     void testHandover();
-    void testIntegrity();
 
 public:
     int Main() override {
         TEST_INIT("feedtoken_test");
 
         testAck();       TEST_FLUSH();
-//        testAutoReply(); TEST_FLUSH();
         testFail();      TEST_FLUSH();
         testHandover();  TEST_FLUSH();
-//        testIntegrity(); TEST_FLUSH();
 
         TEST_DONE();
     }
@@ -63,41 +45,18 @@ void
 Test::testAck()
 {
     LocalTransport transport;
-    mbus::Reply::UP msg(new documentapi::RemoveDocumentReply());
-    FeedToken token(transport, std::move(msg));
+    FeedToken token(transport);
     token.ack();
-    mbus::Reply::UP reply = transport.getReply();
-    ASSERT_TRUE(reply.get() != NULL);
-    EXPECT_TRUE(!reply->hasErrors());
-}
-
-void
-Test::testAutoReply()
-{
-    mbus::Receptor receptor;
-    mbus::Reply::UP reply(new documentapi::RemoveDocumentReply());
-    reply->pushHandler(receptor);
-    {
-        LocalTransport transport;
-        FeedToken token(transport, std::move(reply));
-    }
-    reply = receptor.getReply(0);
-    ASSERT_TRUE(reply.get() != NULL);
-    EXPECT_TRUE(reply->hasErrors());
+    EXPECT_EQUAL(1u, transport.getReceivedCount());
 }
 
 void
 Test::testFail()
 {
     LocalTransport transport;
-    mbus::Reply::UP reply(new documentapi::RemoveDocumentReply());
-    FeedToken token(transport, std::move(reply));
+    FeedToken token(transport);
     token.fail(69, "6699");
-    reply = transport.getReply();
-    ASSERT_TRUE(reply.get() != NULL);
-    EXPECT_EQUAL(1u, reply->getNumErrors());
-    EXPECT_EQUAL(69u, reply->getError(0).getCode());
-    EXPECT_EQUAL("6699", reply->getError(0).getMessage());
+    EXPECT_EQUAL(1u, transport.getReceivedCount());
 }
 
 void
@@ -110,25 +69,11 @@ Test::testHandover()
     };
 
     LocalTransport transport;
-    mbus::Reply::UP reply(new documentapi::RemoveDocumentReply());
 
-    FeedToken token(transport, std::move(reply));
+    FeedToken token(transport);
     token = MyHandover::handover(token);
     token.ack();
-    reply = transport.getReply();
-    ASSERT_TRUE(reply.get() != NULL);
-    EXPECT_TRUE(!reply->hasErrors());
+    EXPECT_EQUAL(1u, transport.getReceivedCount());
 }
 
-void
-Test::testIntegrity()
-{
-    LocalTransport transport;
-    try {
-        FeedToken token(transport, mbus::Reply::UP());
-        EXPECT_TRUE(false); // should throw an exception
-    } catch (vespalib::IllegalArgumentException &e) {
-        (void)e; // expected
-    }
-}