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