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
|
/*
*
* Copyright (c) 2006 The Apache Software Foundation
*
* Licensed 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 "config.h"
#include "qpid/cluster/Connection.h"
#include "qpid/cluster/ConnectionCodec.h"
#include "qpid/cluster/ClusterSettings.h"
#include "qpid/cluster/SecureConnectionFactory.h"
#include "qpid/cluster/Cluster.h"
#include "qpid/cluster/ConnectionCodec.h"
#include "qpid/cluster/UpdateClient.h"
#include "qpid/broker/Broker.h"
#include "qpid/Plugin.h"
#include "qpid/Options.h"
#include "qpid/sys/AtomicValue.h"
#include "qpid/log/Statement.h"
#include "qpid/management/ManagementAgent.h"
#include "qpid/broker/Exchange.h"
#include "qpid/broker/Message.h"
#include "qpid/broker/Queue.h"
#include "qpid/broker/SessionState.h"
#include "qpid/client/ConnectionSettings.h"
#include <boost/shared_ptr.hpp>
#include <boost/utility/in_place_factory.hpp>
#include <boost/scoped_ptr.hpp>
namespace qpid {
namespace cluster {
using namespace std;
using broker::Broker;
using management::ManagementAgent;
/** Note separating options from settings to work around boost version differences.
* Old boost takes a reference to options objects, but new boost makes a copy.
* New boost allows a shared_ptr but that's not compatible with old boost.
*/
struct ClusterOptions : public Options {
ClusterSettings& settings;
ClusterOptions(ClusterSettings& v) : Options("Cluster Options"), settings(v) {
addOptions()
("cluster-name", optValue(settings.name, "NAME"), "Name of cluster to join")
("cluster-url", optValue(settings.url,"URL"),
"Set URL of this individual broker, to be advertized to clients.\n"
"Defaults to a URL listing all the local IP addresses\n")
("cluster-username", optValue(settings.username, ""), "Username for connections between brokers")
("cluster-password", optValue(settings.password, ""), "Password for connections between brokers")
("cluster-mechanism", optValue(settings.mechanism, ""), "Authentication mechanism for connections between brokers")
#if HAVE_LIBCMAN_H
("cluster-cman", optValue(settings.quorum), "Integrate with Cluster Manager (CMAN) cluster.")
#endif
("cluster-size", optValue(settings.size, "N"), "Wait for N cluster members before allowing clients to connect.")
("cluster-read-max", optValue(settings.readMax,"N"), "Experimental: flow-control limit reads per connection. 0=no limit.")
;
}
};
typedef boost::shared_ptr<sys::ConnectionCodec::Factory> CodecFactoryPtr;
struct ClusterPlugin : public Plugin {
ClusterSettings settings;
ClusterOptions options;
Cluster* cluster;
boost::scoped_ptr<ConnectionCodec::Factory> factory;
ClusterPlugin() : options(settings), cluster(0) {}
// Cluster needs to be initialized after the store
int initOrder() const { return Plugin::DEFAULT_INIT_ORDER+500; }
Options* getOptions() { return &options; }
void earlyInitialize(Plugin::Target& target) {
if (settings.name.empty()) return; // Only if --cluster-name option was specified.
Broker* broker = dynamic_cast<Broker*>(&target);
if (!broker) return;
cluster = new Cluster(settings, *broker);
CodecFactoryPtr simpleFactory(new broker::ConnectionFactory(*broker));
CodecFactoryPtr clusterFactory(new ConnectionCodec::Factory(simpleFactory, *cluster));
CodecFactoryPtr secureFactory(new SecureConnectionFactory(clusterFactory));
broker->setConnectionFactory(secureFactory);
}
void disallowManagementMethods(ManagementAgent* agent) {
if (!agent) return;
agent->disallowV1Methods();
}
void initialize(Plugin::Target& target) {
Broker* broker = dynamic_cast<Broker*>(&target);
if (broker && cluster) {
disallowManagementMethods(broker->getManagementAgent());
cluster->initialize();
}
}
};
static ClusterPlugin instance; // Static initialization.
}} // namespace qpid::cluster
|
