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