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