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