blob: 4cf9874562d6ef00b3c25749f9803dada815a2e4 (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
|
// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#pragma once
#include <list>
#include <sstream>
#include <vespa/storageapi/messageapi/storagecommand.h>
#include <string>
#include <vector>
#include <vespa/storage/common/storagelink.h>
#include <vespa/storageapi/message/internal.h>
namespace storage {
class DummyStorageLink : public StorageLink {
mutable std::mutex _lock; // to protect below containers:
std::vector<api::StorageMessage::SP> _commands;
std::vector<api::StorageMessage::SP> _replies;
std::list<api::StorageMessage::SP> _injected;
bool _autoReply;
bool _useDispatch;
bool _ignore;
static DummyStorageLink* _last;
std::mutex _waitMonitor;
std::condition_variable _waitCond;
public:
DummyStorageLink();
~DummyStorageLink() override;
bool onDown(const api::StorageMessage::SP&) override;
bool onUp(const api::StorageMessage::SP&) override;
void addOnTopOfChain(StorageLink& link) {
link.addTestLinkOnTop(this);
}
void print(std::ostream& ost, bool verbose, const std::string& indent) const override
{
(void) verbose;
ost << indent << "DummyStorageLink("
<< "autoreply = " << (_autoReply ? "on" : "off")
<< ", dispatch = " << (_useDispatch ? "on" : "off")
<< ", " << _commands.size() << " commands"
<< ", " << _replies.size() << " replies";
if (_injected.size() > 0)
ost << ", " << _injected.size() << " injected";
ost << ")";
}
void injectReply(api::StorageReply* reply);
void reset();
void setAutoreply(bool autoReply) { _autoReply = autoReply; }
void setIgnore(bool ignore) { _ignore = ignore; }
// Timeout is given in seconds
void waitForMessages(unsigned int msgCount = 1, int timeout = -1);
// Wait for a single message of a given type
void waitForMessage(const api::MessageType&, int timeout = -1);
api::StorageMessage::SP getCommand(size_t i) const {
std::lock_guard guard(_lock);
api::StorageMessage::SP ret = _commands[i];
return ret;
}
api::StorageMessage::SP getReply(size_t i) const {
std::lock_guard guard(_lock);
api::StorageMessage::SP ret = _replies[i];
return ret;
}
size_t getNumCommands() const {
std::lock_guard guard(_lock);
return _commands.size();
}
size_t getNumReplies() const {
std::lock_guard guard(_lock);
return _replies.size();
}
const std::vector<api::StorageMessage::SP>& getCommands() const
{ return _commands; }
const std::vector<api::StorageMessage::SP>& getReplies() const
{ return _replies; }
std::vector<api::StorageMessage::SP> getCommandsOnce() {
std::lock_guard lock(_waitMonitor);
std::vector<api::StorageMessage::SP> retval;
{
std::lock_guard guard(_lock);
retval.swap(_commands);
}
return retval;
}
std::vector<api::StorageMessage::SP> getRepliesOnce() {
std::lock_guard lock(_waitMonitor);
std::vector<api::StorageMessage::SP> retval;
{
std::lock_guard guard(_lock);
retval.swap(_replies);
}
return retval;
}
api::StorageMessage::SP getAndRemoveMessage(const api::MessageType&);
static DummyStorageLink* getLast() { return _last; }
private:
/**
* Auto-reply with an injected message if one is available and return
* whether such an injection took place.
*/
bool handleInjectedReply();
};
}
|
