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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
|
#ifndef _QmfBrokerProxyImpl_
#define _QmfBrokerProxyImpl_
/*
* 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 "qmf/ConsoleEngine.h"
#include "qmf/ObjectImpl.h"
#include "qmf/SchemaImpl.h"
#include "qmf/ValueImpl.h"
#include "qmf/QueryImpl.h"
#include "qmf/SequenceManager.h"
#include "qmf/MessageImpl.h"
#include "qpid/framing/Buffer.h"
#include "qpid/framing/Uuid.h"
#include "qpid/sys/Mutex.h"
#include "boost/shared_ptr.hpp"
#include <memory>
#include <string>
#include <deque>
#include <map>
#include <vector>
namespace qmf {
struct MethodResponseImpl {
typedef boost::shared_ptr<MethodResponseImpl> Ptr;
MethodResponse* envelope;
uint32_t status;
SchemaMethodImpl* schema;
boost::shared_ptr<Value> exception;
boost::shared_ptr<Value> arguments;
MethodResponseImpl(const MethodResponseImpl& from);
MethodResponseImpl(qpid::framing::Buffer& buf, SchemaMethodImpl* schema);
MethodResponseImpl(uint32_t status, const std::string& text);
~MethodResponseImpl() { delete envelope; }
uint32_t getStatus() const { return status; }
const Value* getException() const { return exception.get(); }
const Value* getArgs() const { return arguments.get(); }
};
struct QueryResponseImpl {
typedef boost::shared_ptr<QueryResponseImpl> Ptr;
QueryResponse *envelope;
uint32_t status;
std::auto_ptr<Value> exception;
std::vector<ObjectImpl::Ptr> results;
QueryResponseImpl() : envelope(new QueryResponse(this)), status(0) {}
~QueryResponseImpl() { delete envelope; }
uint32_t getStatus() const { return status; }
const Value* getException() const { return exception.get(); }
uint32_t getObjectCount() const { return results.size(); }
const Object* getObject(uint32_t idx) const;
};
struct BrokerEventImpl {
typedef boost::shared_ptr<BrokerEventImpl> Ptr;
BrokerEvent::EventKind kind;
std::string name;
std::string exchange;
std::string bindingKey;
void* context;
QueryResponseImpl::Ptr queryResponse;
MethodResponseImpl::Ptr methodResponse;
BrokerEventImpl(BrokerEvent::EventKind k) : kind(k), context(0) {}
~BrokerEventImpl() {}
BrokerEvent copy();
};
struct AgentProxyImpl {
typedef boost::shared_ptr<AgentProxyImpl> Ptr;
AgentProxy* envelope;
ConsoleEngineImpl* console;
BrokerProxyImpl* broker;
uint32_t agentBank;
std::string label;
AgentProxyImpl(ConsoleEngineImpl* c, BrokerProxyImpl* b, uint32_t ab, const std::string& l) :
envelope(new AgentProxy(this)), console(c), broker(b), agentBank(ab), label(l) {}
~AgentProxyImpl() {}
const std::string& getLabel() const { return label; }
};
class BrokerProxyImpl {
public:
typedef boost::shared_ptr<BrokerProxyImpl> Ptr;
BrokerProxyImpl(BrokerProxy* e, ConsoleEngine& _console);
~BrokerProxyImpl() {}
void sessionOpened(SessionHandle& sh);
void sessionClosed();
void startProtocol();
void sendBufferLH(qpid::framing::Buffer& buf, const std::string& destination, const std::string& routingKey);
void handleRcvMessage(Message& message);
bool getXmtMessage(Message& item) const;
void popXmt();
bool getEvent(BrokerEvent& event) const;
void popEvent();
uint32_t agentCount() const;
const AgentProxy* getAgent(uint32_t idx) const;
void sendQuery(const Query& query, void* context, const AgentProxy* agent);
void sendGetRequestLH(SequenceContext::Ptr queryContext, const Query& query, const AgentProxyImpl* agent);
std::string encodeMethodArguments(const SchemaMethod* schema, const Value* args, qpid::framing::Buffer& buffer);
void sendMethodRequest(ObjectIdImpl* oid, const SchemaObjectClass* cls, const std::string& method, const Value* args, void* context);
void addBinding(const std::string& exchange, const std::string& key);
void staticRelease() { decOutstanding(); }
private:
friend class StaticContext;
friend class QueryContext;
friend class MethodContext;
mutable qpid::sys::Mutex lock;
BrokerProxy* envelope;
ConsoleEngineImpl* console;
std::string queueName;
qpid::framing::Uuid brokerId;
SequenceManager seqMgr;
uint32_t requestsOutstanding;
bool topicBound;
std::vector<AgentProxyImpl::Ptr> agentList;
std::deque<MessageImpl::Ptr> xmtQueue;
std::deque<BrokerEventImpl::Ptr> eventQueue;
# define MA_BUFFER_SIZE 65536
char outputBuffer[MA_BUFFER_SIZE];
BrokerEventImpl::Ptr eventDeclareQueue(const std::string& queueName);
BrokerEventImpl::Ptr eventBind(const std::string& exchange, const std::string& queue, const std::string& key);
BrokerEventImpl::Ptr eventSetupComplete();
BrokerEventImpl::Ptr eventStable();
BrokerEventImpl::Ptr eventQueryComplete(void* context, QueryResponseImpl::Ptr response);
BrokerEventImpl::Ptr eventMethodResponse(void* context, MethodResponseImpl::Ptr response);
void handleBrokerResponse(qpid::framing::Buffer& inBuffer, uint32_t seq);
void handlePackageIndication(qpid::framing::Buffer& inBuffer, uint32_t seq);
void handleCommandComplete(qpid::framing::Buffer& inBuffer, uint32_t seq);
void handleClassIndication(qpid::framing::Buffer& inBuffer, uint32_t seq);
MethodResponseImpl::Ptr handleMethodResponse(qpid::framing::Buffer& inBuffer, uint32_t seq, SchemaMethodImpl* schema);
void handleHeartbeatIndication(qpid::framing::Buffer& inBuffer, uint32_t seq);
void handleEventIndication(qpid::framing::Buffer& inBuffer, uint32_t seq);
void handleSchemaResponse(qpid::framing::Buffer& inBuffer, uint32_t seq);
ObjectImpl::Ptr handleObjectIndication(qpid::framing::Buffer& inBuffer, uint32_t seq, bool prop, bool stat);
void incOutstandingLH();
void decOutstanding();
};
//
// StaticContext is used to handle:
//
// 1) Responses to console-level requests (for schema info, etc.)
// 2) Unsolicited messages from agents (events, published updates, etc.)
//
struct StaticContext : public SequenceContext {
StaticContext(BrokerProxyImpl& b) : broker(b) {}
virtual ~StaticContext() {}
void reserve() {}
void release() { broker.staticRelease(); }
bool handleMessage(uint8_t opcode, uint32_t sequence, qpid::framing::Buffer& buffer);
BrokerProxyImpl& broker;
};
//
// QueryContext is used to track and handle responses associated with a single Get Query
//
struct QueryContext : public SequenceContext {
QueryContext(BrokerProxyImpl& b, void* u) :
broker(b), userContext(u), requestsOutstanding(0), queryResponse(new QueryResponseImpl()) {}
virtual ~QueryContext() {}
void reserve();
void release();
bool handleMessage(uint8_t opcode, uint32_t sequence, qpid::framing::Buffer& buffer);
mutable qpid::sys::Mutex lock;
BrokerProxyImpl& broker;
void* userContext;
uint32_t requestsOutstanding;
QueryResponseImpl::Ptr queryResponse;
};
struct MethodContext : public SequenceContext {
MethodContext(BrokerProxyImpl& b, void* u, SchemaMethodImpl* s) : broker(b), userContext(u), schema(s) {}
virtual ~MethodContext() {}
void reserve() {}
void release();
bool handleMessage(uint8_t opcode, uint32_t sequence, qpid::framing::Buffer& buffer);
BrokerProxyImpl& broker;
void* userContext;
SchemaMethodImpl* schema;
MethodResponseImpl::Ptr methodResponse;
};
}
#endif
|
