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
|
#ifndef _broker_BrokerMessageBase_h
#define _broker_BrokerMessageBase_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 <string>
#include <boost/shared_ptr.hpp>
#include "Content.h"
#include "framing/amqp_types.h"
namespace qpid {
namespace framing {
class MethodContext;
class ChannelAdapter;
class BasicHeaderProperties;
class FieldTable;
class AMQMethodBody;
class AMQContentBody;
class AMQHeaderBody;
}
namespace broker {
class ConnectionToken;
class MessageStore;
/**
* Base class for all types of internal broker messages
* abstracting away the operations
* TODO; AMS: for the moment this is mostly a placeholder
*/
class Message {
public:
typedef boost::shared_ptr<Message> shared_ptr;
typedef boost::shared_ptr<framing::AMQMethodBody> AMQMethodBodyPtr;
Message(const ConnectionToken* publisher_,
const std::string& _exchange,
const std::string& _routingKey,
bool _mandatory, bool _immediate,
AMQMethodBodyPtr respondTo_) :
publisher(publisher_),
exchange(_exchange),
routingKey(_routingKey),
mandatory(_mandatory),
immediate(_immediate),
persistenceId(0),
redelivered(false),
respondTo(respondTo_)
{}
Message() :
mandatory(false),
immediate(false),
persistenceId(0),
redelivered(false)
{}
virtual ~Message() {};
// Accessors
const std::string& getRoutingKey() const { return routingKey; }
const std::string& getExchange() const { return exchange; }
uint64_t getPersistenceId() const { return persistenceId; }
bool getRedelivered() const { return redelivered; }
AMQMethodBodyPtr getRespondTo() const { return respondTo; }
void setRouting(const std::string& _exchange, const std::string& _routingKey)
{ exchange = _exchange; routingKey = _routingKey; }
void setPersistenceId(uint64_t _persistenceId) { persistenceId = _persistenceId; } // XXXX: Only used in tests?
void redeliver() { redelivered = true; }
/**
* Used to deliver the message from the queue
*/
virtual void deliver(framing::ChannelAdapter& channel,
const std::string& consumerTag,
uint64_t deliveryTag,
uint32_t framesize) = 0;
/**
* Used to return a message in response to a get from a queue
*/
virtual void sendGetOk(const framing::MethodContext& context,
const std::string& destination,
uint32_t messageCount,
uint64_t deliveryTag,
uint32_t framesize) = 0;
virtual bool isComplete() = 0;
virtual uint64_t contentSize() const = 0;
// FIXME aconway 2007-02-06: Get rid of BasicHeaderProperties
// at this level. Expose only generic properties available from both
// message types (e.g. getApplicationHeaders below).
//
virtual framing::BasicHeaderProperties* getHeaderProperties() = 0;
virtual const framing::FieldTable& getApplicationHeaders() = 0;
virtual bool isPersistent() = 0;
virtual const ConnectionToken* getPublisher() const {
return publisher;
}
virtual void encode(framing::Buffer& /*buffer*/) {}; // XXXX: Only used in tests?
virtual void encodeHeader(framing::Buffer& /*buffer*/) {}; // XXXX: Only used in tests?
/**
* @returns the size of the buffer needed to encode this
* message in its entirety
*
* XXXX: Only used in tests?
*/
virtual uint32_t encodedSize() = 0;
/**
* @returns the size of the buffer needed to encode the
* 'header' of this message (not just the header frame,
* but other meta data e.g.routing key and exchange)
*
* XXXX: Only used in tests?
*/
virtual uint32_t encodedHeaderSize() = 0;
/**
* @returns the size of the buffer needed to encode the
* (possibly partial) content held by this message
*/
virtual uint32_t encodedContentSize() = 0;
/**
* If headers have been received, returns the expected
* content size else returns 0.
*/
virtual uint64_t expectedContentSize() = 0;
// TODO: AMS 29/1/2007 Don't think these are really part of base class
/**
* Sets the 'content' implementation of this message (the
* message controls the lifecycle of the content instance
* it uses).
*/
virtual void setContent(std::auto_ptr<Content>& /*content*/) {};
virtual void setHeader(boost::shared_ptr<framing::AMQHeaderBody>) {};
virtual void addContent(boost::shared_ptr<framing::AMQContentBody>) {};
/**
* Releases the in-memory content data held by this
* message. Must pass in a store from which the data can
* be reloaded.
*/
virtual void releaseContent(MessageStore* /*store*/) {};
private:
const ConnectionToken* publisher;
std::string exchange;
std::string routingKey;
const bool mandatory;
const bool immediate;
uint64_t persistenceId;
bool redelivered;
AMQMethodBodyPtr respondTo;
};
}}
#endif /*!_broker_BrokerMessage_h*/
|
