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
119
120
121
122
123
124
125
126
127
128
|
/*
*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*
*/
#include "qpid/broker/QueueReplicator.h"
#include "qpid/broker/Queue.h"
#include "qpid/broker/QueueRegistry.h"
#include "qpid/framing/SequenceSet.h"
#include "qpid/log/Statement.h"
namespace qpid {
namespace broker {
QueueReplicator::QueueReplicator(const std::string& name, boost::shared_ptr<Queue> q) : Exchange(name, 0, 0), queue(q), current(queue->getPosition()) {}
QueueReplicator::~QueueReplicator() {}
namespace {
const std::string DEQUEUE_EVENT("dequeue-event");
const std::string REPLICATOR("qpid.replicator-");
}
void QueueReplicator::route(Deliverable& msg, const std::string& key, const qpid::framing::FieldTable* /*args*/)
{
if (key == DEQUEUE_EVENT) {
std::string content;
msg.getMessage().getFrames().getContent(content);
qpid::framing::Buffer buffer(const_cast<char*>(content.c_str()), content.size());
qpid::framing::SequenceSet latest;
latest.decode(buffer);
//TODO: should be able to optimise the following
for (qpid::framing::SequenceSet::iterator i = latest.begin(); i != latest.end(); i++) {
if (current < *i) {
//haven't got that far yet, record the dequeue
dequeued.add(*i);
QPID_LOG(debug, "Recording dequeue of message at " << *i << " from " << queue->getName());
} else {
QueuedMessage message;
if (queue->acquireMessageAt(*i, message)) {
queue->dequeue(0, message);
QPID_LOG(info, "Dequeued message at " << *i << " from " << queue->getName());
} else {
QPID_LOG(error, "Unable to dequeue message at " << *i << " from " << queue->getName());
}
}
}
} else {
//take account of any gaps in sequence created by messages
//dequeued before our subscription reached them
while (dequeued.contains(++current)) {
dequeued.remove(current);
QPID_LOG(debug, "Skipping dequeued message at " << current << " from " << queue->getName());
queue->setPosition(current);
}
QPID_LOG(info, "Enqueued message on " << queue->getName() << "; currently at " << current);
msg.deliverTo(queue);
}
}
bool QueueReplicator::isReplicatingLink(const std::string& name)
{
return name.find(REPLICATOR) == 0;
}
boost::shared_ptr<Exchange> QueueReplicator::create(const std::string& target, QueueRegistry& queues)
{
boost::shared_ptr<Exchange> exchange;
if (isReplicatingLink(target)) {
std::string queueName = target.substr(REPLICATOR.size());
boost::shared_ptr<Queue> queue = queues.find(queueName);
if (!queue) {
QPID_LOG(warning, "Unable to create replicator, can't find " << queueName);
} else {
//TODO: need to cache the replicator
QPID_LOG(info, "Creating replicator for " << queueName);
exchange.reset(new QueueReplicator(target, queue));
}
}
return exchange;
}
bool QueueReplicator::initReplicationSettings(const std::string& target, QueueRegistry& queues, qpid::framing::FieldTable& settings)
{
if (isReplicatingLink(target)) {
std::string queueName = target.substr(REPLICATOR.size());
boost::shared_ptr<Queue> queue = queues.find(queueName);
if (queue) {
settings.setInt("qpid.replicating-subscription", 1);
settings.setInt("qpid.high_sequence_number", queue->getPosition());
qpid::framing::SequenceNumber oldest;
if (queue->getOldest(oldest)) {
settings.setInt("qpid.low_sequence_number", oldest);
}
}
return true;
} else {
return false;
}
}
bool QueueReplicator::bind(boost::shared_ptr<Queue>, const std::string&, const qpid::framing::FieldTable*) { return false; }
bool QueueReplicator::unbind(boost::shared_ptr<Queue>, const std::string&, const qpid::framing::FieldTable*) { return false; }
bool QueueReplicator::isBound(boost::shared_ptr<Queue>, const std::string* const, const qpid::framing::FieldTable* const) { return false; }
const std::string QueueReplicator::typeName("queue-replicator");
std::string QueueReplicator::getType() const
{
return typeName;
}
}} // namespace qpid::broker
|
