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