--- parser3/src/classes/amqp.C 2026/09/15 17:53:05 1.16 +++ parser3/src/classes/amqp.C 2026/10/09 18:32:15 1.27 @@ -5,27 +5,17 @@ Authors: Konstantin Morshnev */ + +#include "classes.h" #include "pa_vmethod_frame.h" +#include "pa_vamqp.h" #include "pa_request.h" -#include "pa_vstring.h" -#include "pa_vhash.h" #include "pa_varray.h" #include "pa_vbool.h" -#include "pa_vamqp.h" #include "pa_os.h" -#include "pa_globals.h" -#ifdef WITH_AMQP -#include -#include -#include -#include -#include -#include -#endif - -volatile const char * IDENT_AMQP_C="$Id: amqp.C,v 1.16 2026/09/15 17:53:05 moko Exp $" IDENT_PA_VAMQP_H; +volatile const char * IDENT_AMQP_C="$Id: amqp.C,v 1.27 2026/10/09 18:32:15 moko Exp $" IDENT_PA_VAMQP_H; class MAmqp: public Methoded { public: // VStateless_class @@ -93,13 +83,15 @@ static void amqp_connect(VAmqp& self, Re const char* pass_c = "guest"; const char* vhost_c = "/"; const char* locale_c = "en_US"; - int heartbeat = 30; // seconds + int heartbeat_sec = 30; + int reconnect_interval_sec = 0; + bool tls_specified = false; +#ifdef WITH_AMQP_SSL const char* tls_ca = 0; const char* tls_cert = 0; const char* tls_key = 0; - bool tls_specified = false; bool tls_verify = true; - int reconnect_interval_sec = 0; +#endif if(options){ for(HashStringValue::Iterator i(*options); i; i.next()){ @@ -118,10 +110,11 @@ static void amqp_connect(VAmqp& self, Re } else if(key=="locale"){ locale_c=value->as_string().cstr(); } else if(key=="heartbeat"){ - heartbeat=r.process(*value).as_int(); + heartbeat_sec=r.process(*value).as_int(); } else if(key=="auto_reconnect"){ reconnect_interval_sec=r.process(*value).as_int(); } else if(key=="tls"){ +#ifdef WITH_AMQP_SSL tls_specified = true; if(HashStringValue* tls_options=value->get_hash()){ for(HashStringValue::Iterator t(*tls_options); t; t.next()){ @@ -139,6 +132,9 @@ static void amqp_connect(VAmqp& self, Re throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION); } } +#else + throw Exception("amqp", 0, "AMQP SSL support is disabled"); +#endif } else throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION); } @@ -147,6 +143,7 @@ static void amqp_connect(VAmqp& self, Re amqp_connection_state_t conn = amqp_new_connection(); amqp_socket_t* socket = 0; +#ifdef WITH_AMQP_SSL if(tls_specified) { socket = amqp_ssl_socket_new(conn); if(!socket) @@ -169,7 +166,9 @@ static void amqp_connect(VAmqp& self, Re amqp_ssl_socket_set_verify_peer(socket, tls_verify && tls_ca); // If verify=true, enable hostname verification amqp_ssl_socket_set_verify_hostname(socket, tls_verify); - } else { + } else +#endif + { socket = amqp_tcp_socket_new(conn); if(!socket) throw Exception("amqp", 0, "failed to create TCP socket"); @@ -177,13 +176,13 @@ static void amqp_connect(VAmqp& self, Re status_check(self, amqp_socket_open(socket, host_c, port), tls_specified ? "open SSL socket " : "open TCP socket "); - amqp_rpc_reply_t rlogin = amqp_login(conn, vhost_c, 0, 131072, heartbeat, AMQP_SASL_METHOD_PLAIN, user_c, pass_c); + amqp_rpc_reply_t rlogin = amqp_login(conn, vhost_c, 0, 131072, heartbeat_sec, AMQP_SASL_METHOD_PLAIN, user_c, pass_c); if(rlogin.reply_type != AMQP_RESPONSE_NORMAL){ amqp_destroy_connection(conn); check(self, rlogin, "login "); } - int channel = 1; + amqp_channel_t channel = 1; amqp_channel_open(conn, channel); amqp_rpc_reply_t ropen = amqp_get_rpc_reply(conn); if(ropen.reply_type != AMQP_RESPONSE_NORMAL){ @@ -201,7 +200,9 @@ static void amqp_connect(VAmqp& self, Re self.freconnect_interval = reconnect_interval_sec; } -// repairs a channel-level error cheaply: reopen on the same connection +// repairs a channel-level error cheaply: reopen on the same connection. Only valid when the server has +// already sent (or is about to send) channel.close - we're completing that handshake with close-ok, not +// initiating our own. static void amqp_repair_channel(VAmqp& self) { amqp_channel_close_ok_t close_ok; amqp_send_method(self.fconnection, self.fchannel, AMQP_CHANNEL_CLOSE_OK_METHOD, &close_ok); @@ -210,6 +211,13 @@ static void amqp_repair_channel(VAmqp& s self.fstate = VAmqp::ALIVE; } +// voluntarily closes and reopens an otherwise-healthy channel +static void amqp_reset_channel(VAmqp& self) { + amqp_channel_close(self.fconnection, self.fchannel, AMQP_REPLY_SUCCESS); + amqp_channel_open(self.fconnection, self.fchannel); + amqp_get_rpc_reply(self.fconnection); +} + void VAmqp::ensure_connected() { if(fstate == ALIVE) return; @@ -247,6 +255,97 @@ static void _create(Request& r, MethodPa #define AMQP_STRING(s,l) new String(String::C(pa_strdup((const char*)(s), (l)), (l))) #define AMQP_VSTRING(s,l) new VString(*AMQP_STRING(s,l)) +// broker-defined arguments/headers (x-message-ttl, x-dead-letter-exchange, ...) +// Type is taken from how the value was assigned: [] -> string, () -> number, (true|false) -> bool +static amqp_table_t amqp_build_arguments_table(Request& r, HashStringValue* arguments) { + amqp_table_t table = amqp_empty_table; + size_t count = arguments ? arguments->count() : 0; + if(!count) + return table; + + amqp_table_entry_t* entries = (amqp_table_entry_t*)pa_malloc(count * sizeof(amqp_table_entry_t)); + size_t i = 0; + for(HashStringValue::Iterator it(*arguments); it; it.next()){ + entries[i].key = amqp_cstring_bytes(it.key().cstr()); + + Value& value = r.process(*it.value()); + if(value.is_string() || value.is_void()){ + entries[i].value.kind = AMQP_FIELD_KIND_UTF8; + entries[i].value.value.bytes = amqp_cstring_bytes(value.as_string().cstr()); + } else if(value.is_bool()){ + entries[i].value.kind = AMQP_FIELD_KIND_BOOLEAN; + entries[i].value.value.boolean = value.as_bool(); + } else { + double d = value.as_double(); + double rounded = round(d); + if(ulp_eq_double(d, rounded)){ + entries[i].value.kind = AMQP_FIELD_KIND_I64; + entries[i].value.value.i64 = (int64_t)rounded; + } else { + entries[i].value.kind = AMQP_FIELD_KIND_F64; + entries[i].value.value.f64 = d; + } + } + i++; + } + + table.num_entries = (int)count; + table.entries = entries; + return table; +} + +// reverse of amqp_build_arguments_table - decodes a broker-provided field table (e.g. message headers) back into a hash. +static VHash* amqp_table_to_hash(amqp_table_t& table); + +static Value* amqp_field_to_value(amqp_field_value_t& f) { + switch(f.kind){ + case AMQP_FIELD_KIND_UTF8: + case AMQP_FIELD_KIND_BYTES: + return AMQP_VSTRING(f.value.bytes.bytes, f.value.bytes.len); + case AMQP_FIELD_KIND_BOOLEAN: + return &VBool::get(f.value.boolean); + case AMQP_FIELD_KIND_I8: return new VInt((pa_wint)f.value.i8); + case AMQP_FIELD_KIND_U8: return new VInt((pa_wint)f.value.u8); + case AMQP_FIELD_KIND_I16: return new VInt((pa_wint)f.value.i16); + case AMQP_FIELD_KIND_U16: return new VInt((pa_wint)f.value.u16); + case AMQP_FIELD_KIND_I32: return new VInt((pa_wint)f.value.i32); + case AMQP_FIELD_KIND_U32: return new VInt((pa_wint)f.value.u32); + case AMQP_FIELD_KIND_I64: return new VInt((pa_wint)f.value.i64); + case AMQP_FIELD_KIND_U64: return new VInt((pa_wint)f.value.u64); + case AMQP_FIELD_KIND_TIMESTAMP: return new VInt((pa_wint)f.value.u64); + case AMQP_FIELD_KIND_F32: return new VDouble((double)f.value.f32); + case AMQP_FIELD_KIND_F64: return new VDouble(f.value.f64); + case AMQP_FIELD_KIND_DECIMAL: { + double divisor=1; + for(uint8_t i=0; iarray(); + for(int i=0; iget_hash(); + for(int i=0; iput(*AMQP_STRING(e.key.bytes, e.key.len), amqp_field_to_value(e.value)); + } + return result; +} + static void _publish(Request& r, MethodParams& params) { VAmqp& self=GET_SELF(r, VAmqp); const String &msg=params.as_string(0, "msg must be string"); @@ -318,24 +417,14 @@ static void _publish(Request& r, MethodP const char* v=value->as_string().cstr(); props.app_id=amqp_cstring_bytes(v); props._flags|=AMQP_BASIC_APP_ID_FLAG; + } else if(key=="cluster_id"){ + const char* v=value->as_string().cstr(); + props.cluster_id=amqp_cstring_bytes(v); + props._flags|=AMQP_BASIC_CLUSTER_ID_FLAG; } else if(key=="headers"){ -/* if(HashStringValue* hh=pval->get_hash()){ - size_t count=hh->count(); - amqp_table_entry_t* entries=count ? new amqp_table_entry_t[count] : 0; - size_t idx=0; - for(HashStringValue::Iterator hi(*hh); hi; hi.next()){ - String::Body hkey=hi.key(); - const char* hv=hi.value()->as_string().cstr(); - entries[idx].key=amqp_cstring_bytes(hkey.cstr()); - entries[idx].value.kind=AMQP_FIELD_KIND_UTF8; - entries[idx].value.value.bytes=amqp_cstring_bytes(hv); - idx++; - } - props.headers.num_entries=(int)count; - props.headers.entries=entries; - props._flags|=AMQP_BASIC_HEADERS_FLAG; - } -*/ } else + props.headers=amqp_build_arguments_table(r, value->get_hash()); + props._flags|=AMQP_BASIC_HEADERS_FLAG; + } else throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION); } } @@ -351,11 +440,6 @@ static void _publish(Request& r, MethodP int ret = amqp_basic_publish(self.connection(), self.channel(), amqp_cstring_bytes(exchange_c), amqp_cstring_bytes(routing_key_c), mandatory, 0, &props, body); status_check(self, ret, "publish "); - - // free temporary headers entries if allocated - if(props._flags & AMQP_BASIC_HEADERS_FLAG){ -// delete [] props.headers.entries; - } } static void _release(Request& r, MethodParams&) { @@ -443,6 +527,7 @@ static void _declare(Request& r, MethodP const char* queue_c = 0; const char* type_c = "direct"; bool passive=false, durable=false, auto_delete=false, internal=false, exclusive=false; + amqp_table_t arguments=amqp_empty_table; if(HashStringValue* options=params.as_hash(0)){ for(HashStringValue::Iterator i(*options); i; i.next()){ String::Body key=i.key(); @@ -463,6 +548,8 @@ static void _declare(Request& r, MethodP internal=r.process(*value).as_bool(); } else if(key=="exclusive"){ exclusive=r.process(*value).as_bool(); + } else if(key=="arguments"){ + arguments=amqp_build_arguments_table(r, value->get_hash()); } else throw Exception(PARSER_RUNTIME, 0, CALLED_WITH_INVALID_OPTION); } @@ -471,12 +558,12 @@ static void _declare(Request& r, MethodP throw Exception("amqp", 0, "exchange or queue must be specified"); if(exchange_c){ - amqp_exchange_declare(self.connection(), self.channel(), amqp_cstring_bytes(exchange_c), amqp_cstring_bytes(type_c), passive, durable, auto_delete, internal, amqp_empty_table); + amqp_exchange_declare(self.connection(), self.channel(), amqp_cstring_bytes(exchange_c), amqp_cstring_bytes(type_c), passive, durable, auto_delete, internal, arguments); check(self, amqp_get_rpc_reply(self.connection())); } if(queue_c){ - amqp_queue_declare_ok_t *ok = amqp_queue_declare(self.connection(), self.channel(), *queue_c ? amqp_cstring_bytes(queue_c) : amqp_empty_bytes, passive, durable, exclusive, auto_delete, amqp_empty_table); + 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); check(self, amqp_get_rpc_reply(self.connection())); if(!*queue_c && ok){ r.write(*AMQP_STRING(ok->queue.bytes, ok->queue.len)); @@ -626,6 +713,41 @@ static VHash *amqp_message_hash(amqp_env h->put("delivery_tag", new VInt(envelope.delivery_tag)); h->put("consumer_tag", AMQP_VSTRING(envelope.consumer_tag.bytes, envelope.consumer_tag.len)); h->put("exchange", AMQP_VSTRING(envelope.exchange.bytes, envelope.exchange.len)); + h->put("routing_key", AMQP_VSTRING(envelope.routing_key.bytes, envelope.routing_key.len)); + h->put("redelivered", &VBool::get(envelope.redelivered)); + + // properties are only put when the publisher actually set them (per the wire's own _flags bitmask) - + // same handful of allocations as the fields above, just conditional on what's actually present + amqp_basic_properties_t& props=envelope.message.properties; + if(props._flags & AMQP_BASIC_CONTENT_TYPE_FLAG) + h->put("content_type", AMQP_VSTRING(props.content_type.bytes, props.content_type.len)); + if(props._flags & AMQP_BASIC_CONTENT_ENCODING_FLAG) + h->put("content_encoding", AMQP_VSTRING(props.content_encoding.bytes, props.content_encoding.len)); + if(props._flags & AMQP_BASIC_HEADERS_FLAG) + h->put("headers", amqp_table_to_hash(props.headers)); + if(props._flags & AMQP_BASIC_DELIVERY_MODE_FLAG) + h->put("delivery_mode", new VInt(props.delivery_mode)); + if(props._flags & AMQP_BASIC_PRIORITY_FLAG) + h->put("priority", new VInt(props.priority)); + if(props._flags & AMQP_BASIC_CORRELATION_ID_FLAG) + h->put("correlation_id", AMQP_VSTRING(props.correlation_id.bytes, props.correlation_id.len)); + if(props._flags & AMQP_BASIC_REPLY_TO_FLAG) + h->put("reply_to", AMQP_VSTRING(props.reply_to.bytes, props.reply_to.len)); + if(props._flags & AMQP_BASIC_EXPIRATION_FLAG) + h->put("expiration", AMQP_VSTRING(props.expiration.bytes, props.expiration.len)); + if(props._flags & AMQP_BASIC_MESSAGE_ID_FLAG) + h->put("message_id", AMQP_VSTRING(props.message_id.bytes, props.message_id.len)); + if(props._flags & AMQP_BASIC_TIMESTAMP_FLAG) + h->put("timestamp", new VInt((pa_wint)props.timestamp)); + if(props._flags & AMQP_BASIC_TYPE_FLAG) + h->put("type", AMQP_VSTRING(props.type.bytes, props.type.len)); + if(props._flags & AMQP_BASIC_USER_ID_FLAG) + h->put("user_id", AMQP_VSTRING(props.user_id.bytes, props.user_id.len)); + if(props._flags & AMQP_BASIC_APP_ID_FLAG) + h->put("app_id", AMQP_VSTRING(props.app_id.bytes, props.app_id.len)); + if(props._flags & AMQP_BASIC_CLUSTER_ID_FLAG) + h->put("cluster_id", AMQP_VSTRING(props.cluster_id.bytes, props.cluster_id.len)); + return result; } @@ -638,39 +760,68 @@ static struct timeval* amqp_wait_tv(stru return &tv; } -// amqp_consume_message() returns AMQP_STATUS_UNEXPECTED_STATE when it read a frame other than a deliver -// while polling - typically the broker notifying a channel/connection close after a protocol violation -// Per its own documented contract, read that pending frame to find out which, so check() can classify -// it correctly (cheap channel repair vs full reconnect) instead of always assuming the worse. -static amqp_rpc_reply_t amqp_resolve_unexpected_state(amqp_connection_state_t conn) { - amqp_rpc_reply_t result; - memset(&result, 0, sizeof(result)); - - amqp_frame_t frame; - if(amqp_simple_wait_frame(conn, &frame) != AMQP_STATUS_OK || frame.frame_type != AMQP_FRAME_METHOD) { - result.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION; - result.library_error = AMQP_STATUS_UNEXPECTED_STATE; - return result; - } - - result.reply_type = AMQP_RESPONSE_SERVER_EXCEPTION; - result.reply.id = frame.payload.method.id; - result.reply.decoded = frame.payload.method.decoded; - return result; +// amqp_consume_message() returns AMQP_STATUS_UNEXPECTED_STATE when it read a frame other than a deliver while polling. +// Per its own documented contract, read that pending frame to find out what it actually was: +// - a basic.return for an earlier mandatory publish that found no route: benign, doesn't close anything; +// drain the returned message (header+body, still pending right behind it) and retry - the real +// message (if any) is still coming +// - anything else: a genuine channel/connection close (typically after a protocol violation) - build a +// reply so check() can classify it correctly (cheap channel repair vs full reconnect) +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) { + for(;;) { + amqp_rpc_reply_t res = amqp_consume_message(conn, envelope, tv, 0); + if(res.reply_type != AMQP_RESPONSE_LIBRARY_EXCEPTION || res.library_error != AMQP_STATUS_UNEXPECTED_STATE) + return res; + + amqp_frame_t frame; + if(amqp_simple_wait_frame(conn, &frame) != AMQP_STATUS_OK || frame.frame_type != AMQP_FRAME_METHOD) + return res; // give up, let the original UNEXPECTED_STATE be treated as fatal + + if(frame.payload.method.id == AMQP_BASIC_RETURN_METHOD) { + amqp_message_t message; + if(amqp_read_message(conn, channel, &message, 0).reply_type == AMQP_RESPONSE_NORMAL) + amqp_destroy_message(&message); + continue; + } + + res.reply_type = AMQP_RESPONSE_SERVER_EXCEPTION; + res.reply.id = frame.payload.method.id; + res.reply.decoded = frame.payload.method.decoded; + return res; + } } -static void amqp_check_consume_result(VAmqp& self, amqp_rpc_reply_t& res) { - if(res.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION && res.library_error == AMQP_STATUS_UNEXPECTED_STATE) - check(self, amqp_resolve_unexpected_state(self.connection()), "consume "); // read the real frame, classify properly - else - check(self, res, "consume "); // real error - throw -} +// Auto-object used for cancelling the consumer when the enclosing scope exits. A cancel failure here must +// never mask whatever error is already propagating, so its result is discarded rather than check()ed. +class Temp_amqp_consumer { +public: + Temp_amqp_consumer(VAmqp& aself, const char* aconsumer_tag, bool adrain): self(aself), consumer_tag(aconsumer_tag), drain(adrain) {} + ~Temp_amqp_consumer() { + if(!consumer_tag || self.fstate != VAmqp::ALIVE) return; + amqp_basic_cancel(self.fconnection, self.fchannel, amqp_cstring_bytes(consumer_tag)); + amqp_get_rpc_reply(self.fconnection); + if(drain){ + // The broker may have already sent a deliver frame for this (now cancelled) consumer before it even saw our cancel, + // so it can still be sitting in the library's own buffer. Its delivery tag belongs to the pre-reset channel and + // won't exist anymore afterward, so drain anything already in flight - otherwise it would surface later, misattributed. + amqp_envelope_t envelope; + struct timeval no_wait = {0,0}; + while(amqp_consume_message(self.fconnection, &envelope, &no_wait, 0).reply_type == AMQP_RESPONSE_NORMAL) + amqp_destroy_envelope(&envelope); + amqp_reset_channel(self); + } + } +private: + VAmqp& self; + const char* consumer_tag; + bool drain; +}; static void _consume(Request& r, MethodParams& params) { VAmqp& self=GET_SELF(r, VAmqp); const char* queue_c=0; const char* consumer_tag_c=0; - bool no_ack=false, exclusive=false; + bool ack=true, exclusive=false; int count=1; bool count_specified=false; int timeout=0; bool timeout_specified=false; Junction* callback=0; @@ -685,8 +836,8 @@ static void _consume(Request& r, MethodP queue_c=value->as_string().cstr(); } else if(key=="consumer_tag"){ consumer_tag_c=value->as_string().cstr(); - } else if(key=="no_ack"){ - no_ack=r.process(*value).as_bool(); + } else if(key=="ack"){ + ack=r.process(*value).as_bool(); } else if(key=="exclusive"){ exclusive=r.process(*value).as_bool(); } else if(key=="count"){ @@ -712,11 +863,12 @@ static void _consume(Request& r, MethodP amqp_basic_consume_ok_t *ok = amqp_basic_consume(self.connection(), self.channel(), amqp_cstring_bytes(queue_c), consumer_tag_c ? amqp_cstring_bytes(consumer_tag_c) : amqp_empty_bytes, - 0 /*no_local*/, no_ack, exclusive, amqp_empty_table); + 0 /*no_local*/, !ack, exclusive, amqp_empty_table); check(self, amqp_get_rpc_reply(self.connection())); // amqp_maybe_release_buffers() recycles ok->consumer_tag const char* consumer_tag_copy = ok ? pa_strdup((const char*)ok->consumer_tag.bytes, ok->consumer_tag.len) : 0; + Temp_amqp_consumer cancel_consumer(self, consumer_tag_copy, ack && callback); struct timeval tv; @@ -727,7 +879,7 @@ static void _consume(Request& r, MethodP memset(&envelope, 0, sizeof(envelope)); amqp_maybe_release_buffers(self.connection()); // timeout is the max idle gap between messages, re-armed on every wait - amqp_rpc_reply_t res = amqp_consume_message(self.connection(), &envelope, amqp_wait_tv(tv, timeout), 0); + amqp_rpc_reply_t res = amqp_consume_message_checked(self.connection(), self.channel(), &envelope, amqp_wait_tv(tv, timeout)); if(res.reply_type == AMQP_RESPONSE_NORMAL){ VHash *vh=amqp_message_hash(envelope); Value *params_cb[]={vh}; @@ -739,7 +891,7 @@ static void _consume(Request& r, MethodP } else if(res.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION && res.library_error == AMQP_STATUS_TIMEOUT) { break; // idle too long - stop cleanly } else { - amqp_check_consume_result(self, res); + check(self, res, "consume "); } } } else { @@ -752,23 +904,18 @@ static void _consume(Request& r, MethodP amqp_maybe_release_buffers(self.connection()); // timeout applies only to the first message of the batch; // the rest is drained with a non-blocking poll until nothing more is ready - amqp_rpc_reply_t res = amqp_consume_message(self.connection(), &envelope, amqp_wait_tv(tv, received==0 ? timeout : 0), 0); + amqp_rpc_reply_t res = amqp_consume_message_checked(self.connection(), self.channel(), &envelope, amqp_wait_tv(tv, received==0 ? timeout : 0)); if(res.reply_type == AMQP_RESPONSE_NORMAL){ result_array+=amqp_message_hash(envelope); amqp_destroy_envelope(&envelope); } else if(res.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION && res.library_error == AMQP_STATUS_TIMEOUT) { break; // nothing (more) ready - return what we have } else { - amqp_check_consume_result(self, res); + check(self, res, "consume "); } } r.write(result); } - - if(ok){ - amqp_basic_cancel(self.connection(), self.channel(), amqp_cstring_bytes(consumer_tag_copy)); - check(self, amqp_get_rpc_reply(self.connection())); - } } static void _stop_consume(Request& r, MethodParams&) {