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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
#ifndef QPID_HA_PRIMARY_H
#define QPID_HA_PRIMARY_H
/*
*
* 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 "types.h"
#include "hash.h"
#include "BrokerInfo.h"
#include "LogPrefix.h"
#include "PrimaryQueueLimits.h"
#include "ReplicationTest.h"
#include "Role.h"
#include "qpid/sys/Mutex.h"
#include "qpid/sys/unordered_map.h"
#include <boost/shared_ptr.hpp>
#include <boost/weak_ptr.hpp>
#include <boost/intrusive_ptr.hpp>
#include <string>
namespace qpid {
namespace broker {
class Queue;
class Connection;
class ConnectionObserver;
class BrokerObserver;
class SessionHandlerObserver;
}
namespace sys {
class TimerTask;
}
namespace ha {
class HaBroker;
class ReplicatingSubscription;
class RemoteBackup;
class QueueGuard;
class Membership;
/**
* State associated with a primary broker:
* - sets queue guards and tracks readiness of initial backups till active.
* - sets queue guards on new queues for each backup.
*
* THREAD SAFE: called concurrently in arbitrary connection threads.
*
* Locking rules: BrokerObserver create/destroy functions are called with
* the QueueRegistry lock held. Functions holding Primary::lock *must not*
* directly or indirectly call on the queue registry.
*/
class Primary : public Role
{
public:
typedef boost::shared_ptr<broker::Queue> QueuePtr;
typedef boost::shared_ptr<broker::Exchange> ExchangePtr;
typedef boost::shared_ptr<RemoteBackup> RemoteBackupPtr;
Primary(HaBroker& hb, const BrokerInfo::Set& expectedBackups);
~Primary();
// Role implementation
Role* promote();
void setBrokerUrl(const Url&) {}
void readyReplica(const ReplicatingSubscription&);
// Called via BrokerObserver
void queueCreate(const QueuePtr&);
void queueDestroy(const QueuePtr&);
void exchangeCreate(const ExchangePtr&);
void exchangeDestroy(const ExchangePtr&);
// Called via ConnectionObserver
void opened(broker::Connection& connection);
void closed(broker::Connection& connection);
boost::shared_ptr<QueueGuard> getGuard(const QueuePtr& q, const BrokerInfo&);
// Called in timer thread when the deadline for expected backups expires.
void timeoutExpectedBackups();
private:
typedef sys::unordered_map<
types::Uuid, RemoteBackupPtr, Hasher<types::Uuid> > BackupMap;
typedef std::set<RemoteBackupPtr > BackupSet;
typedef std::pair<types::Uuid, boost::shared_ptr<broker::Queue> > UuidQueue;
RemoteBackupPtr backupConnect(const BrokerInfo&, broker::Connection&, sys::Mutex::ScopedLock&);
void backupDisconnect(RemoteBackupPtr, sys::Mutex::ScopedLock&);
void checkReady();
void checkReady(RemoteBackupPtr);
void setCatchupQueues(const RemoteBackupPtr&, bool createGuards);
void deduplicate();
mutable sys::Mutex lock;
HaBroker& haBroker;
Membership& membership;
const LogPrefix& logPrefix;
bool active;
ReplicationTest replicationTest;
/**
* Set of expected backups that must be ready before we declare ourselves
* active. These are backups that were known and ready before the primary
* crashed. As new primary we expect them to re-connect.
*/
BackupSet expectedBackups;
/**
* Map of all the expected backups plus all connected backups.
*/
BackupMap backups;
boost::shared_ptr<broker::ConnectionObserver> connectionObserver;
boost::shared_ptr<broker::BrokerObserver> brokerObserver;
boost::shared_ptr<broker::SessionHandlerObserver> sessionHandlerObserver;
boost::intrusive_ptr<sys::TimerTask> timerTask;
PrimaryQueueLimits queueLimits;
};
}} // namespace qpid::ha
#endif /*!QPID_HA_PRIMARY_H*/
|
