Annotation of parser3/src/classes/amqp.C, revision 1.29
1.1 moko 1: /** @file
2: Parser: @b amqp parser class.
3:
1.13 moko 4: Copyright (c) 2001-2026 Art. Lebedev Studio (https://www.artlebedev.com)
1.1 moko 5: Authors: Konstantin Morshnev <moko@design.ru>
6: */
7:
1.25 moko 8:
9: #include "classes.h"
1.1 moko 10: #include "pa_vmethod_frame.h"
11:
1.25 moko 12: #include "pa_vamqp.h"
1.1 moko 13: #include "pa_request.h"
1.8 moko 14: #include "pa_varray.h"
1.1 moko 15: #include "pa_vbool.h"
1.16 moko 16: #include "pa_os.h"
1.1 moko 17:
1.29 ! moko 18: volatile const char * IDENT_AMQP_C="$Id: amqp.C,v 1.28 2026/10/09 21:10:38 moko Exp $" IDENT_PA_VAMQP_H;
1.1 moko 19:
20: class MAmqp: public Methoded {
21: public: // VStateless_class
1.28 moko 22: Value* create_new_value(Pool& apool) { return new VAmqp(apool); }
1.1 moko 23: public:
24: MAmqp();
25: };
26:
27: DECLARE_CLASS_VAR(amqp, new MAmqp);
28:
1.6 moko 29: #ifdef WITH_AMQP
30:
1.16 moko 31: static void status_check(VAmqp& self, int ret, const char *detail=""){
1.6 moko 32: if(ret == AMQP_STATUS_OK)
33: return;
34:
1.16 moko 35: self.fstate = VAmqp::CONNECTION_DEAD;
36:
1.6 moko 37: const char* error_str = amqp_error_string2(ret);
38: if(error_str) {
39: throw Exception("amqp", 0, "%sfailed: %s", detail, error_str);
40: } else {
41: throw Exception("amqp", 0, "%sfailed: error %d", detail, ret);
42: }
43: }
44:
1.16 moko 45: static void check(VAmqp& self, amqp_rpc_reply_t rr, const char *detail=""){
1.6 moko 46: if(rr.reply_type == AMQP_RESPONSE_NORMAL)
47: return;
48:
1.16 moko 49: self.fstate = (rr.reply_type == AMQP_RESPONSE_SERVER_EXCEPTION && rr.reply.id == AMQP_CHANNEL_CLOSE_METHOD) ? VAmqp::CHANNEL_DEAD : VAmqp::CONNECTION_DEAD;
50:
1.6 moko 51: // Extract error message from reply
52: const char* error_msg = 0;
53: size_t error_len = 0;
54: if(rr.reply_type == AMQP_RESPONSE_SERVER_EXCEPTION) {
55: if(rr.reply.id == AMQP_CHANNEL_CLOSE_METHOD) {
56: amqp_channel_close_t *m = (amqp_channel_close_t *)rr.reply.decoded;
57: if(m->reply_text.len > 0 && m->reply_text.bytes) {
58: error_msg = (const char*)m->reply_text.bytes;
59: error_len = m->reply_text.len;
60: }
61: } else if(rr.reply.id == AMQP_CONNECTION_CLOSE_METHOD) {
62: amqp_connection_close_t *m = (amqp_connection_close_t *)rr.reply.decoded;
63: if(m->reply_text.len > 0 && m->reply_text.bytes) {
64: error_msg = (const char*)m->reply_text.bytes;
65: error_len = m->reply_text.len;
66: }
67: }
68: }
1.16 moko 69:
1.6 moko 70: if(error_msg) {
71: throw Exception("amqp", 0, "%sfailed: %.*s", detail, (int)error_len, error_msg);
72: } else if(rr.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION) {
1.16 moko 73: status_check(self, rr.library_error, detail);
1.6 moko 74: }
75:
76: throw Exception("amqp", 0, "%sfailed", detail);
77: }
78:
1.29 ! moko 79: // destroys the new connection if connecting fails
! 80: class Temp_amqp_connection {
! 81: amqp_connection_state_t fconnection;
! 82: public:
! 83: Temp_amqp_connection(amqp_connection_state_t aconnection): fconnection(aconnection) {}
! 84: ~Temp_amqp_connection() { if(fconnection) amqp_destroy_connection(fconnection); }
! 85: amqp_connection_state_t release() { amqp_connection_state_t result=fconnection; fconnection=0; return result; }
! 86: };
! 87:
1.16 moko 88: static void amqp_connect(VAmqp& self, Request& r, HashStringValue* options) {
1.1 moko 89: const char* host_c = "localhost";
90: int port = 5672;
91: const char* user_c = "guest";
92: const char* pass_c = "guest";
93: const char* vhost_c = "/";
94: const char* locale_c = "en_US";
1.22 moko 95: int heartbeat_sec = 30;
96: int reconnect_interval_sec = 0;
97: bool tls_specified = false;
1.26 moko 98: #ifdef WITH_AMQP_SSL
1.5 moko 99: const char* tls_ca = 0;
100: const char* tls_cert = 0;
101: const char* tls_key = 0;
102: bool tls_verify = true;
1.22 moko 103: #endif
1.1 moko 104:
1.16 moko 105: if(options){
106: for(HashStringValue::Iterator i(*options); i; i.next()){
107: String::Body key=i.key();
108: Value* value=i.value();
109: if(key=="host"){
110: host_c=value->as_string().cstr();
111: } else if(key=="port"){
112: port=r.process(*value).as_int();
113: } else if(key=="user"){
114: user_c=value->as_string().cstr();
115: } else if(key=="password"){
116: pass_c=value->as_string().cstr();
117: } else if(key=="vhost"){
118: vhost_c=value->as_string().cstr();
119: } else if(key=="locale"){
120: locale_c=value->as_string().cstr();
121: } else if(key=="heartbeat"){
1.22 moko 122: heartbeat_sec=r.process(*value).as_int();
1.16 moko 123: } else if(key=="auto_reconnect"){
124: reconnect_interval_sec=r.process(*value).as_int();
125: } else if(key=="tls"){
1.26 moko 126: #ifdef WITH_AMQP_SSL
1.16 moko 127: tls_specified = true;
128: if(HashStringValue* tls_options=value->get_hash()){
129: for(HashStringValue::Iterator t(*tls_options); t; t.next()){
130: String::Body tkey=t.key();
131: Value* tval=t.value();
132: if(tkey=="ca"){
133: tls_ca=tval->as_string().cstr();
134: } else if(tkey=="cert"){
135: tls_cert=tval->as_string().cstr();
136: } else if(tkey=="key"){
137: tls_key=tval->as_string().cstr();
138: } else if(tkey=="verify"){
139: tls_verify=r.process(*tval).as_bool();
140: } else
141: throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
1.5 moko 142: }
1.16 moko 143: }
1.22 moko 144: #else
145: throw Exception("amqp", 0, "AMQP SSL support is disabled");
146: #endif
1.16 moko 147: } else
148: throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
1.1 moko 149: }
150: }
151:
152: amqp_connection_state_t conn = amqp_new_connection();
1.29 ! moko 153: Temp_amqp_connection temp_connection(conn);
1.5 moko 154: amqp_socket_t* socket = 0;
1.16 moko 155:
1.26 moko 156: #ifdef WITH_AMQP_SSL
1.5 moko 157: if(tls_specified) {
158: socket = amqp_ssl_socket_new(conn);
159: if(!socket)
160: throw Exception("amqp", 0, "failed to create SSL socket");
1.16 moko 161:
1.5 moko 162: // Set CA certificate if provided
163: if(tls_ca)
164: if(amqp_ssl_socket_set_cacert(socket, tls_ca))
165: throw Exception("amqp", 0, "failed to set CA certificate");
1.6 moko 166:
1.5 moko 167: // Set client certificate and key if provided
168: if(tls_cert && tls_key) {
169: if(amqp_ssl_socket_set_key(socket, tls_cert, tls_key))
170: throw Exception("amqp", 0, "failed to set client certificate/key");
171: } else if(tls_cert || tls_key) {
172: throw Exception("amqp", 0, "both cert and key must be specified for TLS");
173: }
1.6 moko 174:
1.5 moko 175: // If CA is provided, peer verification will use it
176: amqp_ssl_socket_set_verify_peer(socket, tls_verify && tls_ca);
177: // If verify=true, enable hostname verification
178: amqp_ssl_socket_set_verify_hostname(socket, tls_verify);
1.22 moko 179: } else
180: #endif
181: {
1.5 moko 182: socket = amqp_tcp_socket_new(conn);
183: if(!socket)
184: throw Exception("amqp", 0, "failed to create TCP socket");
185: }
1.16 moko 186:
187: status_check(self, amqp_socket_open(socket, host_c, port), tls_specified ? "open SSL socket " : "open TCP socket ");
1.1 moko 188:
1.22 moko 189: amqp_rpc_reply_t rlogin = amqp_login(conn, vhost_c, 0, 131072, heartbeat_sec, AMQP_SASL_METHOD_PLAIN, user_c, pass_c);
1.29 ! moko 190: if(rlogin.reply_type != AMQP_RESPONSE_NORMAL)
1.16 moko 191: check(self, rlogin, "login ");
1.1 moko 192:
1.23 moko 193: amqp_channel_t channel = 1;
1.1 moko 194: amqp_channel_open(conn, channel);
195: amqp_rpc_reply_t ropen = amqp_get_rpc_reply(conn);
196: if(ropen.reply_type != AMQP_RESPONSE_NORMAL){
197: amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
1.16 moko 198: check(self, ropen, "open channel ");
1.1 moko 199: }
200:
1.16 moko 201: if(self.fconnection)
202: amqp_destroy_connection(self.fconnection);
203:
1.29 ! moko 204: self.fconnection = temp_connection.release();
1.1 moko 205: self.fchannel = channel;
1.16 moko 206: self.fstate = VAmqp::ALIVE;
207: self.freconnect_interval = reconnect_interval_sec;
208: }
209:
1.19 moko 210: // repairs a channel-level error cheaply: reopen on the same connection. Only valid when the server has
211: // already sent (or is about to send) channel.close - we're completing that handshake with close-ok, not
212: // initiating our own.
1.16 moko 213: static void amqp_repair_channel(VAmqp& self) {
214: amqp_channel_close_ok_t close_ok;
215: amqp_send_method(self.fconnection, self.fchannel, AMQP_CHANNEL_CLOSE_OK_METHOD, &close_ok);
216: amqp_channel_open(self.fconnection, self.fchannel);
217: check(self, amqp_get_rpc_reply(self.fconnection), "channel repair ");
218: self.fstate = VAmqp::ALIVE;
219: }
220:
1.19 moko 221: // voluntarily closes and reopens an otherwise-healthy channel
222: static void amqp_reset_channel(VAmqp& self) {
223: amqp_channel_close(self.fconnection, self.fchannel, AMQP_REPLY_SUCCESS);
224: amqp_channel_open(self.fconnection, self.fchannel);
225: amqp_get_rpc_reply(self.fconnection);
226: }
227:
1.16 moko 228: void VAmqp::ensure_connected() {
229: if(fstate == ALIVE)
230: return;
231:
232: if(fstate == CHANNEL_DEAD){
233: amqp_repair_channel(*this);
234: return;
235: }
236:
237: // CONNECTION_DEAD
238: if(freconnect_interval <= 0)
239: throw Exception("amqp", 0, "connection is dead (auto_reconnect is not enabled)");
240:
241: pa_sleep(freconnect_interval, 0);
242:
243: amqp_connect(*this, pa_thread_request(), fcreate_options);
244: }
245:
246: #endif // WITH_AMQP
247:
248: static void _create(Request& r, MethodParams& params) {
249: VAmqp& self=GET_SELF(r, VAmqp);
250:
251: #ifdef WITH_AMQP
252: self.fcreate_options = params.count()>0 ? params.as_hash(0) : 0;
253: amqp_connect(self, r, self.fcreate_options);
1.1 moko 254: #else
255: (void)params; (void)self;
256: throw Exception("amqp", 0, "compiled without amqp support");
1.6 moko 257: #endif // WITH_AMQP
1.1 moko 258: }
259:
260: #ifdef WITH_AMQP
261:
1.2 moko 262: #define AMQP_STRING(s,l) new String(String::C(pa_strdup((const char*)(s), (l)), (l)))
263: #define AMQP_VSTRING(s,l) new VString(*AMQP_STRING(s,l))
264:
1.17 moko 265: // broker-defined arguments/headers (x-message-ttl, x-dead-letter-exchange, ...)
266: // Type is taken from how the value was assigned: [] -> string, () -> number, (true|false) -> bool
267: static amqp_table_t amqp_build_arguments_table(Request& r, HashStringValue* arguments) {
268: amqp_table_t table = amqp_empty_table;
269: size_t count = arguments ? arguments->count() : 0;
270: if(!count)
271: return table;
272:
273: amqp_table_entry_t* entries = (amqp_table_entry_t*)pa_malloc(count * sizeof(amqp_table_entry_t));
274: size_t i = 0;
275: for(HashStringValue::Iterator it(*arguments); it; it.next()){
276: entries[i].key = amqp_cstring_bytes(it.key().cstr());
277:
278: Value& value = r.process(*it.value());
279: if(value.is_string() || value.is_void()){
280: entries[i].value.kind = AMQP_FIELD_KIND_UTF8;
281: entries[i].value.value.bytes = amqp_cstring_bytes(value.as_string().cstr());
282: } else if(value.is_bool()){
283: entries[i].value.kind = AMQP_FIELD_KIND_BOOLEAN;
284: entries[i].value.value.boolean = value.as_bool();
285: } else {
1.18 moko 286: double d = value.as_double();
287: double rounded = round(d);
288: if(ulp_eq_double(d, rounded)){
289: entries[i].value.kind = AMQP_FIELD_KIND_I64;
290: entries[i].value.value.i64 = (int64_t)rounded;
291: } else {
292: entries[i].value.kind = AMQP_FIELD_KIND_F64;
293: entries[i].value.value.f64 = d;
294: }
1.17 moko 295: }
296: i++;
297: }
298:
299: table.num_entries = (int)count;
300: table.entries = entries;
301: return table;
302: }
303:
304: // reverse of amqp_build_arguments_table - decodes a broker-provided field table (e.g. message headers) back into a hash.
305: static VHash* amqp_table_to_hash(amqp_table_t& table);
306:
307: static Value* amqp_field_to_value(amqp_field_value_t& f) {
308: switch(f.kind){
309: case AMQP_FIELD_KIND_UTF8:
310: case AMQP_FIELD_KIND_BYTES:
311: return AMQP_VSTRING(f.value.bytes.bytes, f.value.bytes.len);
312: case AMQP_FIELD_KIND_BOOLEAN:
313: return &VBool::get(f.value.boolean);
314: case AMQP_FIELD_KIND_I8: return new VInt((pa_wint)f.value.i8);
315: case AMQP_FIELD_KIND_U8: return new VInt((pa_wint)f.value.u8);
316: case AMQP_FIELD_KIND_I16: return new VInt((pa_wint)f.value.i16);
317: case AMQP_FIELD_KIND_U16: return new VInt((pa_wint)f.value.u16);
318: case AMQP_FIELD_KIND_I32: return new VInt((pa_wint)f.value.i32);
319: case AMQP_FIELD_KIND_U32: return new VInt((pa_wint)f.value.u32);
320: case AMQP_FIELD_KIND_I64: return new VInt((pa_wint)f.value.i64);
321: case AMQP_FIELD_KIND_U64: return new VInt((pa_wint)f.value.u64);
322: case AMQP_FIELD_KIND_TIMESTAMP: return new VInt((pa_wint)f.value.u64);
323: case AMQP_FIELD_KIND_F32: return new VDouble((double)f.value.f32);
324: case AMQP_FIELD_KIND_F64: return new VDouble(f.value.f64);
325: case AMQP_FIELD_KIND_DECIMAL: {
326: double divisor=1;
327: for(uint8_t i=0; i<f.value.decimal.decimals; i++) divisor*=10;
328: return new VDouble(f.value.decimal.value/divisor);
329: }
330: case AMQP_FIELD_KIND_TABLE:
331: return amqp_table_to_hash(f.value.table);
332: case AMQP_FIELD_KIND_ARRAY: {
333: VArray* result=new VArray();
334: ArrayValue& av=result->array();
335: for(int i=0; i<f.value.array.num_entries; i++)
336: av+=amqp_field_to_value(f.value.array.entries[i]);
337: return result;
338: }
339: case AMQP_FIELD_KIND_VOID:
340: return VVoid::get();
341: default:
342: return VString::empty();
343: }
344: }
345:
346: static VHash* amqp_table_to_hash(amqp_table_t& table) {
347: VHash* result=new VHash;
348: HashStringValue* h=result->get_hash();
349: for(int i=0; i<table.num_entries; i++){
350: amqp_table_entry_t& e=table.entries[i];
351: h->put(*AMQP_STRING(e.key.bytes, e.key.len), amqp_field_to_value(e.value));
352: }
353: return result;
354: }
355:
1.1 moko 356: static void _publish(Request& r, MethodParams& params) {
357: VAmqp& self=GET_SELF(r, VAmqp);
358: const String &msg=params.as_string(0, "msg must be string");
359: const char* exchange_c = ""; // default exchange
360: const char* routing_key_c = 0;
361: bool mandatory=false;
362:
363: amqp_basic_properties_t props;
364: props._flags = 0;
365:
366: if(params.count()>1){
367: if(HashStringValue* options=params.as_hash(1)){
368: for(HashStringValue::Iterator i(*options); i; i.next()){
369: String::Body key=i.key();
370: Value* value=i.value();
371: if(key=="exchange"){
372: exchange_c=value->as_string().cstr();
373: } else if(key=="routing_key"){
374: routing_key_c=value->as_string().cstr();
375: } else if(key=="queue"){
1.3 moko 376: routing_key_c=value->as_string().cstr();
1.1 moko 377: } else if(key=="mandatory"){
378: mandatory=r.process(*value).as_bool();
1.3 moko 379: } else if(key=="content_type"){
380: const char* v=value->as_string().cstr();
381: props.content_type=amqp_cstring_bytes(v);
382: props._flags|=AMQP_BASIC_CONTENT_TYPE_FLAG;
383: } else if(key=="content_encoding"){
384: const char* v=value->as_string().cstr();
385: props.content_encoding=amqp_cstring_bytes(v);
386: props._flags|=AMQP_BASIC_CONTENT_ENCODING_FLAG;
387: } else if(key=="delivery_mode"){
388: uint8_t dm=(uint8_t)value->as_int();
389: props.delivery_mode=dm;
390: props._flags|=AMQP_BASIC_DELIVERY_MODE_FLAG;
391: } else if(key=="priority"){
392: uint8_t pr=(uint8_t)value->as_int();
393: props.priority=pr;
394: props._flags|=AMQP_BASIC_PRIORITY_FLAG;
395: } else if(key=="correlation_id"){
396: const char* v=value->as_string().cstr();
397: props.correlation_id=amqp_cstring_bytes(v);
398: props._flags|=AMQP_BASIC_CORRELATION_ID_FLAG;
399: } else if(key=="reply_to"){
400: const char* v=value->as_string().cstr();
401: props.reply_to=amqp_cstring_bytes(v);
402: props._flags|=AMQP_BASIC_REPLY_TO_FLAG;
403: } else if(key=="expiration"){
404: const char* v=value->as_string().cstr();
405: props.expiration=amqp_cstring_bytes(v);
406: props._flags|=AMQP_BASIC_EXPIRATION_FLAG;
407: } else if(key=="message_id"){
408: const char* v=value->as_string().cstr();
409: props.message_id=amqp_cstring_bytes(v);
410: props._flags|=AMQP_BASIC_MESSAGE_ID_FLAG;
411: } else if(key=="timestamp"){
412: uint64_t ts=(uint64_t)value->as_double();
413: props.timestamp=ts;
414: props._flags|=AMQP_BASIC_TIMESTAMP_FLAG;
415: } else if(key=="type"){
416: const char* v=value->as_string().cstr();
417: props.type=amqp_cstring_bytes(v);
418: props._flags|=AMQP_BASIC_TYPE_FLAG;
419: } else if(key=="user_id"){
420: const char* v=value->as_string().cstr();
421: props.user_id=amqp_cstring_bytes(v);
422: props._flags|=AMQP_BASIC_USER_ID_FLAG;
423: } else if(key=="app_id"){
424: const char* v=value->as_string().cstr();
425: props.app_id=amqp_cstring_bytes(v);
426: props._flags|=AMQP_BASIC_APP_ID_FLAG;
1.18 moko 427: } else if(key=="cluster_id"){
428: const char* v=value->as_string().cstr();
429: props.cluster_id=amqp_cstring_bytes(v);
430: props._flags|=AMQP_BASIC_CLUSTER_ID_FLAG;
1.3 moko 431: } else if(key=="headers"){
1.17 moko 432: props.headers=amqp_build_arguments_table(r, value->get_hash());
433: props._flags|=AMQP_BASIC_HEADERS_FLAG;
434: } else
1.1 moko 435: throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
436: }
437: }
438: }
439:
440: if(!routing_key_c)
441: throw Exception("amqp", 0, "routing_key or queue must be specified");
442:
443: amqp_bytes_t body;
444: body.len = msg.length();
445: body.bytes=(void*)msg.cstr();
446:
1.3 moko 447: int ret = amqp_basic_publish(self.connection(), self.channel(), amqp_cstring_bytes(exchange_c), amqp_cstring_bytes(routing_key_c), mandatory, 0, &props, body);
1.1 moko 448:
1.16 moko 449: status_check(self, ret, "publish ");
1.1 moko 450: }
451:
452: static void _release(Request& r, MethodParams&) {
1.28 moko 453: GET_SELF(r, VAmqp).release();
1.1 moko 454: }
455:
1.15 moko 456: // to allow both [] and () call syntax
457: static uint64_t as_delivery_tag(MethodParams& params, int index=0) {
458: const String &tag_s=params.as_string(index, "delivery tag must not be code");
459: return pa_atoul(tag_s.cstr());
460: }
461:
1.1 moko 462: static void _ack(Request& r, MethodParams& params) {
463: VAmqp& self=GET_SELF(r, VAmqp);
1.15 moko 464: int ret = amqp_basic_ack(self.connection(), self.channel(), as_delivery_tag(params), 0);
1.16 moko 465: status_check(self, ret, "ack ");
1.1 moko 466: }
467:
468: static void _nack(Request& r, MethodParams& params) {
469: VAmqp& self=GET_SELF(r, VAmqp);
1.15 moko 470: uint64_t tag=as_delivery_tag(params);
1.1 moko 471: bool requeue=false;
472: if(params.count()>1){
473: if(HashStringValue* options=params.as_hash(1)){
474: for(HashStringValue::Iterator i(*options); i; i.next()){
475: if(i.key()=="requeue"){
476: requeue=r.process(*i.value()).as_bool();
477: } else
478: throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
479: }
480: }
481: }
1.15 moko 482: int ret = amqp_basic_nack(self.connection(), self.channel(), tag, 0, requeue);
1.16 moko 483: status_check(self, ret, "nack ");
1.1 moko 484: }
485:
486: static void _qos(Request& r, MethodParams& params) {
487: VAmqp& self=GET_SELF(r, VAmqp);
488: uint16_t prefetch_count=0;
489: if(params.count()>0){
490: if(HashStringValue* options=params.as_hash(0)){
491: for(HashStringValue::Iterator i(*options); i; i.next()){
492: if(i.key()=="prefetch_count"){
493: int pc=r.process(*i.value()).as_int();
494: prefetch_count= pc<0 ? 0 : (pc>65535 ? 65535 : (uint16_t)pc);
495: } else
496: throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
497: }
498: }
499: }
1.16 moko 500: amqp_basic_qos(self.connection(), self.channel(), 0, prefetch_count, 0);
501: check(self, amqp_get_rpc_reply(self.connection()), "qos ");
1.1 moko 502: }
503:
504: static void _reject(Request& r, MethodParams& params) {
505: VAmqp& self=GET_SELF(r, VAmqp);
1.15 moko 506: uint64_t tag=as_delivery_tag(params);
1.1 moko 507: bool requeue=true; // by default return to queue
508: if(params.count()>1){
509: if(HashStringValue* options = params.as_hash(1)){
510: for(HashStringValue::Iterator i(*options); i; i.next()){
511: if(i.key() == "requeue"){
512: requeue=r.process(*i.value()).as_bool();
513: } else
514: throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
515: }
516: }
517: }
1.15 moko 518: int ret = amqp_basic_reject(self.connection(), self.channel(), tag, requeue);
1.16 moko 519: status_check(self, ret, "reject ");
1.1 moko 520: }
521:
1.8 moko 522: static void _declare(Request& r, MethodParams& params) {
1.1 moko 523: VAmqp& self=GET_SELF(r, VAmqp);
1.8 moko 524: const char* exchange_c = 0;
525: const char* queue_c = 0;
1.1 moko 526: const char* type_c = "direct";
1.12 moko 527: bool passive=false, durable=false, auto_delete=false, internal=false, exclusive=false;
1.17 moko 528: amqp_table_t arguments=amqp_empty_table;
1.8 moko 529: if(HashStringValue* options=params.as_hash(0)){
530: for(HashStringValue::Iterator i(*options); i; i.next()){
531: String::Body key=i.key();
532: Value* value=i.value();
533: if(key=="exchange"){
534: exchange_c=value->as_string().cstr();
535: } else if(key=="queue"){
536: queue_c=value->as_string().cstr();
537: } else if(key=="type"){
538: type_c=value->as_string().cstr();
539: } else if(key=="passive"){
540: passive=r.process(*value).as_bool();
541: } else if(key=="durable"){
542: durable=r.process(*value).as_bool();
543: } else if(key=="auto_delete"){
544: auto_delete=r.process(*value).as_bool();
1.11 moko 545: } else if(key=="internal"){
546: internal=r.process(*value).as_bool();
547: } else if(key=="exclusive"){
548: exclusive=r.process(*value).as_bool();
1.17 moko 549: } else if(key=="arguments"){
550: arguments=amqp_build_arguments_table(r, value->get_hash());
1.8 moko 551: } else
552: throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
1.1 moko 553: }
554: }
1.8 moko 555: if(!exchange_c && !queue_c)
556: throw Exception("amqp", 0, "exchange or queue must be specified");
557:
558: if(exchange_c){
1.17 moko 559: amqp_exchange_declare(self.connection(), self.channel(), amqp_cstring_bytes(exchange_c), amqp_cstring_bytes(type_c), passive, durable, auto_delete, internal, arguments);
1.16 moko 560: check(self, amqp_get_rpc_reply(self.connection()));
1.8 moko 561: }
1.1 moko 562:
1.8 moko 563: if(queue_c){
1.17 moko 564: amqp_queue_declare_ok_t *ok = amqp_queue_declare(self.connection(), self.channel(), *queue_c ? amqp_cstring_bytes(queue_c) : amqp_empty_bytes, passive, durable, exclusive, auto_delete, arguments);
1.16 moko 565: check(self, amqp_get_rpc_reply(self.connection()));
1.8 moko 566: if(!*queue_c && ok){
567: r.write(*AMQP_STRING(ok->queue.bytes, ok->queue.len));
1.1 moko 568: }
569: }
570: }
571:
1.8 moko 572: static void _delete(Request& r, MethodParams& params) {
1.1 moko 573: VAmqp& self=GET_SELF(r, VAmqp);
1.8 moko 574: const char* exchange_c = 0;
575: const char* queue_c = 0;
1.11 moko 576: bool if_unused=false, if_empty=false;
1.8 moko 577: if(HashStringValue* options=params.as_hash(0)){
578: for(HashStringValue::Iterator i(*options); i; i.next()){
579: String::Body key=i.key();
580: Value* value=i.value();
581: if(key=="exchange"){
582: exchange_c=value->as_string().cstr();
583: } else if(key=="queue"){
584: queue_c=value->as_string().cstr();
585: } else if(key=="if_unused"){
586: if_unused=r.process(*value).as_bool();
587: } else if(key=="if_empty"){
588: if_empty=r.process(*value).as_bool();
589: } else
590: throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
1.1 moko 591: }
592: }
1.8 moko 593: if(!exchange_c && !queue_c)
594: throw Exception("amqp", 0, "exchange or queue must be specified");
595:
596: if(exchange_c){
597: amqp_exchange_delete(self.connection(), self.channel(), amqp_cstring_bytes(exchange_c), if_unused);
1.16 moko 598: check(self, amqp_get_rpc_reply(self.connection()));
1.1 moko 599: }
600:
1.8 moko 601: if(queue_c){
602: amqp_queue_delete(self.connection(), self.channel(), amqp_cstring_bytes(queue_c), if_unused, if_empty);
1.16 moko 603: check(self, amqp_get_rpc_reply(self.connection()));
1.1 moko 604: }
605: }
606:
1.8 moko 607: static void _bind(Request& r, MethodParams& params) {
1.1 moko 608: VAmqp& self=GET_SELF(r, VAmqp);
609: const char* exchange_c=0;
610: const char* queue_c=0;
611: const char* routing_key_c="";
1.8 moko 612: if(HashStringValue* options=params.as_hash(0)){
613: for(HashStringValue::Iterator i(*options); i; i.next()){
614: String::Body key=i.key();
615: Value* value=i.value();
616: if(key=="exchange"){
617: exchange_c=value->as_string().cstr();
618: } else if(key=="queue"){
619: queue_c=value->as_string().cstr();
620: } else if(key=="routing_key"){
621: routing_key_c=value->as_string().cstr();
622: } else
623: throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
1.1 moko 624: }
625: }
626: if(!exchange_c || !queue_c) throw Exception("amqp", 0, "exchange and queue are required");
627: amqp_queue_bind(self.connection(), self.channel(), amqp_cstring_bytes(queue_c), amqp_cstring_bytes(exchange_c), amqp_cstring_bytes(routing_key_c), amqp_empty_table);
1.16 moko 628: check(self, amqp_get_rpc_reply(self.connection()));
1.1 moko 629: }
630:
1.8 moko 631: static void _unbind(Request& r, MethodParams& params) {
1.1 moko 632: VAmqp& self=GET_SELF(r, VAmqp);
633: const char* exchange_c=0;
634: const char* queue_c=0;
635: const char* routing_key_c="";
1.8 moko 636: if(HashStringValue* options=params.as_hash(0)){
637: for(HashStringValue::Iterator i(*options); i; i.next()){
638: String::Body key=i.key();
639: Value* value=i.value();
640: if(key=="exchange"){
641: exchange_c=value->as_string().cstr();
642: } else if(key=="queue"){
643: queue_c=value->as_string().cstr();
644: } else if(key=="routing_key"){
645: routing_key_c=value->as_string().cstr();
646: } else
647: throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
1.1 moko 648: }
649: }
650: if(!exchange_c || !queue_c) throw Exception("amqp", 0, "exchange and queue are required");
651: amqp_queue_unbind(self.connection(), self.channel(), amqp_cstring_bytes(queue_c), amqp_cstring_bytes(exchange_c), amqp_cstring_bytes(routing_key_c), amqp_empty_table);
1.16 moko 652: check(self, amqp_get_rpc_reply(self.connection()));
1.1 moko 653: }
654:
1.8 moko 655: static void _purge(Request& r, MethodParams& params) {
656: VAmqp& self=GET_SELF(r, VAmqp);
657: const char* queue_c = 0;
658: if(HashStringValue* options=params.as_hash(0)){
659: for(HashStringValue::Iterator i(*options); i; i.next()){
660: String::Body key=i.key();
661: Value* value=i.value();
662: if(key=="queue"){
663: queue_c=value->as_string().cstr();
664: } else
665: throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
666: }
667: }
668: if(!queue_c)
669: throw Exception("amqp", 0, "queue must be specified");
670:
1.9 moko 671: amqp_queue_purge_ok_t *ok = amqp_queue_purge(self.connection(), self.channel(), amqp_cstring_bytes(queue_c));
1.16 moko 672: check(self, amqp_get_rpc_reply(self.connection()));
1.9 moko 673: r.write(*new VInt(ok ? ok->message_count : 0));
1.8 moko 674: }
675:
676: static void _info(Request& r, MethodParams& params) {
677: VAmqp& self = GET_SELF(r, VAmqp);
678: const char* queue_c = 0;
679:
680: if (HashStringValue* options=params.as_hash(0)) {
681: for (HashStringValue::Iterator i(*options); i; i.next()) {
682: String::Body key=i.key();
683: Value* value=i.value();
684: if(key=="queue"){
685: queue_c=value->as_string().cstr();
686: } else {
687: throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
688: }
689: }
690: }
691:
692: if (!queue_c)
693: throw Exception("amqp", 0, "queue must be specified");
694:
695: amqp_queue_declare_ok_t* ok = amqp_queue_declare(self.connection(), self.channel(), amqp_cstring_bytes(queue_c), /*passive*/ 1, 0, 0, 0, amqp_empty_table);
1.16 moko 696: check(self, amqp_get_rpc_reply(self.connection()));
1.8 moko 697:
698: Value& result=*new VHash;
1.9 moko 699: if(ok){
700: result.put_element(*new String("queue"), AMQP_VSTRING(ok->queue.bytes, ok->queue.len));
701: result.put_element(*new String("messages"), new VInt(ok->message_count));
702: result.put_element(*new String("consumers"), new VInt(ok->consumer_count));
703: }
1.8 moko 704: r.write(result);
705: }
706:
707: static VHash *amqp_message_hash(amqp_envelope_t &envelope) {
708: VHash *result=new VHash;
709: HashStringValue* h=result->get_hash();
710: h->put("msg", AMQP_VSTRING(envelope.message.body.bytes, envelope.message.body.len));
711: h->put("delivery_tag", new VInt(envelope.delivery_tag));
712: h->put("consumer_tag", AMQP_VSTRING(envelope.consumer_tag.bytes, envelope.consumer_tag.len));
713: h->put("exchange", AMQP_VSTRING(envelope.exchange.bytes, envelope.exchange.len));
1.17 moko 714: h->put("routing_key", AMQP_VSTRING(envelope.routing_key.bytes, envelope.routing_key.len));
715: h->put("redelivered", &VBool::get(envelope.redelivered));
716:
717: // properties are only put when the publisher actually set them (per the wire's own _flags bitmask) -
718: // same handful of allocations as the fields above, just conditional on what's actually present
719: amqp_basic_properties_t& props=envelope.message.properties;
720: if(props._flags & AMQP_BASIC_CONTENT_TYPE_FLAG)
721: h->put("content_type", AMQP_VSTRING(props.content_type.bytes, props.content_type.len));
722: if(props._flags & AMQP_BASIC_CONTENT_ENCODING_FLAG)
723: h->put("content_encoding", AMQP_VSTRING(props.content_encoding.bytes, props.content_encoding.len));
724: if(props._flags & AMQP_BASIC_HEADERS_FLAG)
725: h->put("headers", amqp_table_to_hash(props.headers));
726: if(props._flags & AMQP_BASIC_DELIVERY_MODE_FLAG)
727: h->put("delivery_mode", new VInt(props.delivery_mode));
728: if(props._flags & AMQP_BASIC_PRIORITY_FLAG)
729: h->put("priority", new VInt(props.priority));
730: if(props._flags & AMQP_BASIC_CORRELATION_ID_FLAG)
731: h->put("correlation_id", AMQP_VSTRING(props.correlation_id.bytes, props.correlation_id.len));
732: if(props._flags & AMQP_BASIC_REPLY_TO_FLAG)
733: h->put("reply_to", AMQP_VSTRING(props.reply_to.bytes, props.reply_to.len));
734: if(props._flags & AMQP_BASIC_EXPIRATION_FLAG)
735: h->put("expiration", AMQP_VSTRING(props.expiration.bytes, props.expiration.len));
736: if(props._flags & AMQP_BASIC_MESSAGE_ID_FLAG)
737: h->put("message_id", AMQP_VSTRING(props.message_id.bytes, props.message_id.len));
738: if(props._flags & AMQP_BASIC_TIMESTAMP_FLAG)
739: h->put("timestamp", new VInt((pa_wint)props.timestamp));
740: if(props._flags & AMQP_BASIC_TYPE_FLAG)
741: h->put("type", AMQP_VSTRING(props.type.bytes, props.type.len));
742: if(props._flags & AMQP_BASIC_USER_ID_FLAG)
743: h->put("user_id", AMQP_VSTRING(props.user_id.bytes, props.user_id.len));
744: if(props._flags & AMQP_BASIC_APP_ID_FLAG)
745: h->put("app_id", AMQP_VSTRING(props.app_id.bytes, props.app_id.len));
746: if(props._flags & AMQP_BASIC_CLUSTER_ID_FLAG)
747: h->put("cluster_id", AMQP_VSTRING(props.cluster_id.bytes, props.cluster_id.len));
748:
1.8 moko 749: return result;
750: }
751:
1.14 moko 752: // timeout<0 - wait forever (NULL), timeout==0 - don't wait (poll), timeout>0 - wait up to timeout seconds
753: static struct timeval* amqp_wait_tv(struct timeval& tv, int timeout){
754: if(timeout<0)
755: return NULL;
756: tv.tv_sec=timeout;
757: tv.tv_usec=0;
758: return &tv;
759: }
760:
1.17 moko 761: // amqp_consume_message() returns AMQP_STATUS_UNEXPECTED_STATE when it read a frame other than a deliver while polling.
762: // Per its own documented contract, read that pending frame to find out what it actually was:
763: // - a basic.return for an earlier mandatory publish that found no route: benign, doesn't close anything;
764: // drain the returned message (header+body, still pending right behind it) and retry - the real
765: // message (if any) is still coming
766: // - anything else: a genuine channel/connection close (typically after a protocol violation) - build a
767: // reply so check() can classify it correctly (cheap channel repair vs full reconnect)
1.23 moko 768: static amqp_rpc_reply_t amqp_consume_message_checked(amqp_connection_state_t conn, amqp_channel_t channel, amqp_envelope_t* envelope, struct timeval* tv) {
1.17 moko 769: for(;;) {
770: amqp_rpc_reply_t res = amqp_consume_message(conn, envelope, tv, 0);
771: if(res.reply_type != AMQP_RESPONSE_LIBRARY_EXCEPTION || res.library_error != AMQP_STATUS_UNEXPECTED_STATE)
772: return res;
773:
774: amqp_frame_t frame;
775: if(amqp_simple_wait_frame(conn, &frame) != AMQP_STATUS_OK || frame.frame_type != AMQP_FRAME_METHOD)
776: return res; // give up, let the original UNEXPECTED_STATE be treated as fatal
777:
778: if(frame.payload.method.id == AMQP_BASIC_RETURN_METHOD) {
779: amqp_message_t message;
780: if(amqp_read_message(conn, channel, &message, 0).reply_type == AMQP_RESPONSE_NORMAL)
781: amqp_destroy_message(&message);
782: continue;
783: }
784:
785: res.reply_type = AMQP_RESPONSE_SERVER_EXCEPTION;
786: res.reply.id = frame.payload.method.id;
787: res.reply.decoded = frame.payload.method.decoded;
788: return res;
789: }
1.16 moko 790: }
791:
1.18 moko 792: // Auto-object used for cancelling the consumer when the enclosing scope exits. A cancel failure here must
793: // never mask whatever error is already propagating, so its result is discarded rather than check()ed.
794: class Temp_amqp_consumer {
795: public:
1.19 moko 796: Temp_amqp_consumer(VAmqp& aself, const char* aconsumer_tag, bool adrain): self(aself), consumer_tag(aconsumer_tag), drain(adrain) {}
1.18 moko 797: ~Temp_amqp_consumer() {
798: if(!consumer_tag || self.fstate != VAmqp::ALIVE) return;
799: amqp_basic_cancel(self.fconnection, self.fchannel, amqp_cstring_bytes(consumer_tag));
800: amqp_get_rpc_reply(self.fconnection);
1.19 moko 801: if(drain){
802: // The broker may have already sent a deliver frame for this (now cancelled) consumer before it even saw our cancel,
803: // so it can still be sitting in the library's own buffer. Its delivery tag belongs to the pre-reset channel and
804: // won't exist anymore afterward, so drain anything already in flight - otherwise it would surface later, misattributed.
805: amqp_envelope_t envelope;
806: struct timeval no_wait = {0,0};
807: while(amqp_consume_message(self.fconnection, &envelope, &no_wait, 0).reply_type == AMQP_RESPONSE_NORMAL)
808: amqp_destroy_envelope(&envelope);
809: amqp_reset_channel(self);
810: }
1.18 moko 811: }
812: private:
813: VAmqp& self;
814: const char* consumer_tag;
1.19 moko 815: bool drain;
1.18 moko 816: };
817:
1.1 moko 818: static void _consume(Request& r, MethodParams& params) {
819: VAmqp& self=GET_SELF(r, VAmqp);
820: const char* queue_c=0;
821: const char* consumer_tag_c=0;
1.17 moko 822: bool ack=true, exclusive=false;
1.14 moko 823: int count=1; bool count_specified=false;
824: int timeout=0; bool timeout_specified=false;
1.1 moko 825: Junction* callback=0;
826:
1.2 moko 827: if(HashStringValue* options=params.as_hash(0)){
828: for(HashStringValue::Iterator i(*options); i; i.next()){
829: String::Body key=i.key();
830: Value* value=i.value();
1.3 moko 831: if(key=="callback"){
832: callback=value->get_junction();
833: } else if(key=="queue"){
1.2 moko 834: queue_c=value->as_string().cstr();
835: } else if(key=="consumer_tag"){
836: consumer_tag_c=value->as_string().cstr();
1.17 moko 837: } else if(key=="ack"){
838: ack=r.process(*value).as_bool();
1.10 moko 839: } else if(key=="exclusive"){
840: exclusive=r.process(*value).as_bool();
1.8 moko 841: } else if(key=="count"){
842: count=r.process(*value).as_int();
1.14 moko 843: count_specified=true;
844: } else if(key=="timeout"){
845: timeout=r.process(*value).as_int();
846: timeout_specified=true;
1.2 moko 847: } else
848: throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION);
1.1 moko 849: }
850: }
851:
1.8 moko 852: if(!queue_c) throw Exception("amqp", 0, "queue must be specified");
1.14 moko 853: if(count==0) throw Exception("amqp", 0, "count must not be zero");
854:
855: // listen mode (callback): count/timeout unlimited by default - runs until stop_consume/timeout/error
856: // pull mode (no callback): count(1)/timeout(0) by default - returns whatever is ready right now
857: if(!count_specified)
858: count = callback ? -1 : 1;
859: if(!timeout_specified)
860: timeout = callback ? -1 : 0;
1.1 moko 861:
1.9 moko 862: amqp_basic_consume_ok_t *ok = amqp_basic_consume(self.connection(), self.channel(), amqp_cstring_bytes(queue_c),
1.1 moko 863: consumer_tag_c ? amqp_cstring_bytes(consumer_tag_c) : amqp_empty_bytes,
1.17 moko 864: 0 /*no_local*/, !ack, exclusive, amqp_empty_table);
1.16 moko 865: check(self, amqp_get_rpc_reply(self.connection()));
1.1 moko 866:
1.15 moko 867: // amqp_maybe_release_buffers() recycles ok->consumer_tag
868: const char* consumer_tag_copy = ok ? pa_strdup((const char*)ok->consumer_tag.bytes, ok->consumer_tag.len) : 0;
1.19 moko 869: Temp_amqp_consumer cancel_consumer(self, consumer_tag_copy, ack && callback);
1.15 moko 870:
1.14 moko 871: struct timeval tv;
872:
1.8 moko 873: if(callback){
874: self.fstop=false;
1.14 moko 875: for(int received=0; !self.fstop && (count<0 || received<count); received++){
1.8 moko 876: amqp_envelope_t envelope;
877: memset(&envelope, 0, sizeof(envelope));
878: amqp_maybe_release_buffers(self.connection());
1.14 moko 879: // timeout is the max idle gap between messages, re-armed on every wait
1.17 moko 880: amqp_rpc_reply_t res = amqp_consume_message_checked(self.connection(), self.channel(), &envelope, amqp_wait_tv(tv, timeout));
1.8 moko 881: if(res.reply_type == AMQP_RESPONSE_NORMAL){
882: VHash *vh=amqp_message_hash(envelope);
883: Value *params_cb[]={vh};
884: METHOD_FRAME_ACTION(*callback->method, r.method_frame, callback->self, {
885: frame.store_params(params_cb, 1);
886: r.call(frame);
887: });
888: amqp_destroy_envelope(&envelope);
889: } else if(res.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION && res.library_error == AMQP_STATUS_TIMEOUT) {
1.14 moko 890: break; // idle too long - stop cleanly
1.8 moko 891: } else {
1.17 moko 892: check(self, res, "consume ");
1.8 moko 893: }
894: }
895: } else {
896: VArray& result=*new VArray();
897: ArrayValue& result_array=result.array();
898:
1.14 moko 899: for(int received=0; count<0 || received<count; received++){
1.8 moko 900: amqp_envelope_t envelope;
901: memset(&envelope, 0, sizeof(envelope));
902: amqp_maybe_release_buffers(self.connection());
1.14 moko 903: // timeout applies only to the first message of the batch;
904: // the rest is drained with a non-blocking poll until nothing more is ready
1.17 moko 905: amqp_rpc_reply_t res = amqp_consume_message_checked(self.connection(), self.channel(), &envelope, amqp_wait_tv(tv, received==0 ? timeout : 0));
1.8 moko 906: if(res.reply_type == AMQP_RESPONSE_NORMAL){
907: result_array+=amqp_message_hash(envelope);
908: amqp_destroy_envelope(&envelope);
1.14 moko 909: } else if(res.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION && res.library_error == AMQP_STATUS_TIMEOUT) {
910: break; // nothing (more) ready - return what we have
1.8 moko 911: } else {
1.17 moko 912: check(self, res, "consume ");
1.8 moko 913: }
1.1 moko 914: }
1.8 moko 915: r.write(result);
1.1 moko 916: }
917: }
918:
919: static void _stop_consume(Request& r, MethodParams&) {
920: VAmqp& self=GET_SELF(r, VAmqp);
921: self.fstop=true;
922: }
923:
1.6 moko 924: #endif // WITH_AMQP
1.1 moko 925:
926: // constructor
927: MAmqp::MAmqp(): Methoded("amqp") {
928: add_native_method("create", Method::CT_DYNAMIC, _create, 0, 1);
929: #ifdef WITH_AMQP
930: add_native_method("publish", Method::CT_DYNAMIC, _publish, 1, 2);
931: add_native_method("release", Method::CT_DYNAMIC, _release, 0, 0);
932: add_native_method("ack", Method::CT_DYNAMIC, _ack, 1, 1);
933: add_native_method("nack", Method::CT_DYNAMIC, _nack, 1, 2);
934: add_native_method("reject", Method::CT_DYNAMIC, _reject, 1, 2);
935: add_native_method("qos", Method::CT_DYNAMIC, _qos, 0, 1);
1.8 moko 936: add_native_method("declare", Method::CT_DYNAMIC, _declare, 1, 1);
937: add_native_method("delete", Method::CT_DYNAMIC, _delete, 1, 1);
938: add_native_method("bind", Method::CT_DYNAMIC, _bind, 1, 1);
939: add_native_method("unbind", Method::CT_DYNAMIC, _unbind, 1, 1);
940: add_native_method("purge", Method::CT_DYNAMIC, _purge, 1, 1);
941: add_native_method("info", Method::CT_DYNAMIC, _info, 1, 1);
1.2 moko 942: add_native_method("consume", Method::CT_DYNAMIC, _consume, 1, 1);
1.1 moko 943: add_native_method("stop_consume", Method::CT_DYNAMIC, _stop_consume, 0, 0);
1.6 moko 944: #endif // WITH_AMQP
1.1 moko 945: }
E-mail:
![[BACK]](/cvsweb/icons/back.gif){kind=icon}
Return to amqp.C CVS log ![[TXT]](/cvsweb/icons/text.gif){kind=icon}
![[DIR]](/cvsweb/icons/dir.gif){kind=icon}
Up to [parser3project] / parser3 / src / classes